diff --git a/packages/ai/README.md b/packages/ai/README.md index 00a44beea1e..bb1a926dc22 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -249,6 +249,107 @@ The published legacy `Service`, `layer`, `clientLayer`, and module-level control over the same implementation, including the legacy live `requests` array. New tests should use `Test` and `testLayer`. +## Provider compaction + +Compaction is opt-in. The package supports automatic compaction in OpenAI/Azure Responses and Anthropic Messages (including Claude on Vertex and Bedrock Messages), and explicit compaction calls in OpenAI/Azure/xAI Responses. Model and deployment support still depends on the provider. + +This is different from prompt caching, server-side history storage, or truncation. Compaction returns provider-owned context that must be replayed to continue the conversation. + +### Automatic compaction + +Inside an `Effect.gen`, enable OpenAI compaction with typed provider options: + +```ts +import { LLM, LLMClient, LLMRequest, Message } from "@opencode-ai/ai" +import { OpenAI } from "@opencode-ai/ai/providers" + +const request = LLM.request({ + model: OpenAI.configure({ apiKey }).responses("gpt-5.3-codex"), + messages, + providerOptions: { + contextManagement: [{ type: "compaction", compactThreshold: 200_000 }], + }, +}) +const response = yield * LLMClient.generate(request) +const next = LLMRequest.update(request, { + messages: [...request.messages, response.message, Message.user("Continue")], +}) +``` + +`store: false` remains the default. Keep the entire `response.message`, not just `response.text`. Compaction events become ordered `CompactionPart`s alongside text and reasoning. The conversation contains everything needed to continue; there is no separate replay object or hidden provider transcript. + +A compaction part has `provider` and exactly one representation: `encrypted` for Responses, or `text` for Anthropic. Responses also preserves the optional checkpoint `id`. These fields survive message serialization without becoming visible assistant text. Sending a checkpoint to another provider or an incompatible API fails rather than silently losing context. + +```ts +import { CompactionPart, ProviderID } from "@opencode-ai/ai" + +CompactionPart.make({ provider: ProviderID.make("openai"), id: "cmp_123", encrypted: "..." }) +CompactionPart.make({ provider: ProviderID.make("anthropic"), text: "Summary of the conversation..." }) +``` + +For Anthropic, use: + +```ts +providerOptions: { + contextManagement: { + edits: [{ + type: "compact_20260112", + trigger: { type: "input_tokens", value: 150_000 }, + pauseAfterCompaction: true, + instructions: "Summarize the task and decisions. Do not call tools while summarizing.", + }], + }, +} +``` + +- The trigger is optional (provider default: 150,000 tokens), with a minimum of 50,000. +- Custom instructions replace Anthropic's default summarization instructions. +- The route adds `compact-2026-01-12` to existing beta headers, including when replaying a checkpoint without enabling new compactions. +- A pause is exposed as `response.finishReason.raw === "compaction"`. The caller explicitly issues the next request; the package never automatically resumes. +- Anthropic can return a compaction block with `content: null` when summarization fails. This becomes a compaction part with `text: null`, which is **not** a successful replacement for prior history. The package never prunes history automatically. +- `Usage` totals include all reported Anthropic `usage.iterations`, including compaction. `contextTokens` separately reports the final message iteration's inclusive input size, when available. A compaction-only pause does not report a post-compaction context size. Raw iteration usage remains in `providerMetadata`. + +Bedrock's Converse API does not support this feature. Select the native Claude Messages route explicitly; the default `.model(...)` remains Converse: + +```ts +import { AmazonBedrock } from "@opencode-ai/ai/providers" + +const model = AmazonBedrock.configure({ region: "us-east-1", credentials }).messages("us.anthropic.claude-opus-4-6-v1") +``` + +The corresponding package entrypoint is `@opencode-ai/ai/providers/amazon-bedrock/messages`. It uses InvokeModelWithResponseStream, AWS event-stream framing, bearer or SigV4 auth, and `anthropic_beta` in the request body. + +### Explicit compaction + +`LLMClient.compact(request)` performs exactly one HTTP call to `/responses/compact`, using the selected route's endpoint, credentials, query, and HTTP middleware. It returns a `CompactionResponse` containing replacement `messages` and usage, not a normal generation response. + +```ts +const compacted = yield * LLMClient.compact(request) +const next = LLMRequest.update(request, { + messages: [...compacted.messages, Message.user("Continue")], +}) +const response = yield * LLMClient.generate(next) +``` + +Replace the prior window with `compacted.messages`. Do not append it to the original transcript or extract only the encrypted item: the provider may retain additional messages in its output. Retained user and assistant messages remain ordinary messages with typed text, media, or reasoning parts, in their original order. Provider-specific message IDs, status, and phase use `providerMetadata`, not a raw output array hidden in an assistant message. Unsupported returned item types fail explicitly. Generation-only body overlays such as `stream` and `store` are not sent to the compact endpoint. + +The input must still fit the model's context window. Explicit compaction is not an overflow-recovery operation. xAI supports this explicit path, not the automatic OpenAI option. Unsupported routes, including Bedrock Mantle, do not inherit an explicit compact endpoint simply because they use a Responses protocol. + +### Ownership and verification + +The AI package transports options and typed conversation parts. It does not schedule compaction, persist Session checkpoints, select history, switch providers, or replace Core's existing local compaction policy. Native compaction is not enabled for OpenCode Sessions by this feature; Session integration must persist these parts before enabling it. The AI SDK bridge rejects native compaction parts rather than dropping them. Provider-executed tool APIs and persistence changes are a separate follow-up. + +Tests cover serialized round trips, real local HTTP plus a tool loop, AWS binary frames and signing, provider errors, malformed blocks, and usage accounting. Live provider tests are gated by `RECORD=true` and the relevant API keys: + +```sh +# Run from packages/ai. Only records the selected new cassette group. +RECORD=true RECORDED_PREFIX=openai-compaction bun test test/provider/compaction.recorded.test.ts +RECORD=true RECORDED_PREFIX=xai-compaction bun test test/provider/compaction.recorded.test.ts +RECORD=true RECORDED_PREFIX=anthropic-compaction bun test test/provider/compaction.recorded.test.ts +``` + +Provider references: [OpenAI](https://developers.openai.com/api/docs/guides/compaction), [Azure](https://learn.microsoft.com/en-us/azure/foundry/openai/how-to/responses#server-side-compaction), [Anthropic](https://platform.claude.com/docs/en/build-with-claude/compaction), [Bedrock](https://docs.aws.amazon.com/bedrock/latest/userguide/claude-messages-compaction.html), [xAI](https://docs.x.ai/developers/advanced-api-usage/context-compaction). + ## Caching Prompt caching is **on by default**. Every `LLMRequest` resolves to `cache: "auto"` unless the caller opts out with `cache: "none"`. Each protocol translates `CacheHint`s to its wire format (`cache_control` on Anthropic, `cachePoint` on Bedrock; OpenAI and Gemini do implicit caching server-side and don't need inline markers — auto is a no-op there). diff --git a/packages/ai/src/protocols/anthropic-messages.ts b/packages/ai/src/protocols/anthropic-messages.ts index 6388d4184f4..5931af7330f 100644 --- a/packages/ai/src/protocols/anthropic-messages.ts +++ b/packages/ai/src/protocols/anthropic-messages.ts @@ -6,8 +6,12 @@ import { Auth } from "../route/auth.js" import { Endpoint } from "../route/endpoint.js" import { Framing } from "../route/framing.js" import { Protocol } from "../route/protocol.js" +import { Headers } from "effect/unstable/http" +import { HttpTransport } from "../route/transport/index.js" import { AIError, + HttpOptions, + LLMRequest, LLMEvent, mergeJsonRecords, Usage, @@ -15,7 +19,6 @@ import { type FinishReasonDetails, type FinishReason, type JsonSchema, - type LLMRequest, type MediaPart, type ProviderMetadata, type ToolCallPart, @@ -61,6 +64,7 @@ export type ThinkingInput = )) export interface OptionsInput { + readonly contextManagement?: ContextManagement readonly [key: string]: unknown readonly thinking?: ThinkingInput readonly effort?: string @@ -89,6 +93,23 @@ export interface OptionsInput { export type ProviderOptionsInput = OptionsInput +export const ContextManagement = Schema.Struct({ + edits: Schema.Array( + Schema.Struct({ + type: Schema.Literal("compact_20260112"), + trigger: Schema.optional( + Schema.Struct({ + type: Schema.Literal("input_tokens"), + value: Schema.Int.check(Schema.isGreaterThanOrEqualTo(50000)), + }), + ), + pauseAfterCompaction: Schema.optional(Schema.Boolean), + instructions: Schema.optional(Schema.String), + }), + ), +}) +export type ContextManagement = typeof ContextManagement.Type + // ============================================================================= // Request Body Schema // ============================================================================= @@ -236,7 +257,12 @@ const AnthropicUserBlock = Schema.Union([ AnthropicToolResultBlock, ]) type AnthropicUserBlock = Schema.Schema.Type +const AnthropicCompactionBlock = Schema.Struct({ + type: Schema.Literal("compaction"), + content: Schema.NullOr(Schema.String), +}) const AnthropicAssistantBlock = Schema.Union([ + AnthropicCompactionBlock, AnthropicTextBlock, AnthropicThinkingBlock, AnthropicRedactedThinkingBlock, @@ -312,6 +338,18 @@ const AnthropicContainer = Schema.Union([ ]) const AnthropicBodyFields = { + context_management: Schema.optional( + Schema.Struct({ + edits: Schema.Array( + Schema.Struct({ + type: Schema.Literal("compact_20260112"), + trigger: ContextManagement.fields.edits.value.fields.trigger, + pause_after_compaction: Schema.optional(Schema.Boolean), + instructions: Schema.optional(Schema.String), + }), + ), + }), + ), model: Schema.String, system: optionalArray(AnthropicTextBlock), messages: Schema.Array(AnthropicMessage), @@ -335,7 +373,7 @@ const AnthropicBodyFields = { export const AnthropicMessagesBody = Schema.Struct(AnthropicBodyFields) export type AnthropicMessagesBody = Schema.Schema.Type -const AnthropicUsage = Schema.StructWithRest( +const AnthropicIterationUsage = Schema.StructWithRest( Schema.Struct({ input_tokens: optionalNull(Schema.Number), output_tokens: Schema.optional(Schema.Number), @@ -354,6 +392,13 @@ const AnthropicUsage = Schema.StructWithRest( }), [Schema.Record(Schema.String, Schema.Unknown)], ) +const AnthropicUsage = Schema.StructWithRest( + Schema.Struct({ + ...AnthropicIterationUsage.schema.fields, + iterations: Schema.optional(Schema.Array(AnthropicIterationUsage)), + }), + [JsonObject], +) type AnthropicUsage = Schema.Schema.Type const AnthropicStreamBlock = Schema.Struct({ @@ -377,6 +422,7 @@ type AnthropicStreamBlock = Schema.Schema.Type const decodeAnthropicStreamBlock = Schema.decodeUnknownOption(AnthropicStreamBlock) const AnthropicStreamDelta = Schema.Struct({ + content: optionalNull(Schema.String), type: Schema.optional(Schema.String), text: Schema.optional(Schema.String), thinking: Schema.optional(Schema.String), @@ -406,6 +452,8 @@ const AnthropicEvent = Schema.Struct({ type AnthropicEvent = Schema.Schema.Type interface ParserState { + readonly provider: LLMRequest["model"]["provider"] + readonly compactions: Readonly> readonly providerMetadataKey: string readonly tools: ToolStream.State readonly reasoningSignatures: Readonly> @@ -848,6 +896,12 @@ const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* ( if (message.role === "assistant") { const content: AnthropicAssistantBlock[] = [] for (const part of message.content) { + if (part.type === "compaction") { + if (part.provider !== request.model.provider || part.text === undefined) + return yield* invalid("Compaction state must be replayed to its originating provider and API") + content.push({ type: "compaction", content: part.text }) + continue + } if (part.type === "text") { if (part.text.trim().length === 0) continue content.push({ type: "text", text: part.text, cache_control: cacheControl(breakpoints, part.cache) }) @@ -1003,6 +1057,9 @@ const resolveThinking = Effect.fn("AnthropicMessages.resolveThinking")(function* }) const fromRequest = Effect.fn("AnthropicMessages.fromRequest")(function* (request: LLMRequest) { + const management = yield* ProviderShared.validateWith( + Schema.decodeUnknownEffect(Schema.UndefinedOr(ContextManagement)), + )(request.providerOptions?.contextManagement) const generation = request.generation const toolSchemaCompatibility = request.model.compatibility?.toolSchema // Allocate the 4-breakpoint budget in invalidation order: tools → system → @@ -1037,7 +1094,7 @@ const fromRequest = Effect.fn("AnthropicMessages.fromRequest")(function* (reques ) } const options = yield* resolveOptions(request) - return { + const body = { model: request.model.id, system, messages, @@ -1058,6 +1115,18 @@ const fromRequest = Effect.fn("AnthropicMessages.fromRequest")(function* (reques metadata: options.metadata, service_tier: options.service_tier, } + if (!management) return body + return { + ...body, + context_management: { + edits: management.edits.map((edit) => ({ + type: edit.type, + trigger: edit.trigger, + pause_after_compaction: edit.pauseAfterCompaction, + instructions: edit.instructions, + })), + }, + } }) // ============================================================================= @@ -1079,18 +1148,31 @@ const mapFinishReason = (reason: string | null | undefined): FinishReason => { // expose that subset through `output_tokens_details.thinking_tokens`. const mapUsage = (usage: AnthropicUsage | undefined, providerMetadataKey: string): Usage | undefined => { if (!usage) return undefined - const nonCached = usage.input_tokens ?? undefined - const cacheRead = usage.cache_read_input_tokens ?? undefined - const cacheWrite = usage.cache_creation_input_tokens ?? undefined + const iterations = usage.iterations?.length ? usage.iterations : [usage] + const last = usage.iterations?.at(-1) + const nonCached = ProviderShared.sumTokens(...iterations.map((item) => item.input_tokens ?? undefined)) + const cacheRead = ProviderShared.sumTokens(...iterations.map((item) => item.cache_read_input_tokens ?? undefined)) + const cacheWrite = ProviderShared.sumTokens( + ...iterations.map((item) => item.cache_creation_input_tokens ?? undefined), + ) const inputTokens = ProviderShared.sumTokens(nonCached, cacheRead, cacheWrite) + const outputTokens = ProviderShared.sumTokens(...iterations.map((item) => item.output_tokens)) return new Usage({ inputTokens, - outputTokens: usage.output_tokens, + outputTokens, + contextTokens: + last?.type === "message" + ? ProviderShared.sumTokens( + last.input_tokens ?? undefined, + last.cache_read_input_tokens ?? undefined, + last.cache_creation_input_tokens ?? undefined, + ) + : undefined, nonCachedInputTokens: nonCached, cacheReadInputTokens: cacheRead, cacheWriteInputTokens: cacheWrite, - reasoningTokens: usage.output_tokens_details?.thinking_tokens, - totalTokens: ProviderShared.totalTokens(inputTokens, usage.output_tokens, undefined), + reasoningTokens: ProviderShared.sumTokens(...iterations.map((item) => item.output_tokens_details?.thinking_tokens)), + totalTokens: ProviderShared.totalTokens(inputTokens, outputTokens, undefined), providerMetadata: { [providerMetadataKey]: usage }, }) } @@ -1112,6 +1194,7 @@ const mergeUsage = (left: Usage | undefined, right: Usage | undefined, providerM return new Usage({ inputTokens, outputTokens, + contextTokens: right.contextTokens ?? left.contextTokens, nonCachedInputTokens, cacheReadInputTokens, cacheWriteInputTokens, @@ -1170,7 +1253,6 @@ const onContentBlockStart = ( event: AnthropicEvent & { readonly content_block: AnthropicStreamBlock }, ): StepResult => { const block = event.content_block - if (!block) return [state, NO_EVENTS] if (block.type === "tool_use" || block.type === "server_tool_use") { if (event.index === undefined || !block.id) return [state, NO_EVENTS] @@ -1265,7 +1347,16 @@ const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(f ) { const delta = event.delta - if (delta?.type === "text_delta" && delta.text) { + if (delta.type === "compaction_delta") { + if (event.index === undefined || !(event.index in state.compactions) || delta.content === undefined) + return yield* ProviderShared.eventError(ADAPTER, "Compaction delta is missing its block or content") + return [ + { ...state, compactions: { ...state.compactions, [event.index]: delta.content } }, + NO_EVENTS, + ] satisfies StepResult + } + + if (delta.type === "text_delta" && delta.text) { if (!state.lifecycle.text.has(`text-${event.index ?? 0}`)) return [state, NO_EVENTS] satisfies StepResult const events: LLMEvent[] = [] return [ @@ -1274,7 +1365,7 @@ const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(f ] satisfies StepResult } - if (delta?.type === "thinking_delta" && delta.thinking) { + if (delta.type === "thinking_delta" && delta.thinking) { if (!state.lifecycle.reasoning.has(`reasoning-${event.index ?? 0}`)) return [state, NO_EVENTS] satisfies StepResult const events: LLMEvent[] = [] return [ @@ -1286,7 +1377,7 @@ const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(f ] satisfies StepResult } - if (delta?.type === "signature_delta" && delta.signature) { + if (delta.type === "signature_delta" && delta.signature) { const index = event.index ?? 0 if (!state.lifecycle.reasoning.has(`reasoning-${index}`)) return [state, NO_EVENTS] satisfies StepResult return [ @@ -1298,7 +1389,7 @@ const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(f ] satisfies StepResult } - if (delta?.type === "input_json_delta" && event.index !== undefined) { + if (delta.type === "input_json_delta" && event.index !== undefined) { if (!delta.partial_json) return [state, NO_EVENTS] satisfies StepResult if (!state.tools[event.index]) return [state, NO_EVENTS] satisfies StepResult const result = ToolStream.appendExisting( @@ -1323,6 +1414,18 @@ const onContentBlockStop = Effect.fn("AnthropicMessages.onContentBlockStop")(fun event: AnthropicEvent, ) { if (event.index === undefined) return [state, NO_EVENTS] satisfies StepResult + if (event.index in state.compactions) { + const { [event.index]: content, ...compactions } = state.compactions + const events: LLMEvent[] = [] + const lifecycle = Lifecycle.stepStart(state.lifecycle, events) + events.push( + LLMEvent.compaction({ + provider: state.provider, + text: content, + }), + ) + return [{ ...state, compactions, lifecycle }, events] satisfies StepResult + } const result = yield* ToolStream.finish(ADAPTER, state.tools, event.index) const events: LLMEvent[] = [] const resultEvents = result.events ?? [] @@ -1374,6 +1477,8 @@ const onMessageDelta = ( } const onMessageStop = Effect.fn("AnthropicMessages.onMessageStop")(function* (state: ParserState) { + if (Object.keys(state.compactions).length) + return yield* ProviderShared.eventError(ADAPTER, "Response ended with an incomplete compaction block") const result = yield* ToolStream.finishAll(ADAPTER, state.tools) const events: LLMEvent[] = [] const lifecycle = result.events.length ? Lifecycle.stepStart(state.lifecycle, events) : state.lifecycle @@ -1418,16 +1523,21 @@ const onError = (event: AnthropicEvent) => { ) } -const isKnownStreamBlockType = (type: string) => - type === "text" || - type === "thinking" || - type === "redacted_thinking" || - type === "tool_use" || - type === "server_tool_use" || - isServerToolResultType(type) - -const isKnownStreamDeltaType = (type: string) => - type === "text_delta" || type === "thinking_delta" || type === "signature_delta" || type === "input_json_delta" +const STREAM_BLOCK_TYPES = new Set([ + "compaction", + "text", + "thinking", + "redacted_thinking", + "tool_use", + "server_tool_use", +]) +const STREAM_DELTA_TYPES = new Set([ + "compaction_delta", + "text_delta", + "thinking_delta", + "signature_delta", + "input_json_delta", +]) const invalidStreamEvent = (event: AnthropicEvent) => Effect.fail( @@ -1456,7 +1566,16 @@ const step = (state: ParserState, event: AnthropicEvent) => { if (event.type === "content_block_start") { if (!ProviderShared.isRecord(event.content_block) || typeof event.content_block.type !== "string") return invalidStreamEvent(event) - if (!isKnownStreamBlockType(event.content_block.type)) return Effect.succeed([state, NO_EVENTS]) + if (event.content_block.type === "compaction") { + const decoded = Schema.decodeUnknownOption(AnthropicCompactionBlock)(event.content_block) + if (event.index === undefined || Option.isNone(decoded)) return invalidStreamEvent(event) + return Effect.succeed([ + { ...state, compactions: { ...state.compactions, [event.index]: decoded.value.content } }, + NO_EVENTS, + ]) + } + if (!STREAM_BLOCK_TYPES.has(event.content_block.type) && !isServerToolResultType(event.content_block.type)) + return Effect.succeed([state, NO_EVENTS]) const decoded = decodeAnthropicStreamBlock(event.content_block) if (Option.isNone(decoded)) return invalidStreamEvent(event) const block = decoded.value @@ -1470,7 +1589,7 @@ const step = (state: ParserState, event: AnthropicEvent) => { } if (event.type === "content_block_delta") { if (!ProviderShared.isRecord(event.delta)) return invalidStreamEvent(event) - if (typeof event.delta.type === "string" && !isKnownStreamDeltaType(event.delta.type)) + if (typeof event.delta.type === "string" && !STREAM_DELTA_TYPES.has(event.delta.type)) return Effect.succeed([state, NO_EVENTS]) const decoded = decodeAnthropicStreamDelta(event.delta) if (Option.isNone(decoded)) return invalidStreamEvent(event) @@ -1504,6 +1623,8 @@ export const protocol = Protocol.make({ stream: { event: Protocol.jsonEvent(AnthropicEvent), initial: (request) => ({ + provider: request.model.provider, + compactions: {}, providerMetadataKey: request.model.route.providerMetadataKey ?? String(request.model.provider), tools: ToolStream.empty(), reasoningSignatures: {}, @@ -1513,6 +1634,37 @@ export const protocol = Protocol.make({ }, }) +export const transport = >() => { + const http = HttpTransport.httpJson({ framing }) + return { + ...http, + prepare: (input: Parameters[0]) => { + if ( + !input.body.context_management?.edits.length && + !input.body.messages.some((message) => message.content.some((block) => block.type === "compaction")) + ) + return http.prepare(input) + const headers = Headers.fromInput(input.request.http?.headers) + const betas = new Set( + (headers["anthropic-beta"] ?? "") + .split(",") + .map((item) => item.trim()) + .filter(Boolean), + ) + betas.add("compact-2026-01-12") + return http.prepare({ + ...input, + request: LLMRequest.update(input.request, { + http: new HttpOptions({ + ...input.request.http, + headers: { ...headers, "anthropic-beta": [...betas].join(",") }, + }), + }), + }) + }, + } +} + export const route = Route.make({ id: ADAPTER, provider: "anthropic", @@ -1522,7 +1674,7 @@ export const route = Route.make({ baseURL: DEFAULT_BASE_URL, }), auth: Auth.none, - framing, + transport: transport(), headers: () => ({ "anthropic-version": "2023-06-01" }), }) diff --git a/packages/ai/src/protocols/bedrock-messages.ts b/packages/ai/src/protocols/bedrock-messages.ts index 7a553494057..71dbc38f6b1 100644 --- a/packages/ai/src/protocols/bedrock-messages.ts +++ b/packages/ai/src/protocols/bedrock-messages.ts @@ -43,6 +43,11 @@ export const protocol = Protocol.make({ .map((value) => value.trim()) .filter(Boolean), ) + if ( + body.context_management?.edits.length || + body.messages.some((message) => message.content.some((block) => block.type === "compaction")) + ) + betas.add("compact-2026-01-12") return { ...Struct.omit(body, ["model", "stream"]), anthropic_version: VERSION, diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 75051c335c8..76095307827 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -32,11 +32,11 @@ export const PATH = "/responses" // ============================================================================= // Request Body Schema // ============================================================================= -const OpenResponsesInputText = Schema.Struct({ +export const OpenResponsesInputText = Schema.Struct({ type: Schema.tag("input_text"), text: Schema.String, }) -const OpenResponsesInputImage = Schema.Struct({ +export const OpenResponsesInputImage = Schema.Struct({ type: Schema.tag("input_image"), image_url: Schema.String, detail: Schema.optional(Schema.String), @@ -55,7 +55,7 @@ const MediaInput = Schema.Union([OpenResponsesInputImage, OpenResponsesInputFile export type MediaInput = Schema.Schema.Type const OpenResponsesInputContent = Schema.Union([OpenResponsesInputText, MediaInput]) -const OpenResponsesOutputText = Schema.Struct({ +export const OpenResponsesOutputText = Schema.Struct({ type: Schema.tag("output_text"), text: Schema.String, }) @@ -63,6 +63,13 @@ const OpenResponsesOutputText = Schema.Struct({ export const MessagePhase = Schema.NullOr(Schema.Literals(["commentary", "final_answer"])) type MessagePhase = Schema.Schema.Type +export const MessageMetadata = Schema.Struct({ + itemId: Schema.optional(Schema.String), + type: Schema.optional(Schema.Literal("message")), + status: Schema.optional(Schema.String), + phase: Schema.optional(MessagePhase), +}) + const messagePhase = (value: unknown): MessagePhase | undefined => { if (value === null || value === "commentary" || value === "final_answer") return value return undefined @@ -73,7 +80,7 @@ const OpenResponsesReasoningSummaryText = Schema.Struct({ text: Schema.String, }) -const OpenResponsesReasoningItem = Schema.Struct({ +export const OpenResponsesReasoningItem = Schema.Struct({ type: Schema.tag("reasoning"), id: Schema.optionalKey(Schema.String), summary: Schema.Array(OpenResponsesReasoningSummaryText), @@ -150,16 +157,30 @@ const OpenResponsesFunctionCallOutput = Schema.Union([ Schema.Array(OpenResponsesFunctionCallOutputContent), ]) +export const CompactionItem = Schema.Struct({ + type: Schema.Literal("compaction"), + id: optionalNull(Schema.String), + encrypted_content: Schema.String, +}) + export const InputItem = Schema.Union([ + CompactionItem, Schema.Struct({ role: Schema.tag("system"), content: Schema.String }), Schema.Struct({ role: Schema.tag("developer"), content: Schema.String }), - Schema.Struct({ role: Schema.tag("user"), content: Schema.Array(OpenResponsesInputContent) }), + Schema.Struct({ + role: Schema.tag("user"), + content: Schema.Array(OpenResponsesInputContent), + type: Schema.optional(Schema.Literal("message")), + id: Schema.optional(Schema.String), + status: Schema.optional(Schema.String), + }), Schema.Struct({ type: Schema.tag("message"), id: Schema.optionalKey(Schema.String), role: Schema.tag("assistant"), content: Schema.Array(OpenResponsesOutputText), phase: Schema.optionalKey(MessagePhase), + status: Schema.optional(Schema.String), }), OpenResponsesReasoningItem, Schema.Struct({ @@ -268,7 +289,7 @@ const OpenResponsesBody = Schema.Struct({ }) export type OpenResponsesBody = Schema.Schema.Type -const OpenResponsesUsage = Schema.Struct({ +export const OpenResponsesUsage = Schema.Struct({ input_tokens: Schema.optional(Schema.Number), input_tokens_details: optionalNull( Schema.Struct({ @@ -388,6 +409,8 @@ export interface Extension { const BASE: Extension = { id: ADAPTER, name: NAME } export interface ParserState { + readonly provider: LLMRequest["model"]["provider"] + readonly completedCompactions: ReadonlySet readonly id: string readonly name: string readonly providerMetadataKey: string @@ -575,6 +598,9 @@ const lowerMessages = Effect.fn("OpenResponses.lowerMessages")(function* (reques const providerMetadataKey = request.model.route.providerMetadataKey ?? "openresponses" for (const message of request.messages) { + const metadata = yield* ProviderShared.validateWith( + Schema.decodeUnknownEffect(Schema.UndefinedOr(MessageMetadata)), + )(message.providerMetadata?.[providerMetadataKey]) if (message.role === "system") { input.push({ role: "developer", @@ -585,7 +611,8 @@ const lowerMessages = Effect.fn("OpenResponses.lowerMessages")(function* (reques if (message.role === "user") { const content = yield* Effect.forEach(message.content, (part) => lowerUserContent(part, request, extension)) - if (content.length > 0) input.push({ role: "user", content }) + if (content.length > 0) + input.push({ role: "user", content, type: metadata?.type, id: metadata?.itemId, status: metadata?.status }) continue } @@ -598,9 +625,10 @@ const lowerMessages = Effect.fn("OpenResponses.lowerMessages")(function* (reques const groups = content.reduce< Array<{ id: string | undefined; phase: MessagePhase | null | undefined; parts: TextPart[] }> >((groups, part) => { - const metadata = part.providerMetadata?.[providerMetadataKey] - const id = itemID(part.providerMetadata, providerMetadataKey) - const phase = ProviderShared.isRecord(metadata) ? messagePhase(metadata.phase) : undefined + const partMetadata = part.providerMetadata?.[providerMetadataKey] + const id = itemID(part.providerMetadata, providerMetadataKey) ?? metadata?.itemId + const partPhase = messagePhase(partMetadata?.phase) + const phase = partPhase === undefined ? metadata?.phase : partPhase const group = groups.at(-1) if (group && group.id === id && group.phase === phase) group.parts.push(part) else groups.push({ id, phase, parts: [part] }) @@ -611,6 +639,7 @@ const lowerMessages = Effect.fn("OpenResponses.lowerMessages")(function* (reques type: "message" as const, ...(group.id === undefined ? {} : { id: group.id }), role: "assistant" as const, + status: metadata?.status, content: group.parts.map((part) => ({ type: "output_text" as const, text: part.text })), ...(group.phase === undefined ? {} : { phase: group.phase }), })), @@ -618,6 +647,15 @@ const lowerMessages = Effect.fn("OpenResponses.lowerMessages")(function* (reques content.splice(0, content.length) } for (const part of message.content) { + if (part.type === "compaction") { + flushText() + if (part.provider !== request.model.provider || part.encrypted === undefined) + return yield* ProviderShared.invalidRequest( + "Compaction state must be replayed to its originating provider and API", + ) + input.push({ type: "compaction", id: part.id, encrypted_content: part.encrypted }) + continue + } if (part.type === "text") { content.push(part) continue @@ -794,7 +832,7 @@ export const fromRequest = Effect.fn("OpenResponses.fromRequest")(function* (req // cached-read and cache-write subsets, and `output_tokens` (inclusive total) // with a `reasoning_tokens` subset. Pass the totals through and derive the // non-cached breakdown. -const mapUsage = (usage: OpenResponsesUsage | null | undefined, providerMetadataKey: string) => { +export const mapUsage = (usage: OpenResponsesUsage | null | undefined, providerMetadataKey: string) => { if (!usage) return undefined const cached = usage.input_tokens_details?.cached_tokens const cacheWrite = usage.input_tokens_details?.cache_write_tokens @@ -1100,6 +1138,25 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( ) { if (!item) return [state, NO_EVENTS] satisfies StepResult + if (item.type === "compaction") { + if (!item.id || typeof item.encrypted_content !== "string") + return yield* ProviderShared.eventError(state.id, "Compaction output is missing its id or encrypted content") + if (state.completedCompactions.has(item.id)) return [state, NO_EVENTS] satisfies StepResult + const events: LLMEvent[] = [] + const lifecycle = Lifecycle.stepStart(state.lifecycle, events) + events.push( + LLMEvent.compaction({ + provider: state.provider, + id: item.id, + encrypted: item.encrypted_content, + }), + ) + return [ + { ...state, lifecycle, completedCompactions: new Set([...state.completedCompactions, item.id]) }, + events, + ] satisfies StepResult + } + if (item.type === "message" && item.id !== undefined) { const message = state.message?.id === item.id ? state.message : undefined const itemPhase = messagePhase(item.phase) @@ -1259,26 +1316,34 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* ( const events: LLMEvent[] = [] if (event.type === "response.completed") { for (const item of event.response?.output ?? []) { - const id = item.id ?? (item.type === "function_call" ? item.call_id : undefined) - if (id === undefined) continue - if (item.type !== "function_call" || !current.tools[id]) continue + if (item.type !== "compaction" && item.type !== "function_call") continue + if (item.type === "compaction") { + // Terminal recovery cannot insert a checkpoint before already-emitted content. + if (state.lifecycle.stepStarted && !state.completedCompactions.has(item.id ?? "")) + return yield* ProviderShared.eventError( + state.id, + "Cannot recover a compaction checkpoint after output has been emitted", + ) + } + if (item.type === "function_call" && !current.tools[item.id ?? item.call_id ?? ""]) continue const [next, emitted] = yield* onOutputItemDone(current, item) current = next events.push(...emitted) } + // Some compatible providers omit output_item.done even after completing the response. + const pending = yield* ToolStream.finishAll(current.id, current.tools) + current = { + ...current, + tools: pending.tools, + hasFunctionCall: + current.hasFunctionCall || + pending.events.some((event) => LLMEvent.is.toolCall(event) || LLMEvent.is.toolInputError(event)), + } + events.push(...pending.events) } - // Some compatible providers omit output_item.done even after completing the response. - const pending = - event.type === "response.completed" - ? yield* ToolStream.finishAll(current.id, current.tools) - : { tools: current.tools, events: NO_EVENTS } - events.push(...pending.events) - const hasFunctionCall = - pending.events.some((event) => LLMEvent.is.toolCall(event) || LLMEvent.is.toolInputError(event)) || - current.hasFunctionCall const lifecycle = Lifecycle.finish(current.lifecycle, events, { reason: { - normalized: mapFinishReason(event, hasFunctionCall), + normalized: mapFinishReason(event, current.hasFunctionCall), raw: event.response?.incomplete_details?.reason, }, usage: mapUsage(event.response?.usage, current.providerMetadataKey), @@ -1290,7 +1355,7 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* ( }) : undefined, }) - return [{ ...current, lifecycle, hasFunctionCall, tools: pending.tools }, events] satisfies StepResult + return [{ ...current, lifecycle }, events] satisfies StepResult }) // Build the prettiest summary available from whatever the provider supplied. @@ -1425,6 +1490,8 @@ export const step = (state: ParserState, input: Event) => { * implementations compose this baseline with their own tools and event variants. */ export const initial = (request: LLMRequest, extension: Extension = BASE): ParserState => ({ + provider: request.model.provider, + completedCompactions: new Set(), id: extension.id, name: extension.name, providerMetadataKey: request.model.route.providerMetadataKey ?? "openresponses", diff --git a/packages/ai/src/protocols/openai-responses.ts b/packages/ai/src/protocols/openai-responses.ts index 35942d60e21..1fae92a7a4c 100644 --- a/packages/ai/src/protocols/openai-responses.ts +++ b/packages/ai/src/protocols/openai-responses.ts @@ -12,6 +12,7 @@ import { OpenAIImage } from "./utils/openai-image.js" import { ResponsesHostedTools } from "./utils/responses-hosted-tools.js" import { ToolSchemaProjection } from "./utils/tool-schema.js" import { OpenResponsesChannel } from "./open-responses-channel.js" +import { ResponsesCompaction } from "./utils/responses-compaction.js" const ADAPTER = "openai-responses" const NAME = "OpenAI Responses" @@ -20,6 +21,14 @@ const WEBSOCKET_ROTATE_AFTER_MS = 55 * 60 * 1000 export const DEFAULT_BASE_URL = "https://api.openai.com/v1" export const PATH = OpenResponses.PATH +export const ContextManagement = Schema.Array( + Schema.Struct({ + type: Schema.Literal("compaction"), + compactThreshold: Schema.optional(Schema.Int.check(Schema.isGreaterThan(0))), + }), +) +export type ContextManagement = typeof ContextManagement.Type + const OpenAIResponsesImageGenerationTool = Schema.Struct({ type: Schema.tag("image_generation"), action: Schema.optional(Schema.Literals(["auto", "generate", "edit"])), @@ -78,6 +87,14 @@ const OpenAIResponsesCoreFields = { input: Schema.Array(Schema.Union([OpenResponses.InputItem, OpenAIResponsesHostedToolItem])), tools: optionalArray(OpenAIResponsesTools), tool_choice: Schema.optional(OpenAIResponsesToolChoice), + context_management: Schema.optional( + Schema.Array( + Schema.Struct({ + type: Schema.Literal("compaction"), + compact_threshold: Schema.optional(Schema.Int.check(Schema.isGreaterThan(0))), + }), + ), + ), } const OpenAIResponsesBody = Schema.Struct({ @@ -125,10 +142,14 @@ const lowerToolChoice = (toolChoice: NonNullable, tool const decodeBody = ProviderShared.validateWith(Schema.decodeUnknownEffect(OpenAIResponsesBody)) const fromRequest = Effect.fn("OpenAIResponses.fromRequest")(function* (request: LLMRequest) { + const management = yield* ProviderShared.validateWith( + Schema.decodeUnknownEffect(Schema.UndefinedOr(ContextManagement)), + )(request.providerOptions?.contextManagement) const toolSchemaCompatibility = request.model.compatibility?.toolSchema return yield* decodeBody({ ...(yield* OpenResponses.lowerConversation(request, extension)), ...OpenResponses.lowerGeneration(request), + context_management: management?.map((edit) => ({ type: edit.type, compact_threshold: edit.compactThreshold })), tools: request.tools.length === 0 ? undefined @@ -219,6 +240,7 @@ export const transport = channelTransport({ }) export const route = Route.make({ + compact: ResponsesCompaction.make(extension), id: ADAPTER, provider: "openai", providerMetadataKey: "openai", diff --git a/packages/ai/src/protocols/utils/responses-compaction.ts b/packages/ai/src/protocols/utils/responses-compaction.ts new file mode 100644 index 00000000000..d1d4aede49d --- /dev/null +++ b/packages/ai/src/protocols/utils/responses-compaction.ts @@ -0,0 +1,152 @@ +import { Effect, Schema, Stream } from "effect" +import { + AIError, + InvalidProviderOutputError, + CompactionPart, + CompactionResponse, + HttpOptions, + LLMRequest, + Message, + type ContentPart, + mergeJsonRecords, +} from "../../schema/index.js" +import type { CompactOperation } from "../../route/client.js" +import { Endpoint } from "../../route/endpoint.js" +import { RequestExecutor } from "../../route/executor.js" +import { HttpTransport } from "../../route/transport/index.js" +import { OpenResponses } from "../open-responses.js" +import { JsonObject, ProviderShared } from "../shared.js" + +const Body = Schema.Struct({ + model: Schema.String, + input: Schema.Array(Schema.Unknown), + instructions: Schema.optional(Schema.String), + previous_response_id: Schema.optional(Schema.String), +}) + +const Text = Schema.Union([OpenResponses.OpenResponsesInputText, OpenResponses.OpenResponsesOutputText]) +const File = Schema.Union([ + Schema.Struct({ type: Schema.Literal("input_file"), filename: Schema.String, file_url: Schema.String }), + Schema.Struct({ type: Schema.Literal("input_file"), filename: Schema.String, file_data: Schema.String }), +]) +const MessageFields = { + type: Schema.Literal("message"), + id: Schema.optional(Schema.String), + status: Schema.optional(Schema.String), + phase: Schema.optional(OpenResponses.MessagePhase), +} +const Response = Schema.Struct({ + object: Schema.Literal("response.compaction"), + output: Schema.Array( + Schema.Union([ + OpenResponses.CompactionItem, + OpenResponses.OpenResponsesReasoningItem, + Schema.Struct({ + ...MessageFields, + role: Schema.Literal("user"), + content: Schema.Array(Schema.Union([Text, OpenResponses.OpenResponsesInputImage, File])).check( + Schema.isMinLength(1), + ), + }), + Schema.Struct({ + ...MessageFields, + role: Schema.Literal("assistant"), + content: Schema.Array(Text).check(Schema.isMinLength(1)), + }), + ]), + ), + usage: Schema.optional(Schema.StructWithRest(OpenResponses.OpenResponsesUsage, [JsonObject])), +}) + +export const make = (extension: OpenResponses.Extension): CompactOperation => + Effect.fn("ResponsesCompaction.execute")(function* (request, executor, options) { + const route = request.model.route + const native = yield* OpenResponses.lowerConversation(request, extension) + const body = yield* ProviderShared.validateWith(Schema.decodeUnknownEffect(Body))( + mergeJsonRecords(native, request.http?.body), + ) + const url = Endpoint.render(route.endpoint, { request, body: native }) + url.pathname = `${url.pathname.replace(/\/$/, "")}/compact` + const parts = yield* HttpTransport.jsonRequestParts({ + request: LLMRequest.update(request, { + http: request.http === undefined ? undefined : new HttpOptions({ ...request.http, body: undefined }), + }), + body, + endpoint: Endpoint.path(url.toString()), + auth: route.auth, + encodeBody: Schema.encodeSync(Schema.fromJsonString(Body)), + }) + const response = yield* executor.execute( + ProviderShared.jsonPost({ url: parts.url, body: parts.bodyText, headers: parts.headers }), + options?.http, + ) + const text = yield* RequestExecutor.responseStream(response).pipe( + Stream.decodeText(), + Stream.runFold( + () => "", + (text, chunk) => text + chunk, + ), + ) + const invalid = (message: string, cause?: unknown) => + new AIError({ + reason: new InvalidProviderOutputError({ + route: route.id, + message, + body: text, + cause, + http: RequestExecutor.responseHttp(response), + }), + }) + const result = yield* Schema.decodeUnknownEffect(Schema.fromJsonString(Response))(text).pipe( + Effect.mapError((cause) => invalid("Invalid compaction response", cause)), + ) + if (!result.output.some((item) => item.type === "compaction")) + return yield* invalid("Compaction response did not contain a checkpoint") + return new CompactionResponse({ + messages: result.output.map((item) => toMessage(item, request.model)), + usage: OpenResponses.mapUsage(result.usage, route.providerMetadataKey ?? String(request.model.provider)), + }) + }) + +function toMessage(item: (typeof Response.Type.output)[number], model: LLMRequest["model"]): Message { + if (item.type === "compaction") + return Message.assistant( + CompactionPart.make({ provider: model.provider, id: item.id ?? undefined, encrypted: item.encrypted_content }), + ) + + const key = model.route.providerMetadataKey ?? String(model.provider) + if (item.type === "reasoning") { + const summary = item.summary.length ? item.summary : [{ text: "" }] + return Message.assistant( + summary.map((part) => ({ + type: "reasoning" as const, + text: part.text, + providerMetadata: { [key]: { itemId: item.id, reasoningEncryptedContent: item.encrypted_content } }, + })), + ) + } + + return Message.make({ + role: item.role, + providerMetadata: { [key]: { itemId: item.id, type: item.type, status: item.status, phase: item.phase } }, + content: item.content.map((part): ContentPart => { + if (part.type === "input_text" || part.type === "output_text") return { type: "text", text: part.text } + if (part.type === "input_image") + return { + type: "media", + data: part.image_url, + mediaType: /^data:([^;,]+)/.exec(part.image_url)?.[1] ?? "image/*", + providerMetadata: part.detail === undefined ? undefined : { [key]: { detail: part.detail } }, + } + const data = "file_url" in part ? part.file_url : part.file_data + return { + type: "media", + data, + filename: part.filename, + mediaType: /^data:([^;,]+)/.exec(data)?.[1] ?? "application/octet-stream", + } + }), + }) +} + +export * as ResponsesCompaction from "./responses-compaction.js" diff --git a/packages/ai/src/protocols/xai-responses.ts b/packages/ai/src/protocols/xai-responses.ts index 150237b07e8..c90adbcfac5 100644 --- a/packages/ai/src/protocols/xai-responses.ts +++ b/packages/ai/src/protocols/xai-responses.ts @@ -4,6 +4,7 @@ import type { LLMRequest } from "../schema/index.js" import { OpenResponses } from "./open-responses.js" import { JsonObject, optionalNull, ProviderShared } from "./shared.js" import { ResponsesHostedTools } from "./utils/responses-hosted-tools.js" +import { ResponsesCompaction } from "./utils/responses-compaction.js" const ADAPTER = "xai-responses" const NAME = "xAI Responses" @@ -44,6 +45,10 @@ const extension = { const decodeBody = ProviderShared.validateWith(Schema.decodeUnknownEffect(XAIResponsesBody)) const fromRequest = Effect.fn("XAIResponses.fromRequest")(function* (request: LLMRequest) { + if (request.providerOptions?.contextManagement !== undefined) + return yield* ProviderShared.invalidRequest( + "xAI requires explicit compaction through LLMClient.compact; automatic context management is not supported", + ) return yield* decodeBody(yield* OpenResponses.fromRequestWithExtension(request, extension)) }) @@ -84,4 +89,6 @@ export const protocol = Protocol.make({ }, }) +export const compact = ResponsesCompaction.make(extension) + export * as XAIResponses from "./xai-responses.js" diff --git a/packages/ai/src/providers/google-vertex-messages.ts b/packages/ai/src/providers/google-vertex-messages.ts index 2d67c0cde99..4449992a8b9 100644 --- a/packages/ai/src/providers/google-vertex-messages.ts +++ b/packages/ai/src/providers/google-vertex-messages.ts @@ -57,7 +57,9 @@ const route = Route.make({ }), endpoint: Endpoint.path(({ request }) => `/${request.model.id}:streamRawPredict`), auth: Auth.none, - framing: AnthropicMessages.framing, + transport: AnthropicMessages.transport< + Omit & { readonly anthropic_version: typeof VERSION } + >(), headers: () => ({ "anthropic-version": HEADER_VERSION }), }) diff --git a/packages/ai/src/providers/openai-options.ts b/packages/ai/src/providers/openai-options.ts index 5adaf555a1f..a71b76d11e8 100644 --- a/packages/ai/src/providers/openai-options.ts +++ b/packages/ai/src/providers/openai-options.ts @@ -1,10 +1,12 @@ import { mergeProviderOptions, type ProviderOptions } from "../schema/index.js" import type { OpenAIServiceTier } from "../protocols/utils/openai-options.js" import type { Options } from "../protocols/utils/open-responses-options.js" +import type { ContextManagement } from "../protocols/openai-responses.js" export type { OpenAIResponseIncludable, OpenAIServiceTier } from "../protocols/utils/openai-options.js" export type OpenAIOptionsInput = Omit & { + readonly contextManagement?: ContextManagement readonly serviceTier?: OpenAIServiceTier readonly [key: string]: unknown } diff --git a/packages/ai/src/providers/xai.ts b/packages/ai/src/providers/xai.ts index 50f8973c2a9..58623030ae0 100644 --- a/packages/ai/src/providers/xai.ts +++ b/packages/ai/src/providers/xai.ts @@ -13,7 +13,7 @@ import type { ProviderPackage } from "../provider-package.js" export const id = ProviderID.make("xai") -export type XAIProviderOptionsInput = OpenAIOptionsInput +export type XAIProviderOptionsInput = OpenAIOptionsInput & { readonly contextManagement?: never } export type LanguageModelOptions = Omit & ProviderAuthOption<"optional"> & { @@ -32,6 +32,7 @@ export type { XAIImageOptions } from "../protocols/xai-images.js" const RESPONSES_WEBSOCKET_ROTATE_AFTER_MS = 24 * 60 * 1000 const responsesRoute = Route.make({ + compact: XAIResponses.compact, id: "openai-responses", provider: id, providerMetadataKey: "xai", diff --git a/packages/ai/src/route/client.ts b/packages/ai/src/route/client.ts index 186c20143f8..91625afdba8 100644 --- a/packages/ai/src/route/client.ts +++ b/packages/ai/src/route/client.ts @@ -13,6 +13,7 @@ import * as ProviderShared from "../protocols/shared.js" import type { ProtocolID, ProviderOptions } from "../schema/index.js" import { AIError, + CompactionResponse, AIErrorReason, GenerationOptions, HttpOptions, @@ -35,6 +36,7 @@ export interface RouteBody { } export interface Route { + readonly compact?: CompactOperation readonly id: string readonly provider?: ProviderID /** ProviderMetadata namespace emitted and consumed by this route. */ @@ -42,6 +44,8 @@ export interface Route { readonly protocol: ProtocolID readonly endpoint: Endpoint.Definition readonly auth: Auth.Definition + /** Deployment headers resolved once for every operation, before transport authentication. */ + readonly headers?: (input: { readonly request: LLMRequest }) => Record readonly transport: Transport readonly defaults: RouteDefaults readonly body: RouteBody @@ -150,6 +154,10 @@ export const httpOptions = (input: HttpOptionsInput | undefined) => { } export interface Interface { + readonly compact: ( + request: LLMRequest, + options?: Pick, + ) => Effect.Effect readonly stream: StreamMethod readonly generate: GenerateMethod } @@ -167,6 +175,12 @@ export interface GenerateMethod { (request: LLMRequest, options?: StreamOptions): Effect.Effect } +export type CompactOperation = ( + request: LLMRequest, + executor: RequestExecutor.Interface, + options?: Pick, +) => Effect.Effect + export class Service extends Context.Service()("@opencode/LLMClient") {} const resolveRequestOptions = (request: LLMRequest) => { @@ -187,6 +201,7 @@ const resolveRequestOptions = (request: LLMRequest) => { } export interface MakeInput { + readonly compact?: CompactOperation /** Route id used in diagnostics and prepared request metadata. */ readonly id: string /** Provider identity for route-owned model construction. */ @@ -208,6 +223,7 @@ export interface MakeInput { } export interface MakeTransportInput { + readonly compact?: CompactOperation /** Route id used in diagnostics and prepared request metadata. */ readonly id: string /** Provider identity for route-owned model construction. */ @@ -283,12 +299,14 @@ function makeFromTransport( const build = (routeInput: BuiltRouteInput): Route => { const route: Route = { + compact: routeInput.compact, id: routeInput.id, provider: routeInput.provider === undefined ? undefined : ProviderID.make(routeInput.provider), providerMetadataKey: routeInput.providerMetadataKey, protocol: protocol.id, endpoint: routeInput.endpoint, auth: routeInput.auth ?? Auth.none, + headers: routeInput.headers, transport: routeInput.transport, defaults: routeInput.defaults ?? {}, body: protocol.body, @@ -318,7 +336,6 @@ function makeFromTransport( endpoint: routeInput.endpoint, auth: routeInput.auth ?? Auth.none, encodeBody, - headers: routeInput.headers, middleware: options?.http, webSocket: options?.webSocket, }), @@ -435,6 +452,7 @@ export function make( if ("transport" in input) return makeFromTransport(input) const protocol = input.protocol return makeFromTransport({ + compact: input.compact, id: input.id, provider: input.provider, providerMetadataKey: input.providerMetadataKey, @@ -447,11 +465,19 @@ export function make( }) } -const compile = Effect.fn("LLM.compile")(function* (request: LLMRequest, options?: StreamOptions) { +const prepareRequest = (request: LLMRequest) => { const original = applyCachePolicy(resolveRequestOptions(request)) const sanitized = LLMRequest.update(original, sanitizeSurrogates({ ...LLMRequest.input(original), model: undefined })) const tools = [...new Map(sanitized.tools.map((tool) => [tool.name, tool])).values()] const resolved = tools.length === sanitized.tools.length ? sanitized : LLMRequest.update(sanitized, { tools }) + const headers = resolved.model.route.headers?.({ request: resolved }) + return headers === undefined + ? resolved + : LLMRequest.update(resolved, { http: mergeHttpOptions(new HttpOptions({ headers }), resolved.http) }) +} + +const compile = Effect.fn("LLM.compile")(function* (request: LLMRequest, options?: StreamOptions) { + const resolved = prepareRequest(request) const route = resolved.model.route const body = yield* route.body @@ -510,6 +536,15 @@ export function generate(request: LLMRequest, options?: StreamOptions): Effect.E }) } +export const compact = ( + request: LLMRequest, + options?: Pick, +): Effect.Effect => + Effect.gen(function* () { + const client = yield* Service + return yield* client.compact(request, options) + }) + export const streamRequest = (request: LLMRequest, options?: StreamOptions) => Stream.unwrap( Effect.gen(function* () { @@ -520,16 +555,28 @@ export const streamRequest = (request: LLMRequest, options?: StreamOptions) => export const layer: Layer.Layer = Layer.effect( Service, Effect.gen(function* () { - const stream = streamRequestWith({ - http: yield* RequestExecutor.Service, + const executor = yield* RequestExecutor.Service + const stream = streamRequestWith({ http: executor }) + return Service.of({ + stream, + generate: generateWith(stream), + compact: (request, options) => + Effect.suspend(() => { + const operation = request.model.route.compact + if (!operation) + return ProviderShared.invalidRequest( + `${request.model.provider}/${request.model.route.id} does not support explicit compaction`, + ) + return operation(prepareRequest(request), executor, options) + }), }) - return Service.of({ stream, generate: generateWith(stream) }) }), ) export const Route = { make } as const export const LLMClient = { + compact, Service, layer, stream, diff --git a/packages/ai/src/schema/events.ts b/packages/ai/src/schema/events.ts index c1357fb8f1f..2491ebfee45 100644 --- a/packages/ai/src/schema/events.ts +++ b/packages/ai/src/schema/events.ts @@ -3,6 +3,7 @@ import { LLM } from "@opencode-ai/schema/llm" import { ContentBlockID, ToolCallID } from "./ids.js" import { Message, + CompactionPart, ProviderMetadata, ToolCallPart, ToolOutput, @@ -62,6 +63,8 @@ export { ProviderMetadata } from "./messages.js" * Matches the same escape-hatch field on `LLMEvent`. */ export class Usage extends Schema.Class("AI.Usage")({ + /** Effective input size of the final message iteration, when reported; not billed totals. */ + contextTokens: Schema.optional(Schema.Number), inputTokens: Schema.optional(Schema.Number), outputTokens: Schema.optional(Schema.Number), nonCachedInputTokens: Schema.optional(Schema.Number), @@ -72,7 +75,7 @@ export class Usage extends Schema.Class("AI.Usage")({ providerMetadata: Schema.optional(ProviderMetadata), }) { /** - * Visible output tokens — `outputTokens` minus `reasoningTokens`, clamped + * Non-reasoning output tokens (including compaction summaries) — `outputTokens` minus `reasoningTokens`, clamped * to zero. The one place subtraction happens in this contract; the clamp * means a provider reporting `reasoningTokens > outputTokens` produces a * harmless zero rather than a negative that crashes downstream schemas. @@ -88,6 +91,12 @@ export class Usage extends Schema.Class("AI.Usage")({ export type UsageInput = Usage | ConstructorParameters[0] +/** A replacement context window. Replace prior history with these messages. */ +export class CompactionResponse extends Schema.Class("LLM.CompactionResponse")({ + messages: Schema.Array(Message), + usage: Schema.optional(Usage), +}) {} + export const StepStart = Schema.Struct({ type: Schema.tag("step-start"), index: Schema.Number, @@ -241,6 +250,7 @@ export const ProviderErrorEvent = Schema.Struct({ export type ProviderErrorEvent = Schema.Schema.Type const llmEventTagged = Schema.Union([ + CompactionPart, StepStart, TextStart, TextDelta, @@ -274,6 +284,7 @@ const toolCallID = (value: ToolCallID | string) => ToolCallID.make(value) * `events.filter(LLMEvent.guards["tool-call"])`. */ export const LLMEvent = Object.assign(llmEventTagged, { + compaction: CompactionPart.make, stepStart: StepStart.make, textStart: (input: WithID) => TextStart.make({ ...input, id: contentBlockID(input.id) }), textDelta: (input: WithID) => TextDelta.make({ ...input, id: contentBlockID(input.id) }), @@ -311,6 +322,7 @@ export const LLMEvent = Object.assign(llmEventTagged, { }), providerError: ProviderErrorEvent.make, is: { + compaction: llmEventTagged.guards.compaction, stepStart: llmEventTagged.guards["step-start"], textStart: llmEventTagged.guards["text-start"], textDelta: llmEventTagged.guards["text-delta"], @@ -333,10 +345,10 @@ export const LLMEvent = Object.assign(llmEventTagged, { export type LLMEvent = Schema.Schema.Type /** Joins deltas per fragment, letting an authoritative end value replace that fragment's accumulated deltas. */ -const joinFragments = ( +const joinFragments = ( events: ReadonlyArray, - isDelta: (event: LLMEvent) => event is Extract, - isEnd: (event: LLMEvent) => event is Extract, + isDelta: (event: LLMEvent) => event is LLMEvent & { id: string; text: string }, + isEnd: (event: LLMEvent) => event is LLMEvent & { id: string; text?: string }, ) => { const order: string[] = [] const parts = new Map() @@ -563,6 +575,8 @@ const reduceToolCall = (state: ResponseState, event: ToolCall): ResponseState => const reduceResponseState = (state: ResponseState, event: LLMEvent): ResponseState => { const next = appendEvent(state, event) switch (event.type) { + case "compaction": + return appendContent(next, event) case "text-start": return ensureText(next, event.id, event.providerMetadata) case "text-delta": diff --git a/packages/ai/src/schema/messages.ts b/packages/ai/src/schema/messages.ts index 77b04b03112..afdd00bd39b 100644 --- a/packages/ai/src/schema/messages.ts +++ b/packages/ai/src/schema/messages.ts @@ -9,6 +9,7 @@ import { LanguageModelSchema, ProviderOptions, } from "./options.js" +import { ProviderID } from "./ids.js" export const MessageRole = Schema.Literals(["system", "user", "assistant", "tool"]) export type MessageRole = Schema.Schema.Type @@ -186,9 +187,40 @@ export const ReasoningPart = Schema.Struct({ }).annotate({ identifier: "LLM.Content.Reasoning" }) export type ReasoningPart = Schema.Schema.Type -export const ContentPart = Schema.Union([TextPart, MediaPart, ToolCallPart, ToolResultPart, ReasoningPart]).pipe( - Schema.toTaggedUnion("type"), -) +/** A provider-generated context checkpoint, distinct from visible assistant text. */ +type CompactionContent = + | { readonly encrypted: string; readonly text?: never } + | { readonly text: string | null; readonly encrypted?: never } + +const compactionPartSchema = Schema.Struct({ + type: Schema.Literal("compaction"), + provider: ProviderID, + id: Schema.optional(Schema.String), + encrypted: Schema.optional(Schema.String), + /** Null means the provider failed to produce a summary; prior history must be retained. */ + text: Schema.optional(Schema.NullOr(Schema.String)), +}) + .pipe( + Schema.refine( + (part): part is typeof part & CompactionContent => (part.encrypted !== undefined) !== (part.text !== undefined), + { message: "Compaction requires either encrypted content or a summary" }, + ), + ) + .annotate({ identifier: "LLM.Content.Compaction" }) +export type CompactionPart = typeof compactionPartSchema.Type +export const CompactionPart = Object.assign(compactionPartSchema, { + make: (input: Omit & CompactionContent): CompactionPart => + Schema.decodeUnknownSync(compactionPartSchema)({ type: "compaction", ...input }), +}) + +export const ContentPart = Schema.Union([ + TextPart, + MediaPart, + ToolCallPart, + ToolResultPart, + ReasoningPart, + CompactionPart, +]).pipe(Schema.toTaggedUnion("type")) export type ContentPart = Schema.Schema.Type export class Message extends Schema.Class("LLM.Message")({ @@ -196,6 +228,7 @@ export class Message extends Schema.Class("LLM.Message")({ role: MessageRole, content: Schema.Array(ContentPart), metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)), + providerMetadata: Schema.optional(ProviderMetadata), native: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)), }) {} diff --git a/packages/ai/src/testing.ts b/packages/ai/src/testing.ts index a9beba784ba..1a2d3ac5d98 100644 --- a/packages/ai/src/testing.ts +++ b/packages/ai/src/testing.ts @@ -4,6 +4,7 @@ import { LLMClient } from "./route/client.js" import { LLMEvent, LLMResponse, + CompactionResponse, type FinishReasonDetails, type AIError, type LLMRequest, @@ -133,6 +134,16 @@ const make = (options: LayerOptions) => } }) const test = Test.of({ + compact: (request) => + stream(request).pipe( + Stream.runFold(LLMResponse.empty, LLMResponse.reduce), + Effect.flatMap((state) => { + const response = LLMResponse.complete(state) + if (!response?.message.content.some((part) => part.type === "compaction")) + return Effect.die("TestLLM compaction response must contain a checkpoint and terminal finish event") + return Effect.succeed(new CompactionResponse({ messages: [response.message], usage: response.usage })) + }), + ), stream, generate: (request) => stream(request).pipe( diff --git a/packages/ai/src/tool-history.ts b/packages/ai/src/tool-history.ts index a36bbbee3cc..40e721117f7 100644 --- a/packages/ai/src/tool-history.ts +++ b/packages/ai/src/tool-history.ts @@ -56,6 +56,7 @@ function normalizeToolMessage(message: Message, pending: Map + Effect.gen(function* () { + const checkpoint = { type: "compaction", id: "cmp_local", encrypted_content: "opaque-local-state" } + const calls: string[] = [] + const server = yield* Effect.acquireRelease( + Effect.sync(() => + Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(request) { + const path = new URL(request.url).pathname + calls.push(path) + const body = await request.json() + expect(request.headers.get("authorization")).toBe("Bearer fixture") + if (path === "/v1/responses/compact") { + expect(body.stream).toBeUndefined() + return Response.json({ + object: "response.compaction", + output: [checkpoint], + usage: { input_tokens: 100, output_tokens: 10, total_tokens: 110 }, + }) + } + expect(body.input[0]).toEqual(checkpoint) + expect(body.stream).toBe(true) + if (calls.length === 2) + return new Response( + sseEvents( + { + type: "response.output_item.done", + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "lookup", arguments: "{}" }, + }, + { type: "response.completed", response: { id: "resp_1" } }, + ), + { headers: { "content-type": "text/event-stream" } }, + ) + expect(body.input.at(-2)).toMatchObject({ type: "function_call", call_id: "call_1" }) + expect(body.input.at(-1)).toEqual({ type: "function_call_output", call_id: "call_1", output: "42" }) + const output = sseEvents( + { type: "response.output_item.added", item: { type: "message", id: "msg_1" } }, + { type: "response.output_text.delta", item_id: "msg_1", delta: "The answer is 42." }, + { + type: "response.output_item.done", + item: { type: "message", id: "msg_1", content: [{ type: "output_text", text: "The answer is 42." }] }, + }, + { type: "response.completed", response: { id: "resp_2" } }, + ) + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(output.slice(0, 37))) + controller.enqueue(new TextEncoder().encode(output.slice(37))) + controller.close() + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ) + }, + }), + ), + (server) => Effect.sync(() => server.stop(true)), + ) + const model = OpenAI.configure({ apiKey: "fixture", baseURL: `http://127.0.0.1:${server.port}/v1` }).responses( + "fixture", + ) + const request = LLM.request({ + model, + prompt: "original", + tools: [{ name: "lookup", description: "Lookup a number", inputSchema: { type: "object", properties: {} } }], + }) + const compacted = yield* LLMClient.compact(request) + const messages = [...compacted.messages, Message.user("Look up the answer")] + const first = yield* LLMClient.generate(LLMRequest.update(request, { messages })) + expect(first.toolCalls).toHaveLength(1) + const call = first.toolCalls[0]! + const last = yield* LLMClient.generate( + LLMRequest.update(request, { + messages: [ + ...messages, + first.message, + Message.tool({ id: call.id, name: call.name, result: "42", resultType: "text" }), + ], + }), + ) + expect(last.text).toBe("The answer is 42.") + expect(calls).toEqual(["/v1/responses/compact", "/v1/responses", "/v1/responses"]) + }), +) diff --git a/packages/ai/test/compaction.test.ts b/packages/ai/test/compaction.test.ts new file mode 100644 index 00000000000..3417fa6bf3c --- /dev/null +++ b/packages/ai/test/compaction.test.ts @@ -0,0 +1,49 @@ +import { expect, test } from "bun:test" +import { Schema } from "effect" +import { CompactionPart, LLMEvent, LLMResponse, Message, ProviderID } from "../src/schema/index.js" + +test("compaction survives event assembly and message serialization without becoming text", () => { + const part = CompactionPart.make({ + provider: ProviderID.make("openai"), + id: "cmp_1", + encrypted: "opaque", + }) + const response = LLMResponse.fromEvents([ + LLMEvent.textStart({ id: "before" }), + LLMEvent.textDelta({ id: "before", text: "Before" }), + LLMEvent.textEnd({ id: "before" }), + part, + LLMEvent.textStart({ id: "after" }), + LLMEvent.textDelta({ id: "after", text: "After" }), + LLMEvent.textEnd({ id: "after" }), + LLMEvent.finish({ reason: { normalized: "stop" } }), + ])! + expect(response.message.content.map((part) => part.type)).toEqual(["text", "compaction", "text"]) + expect(response.text).toBe("BeforeAfter") + expect(response.reasoning).toBe("") + expect(response.events.filter(LLMEvent.is.compaction)).toEqual([part]) + const codec = Schema.fromJsonString(Message) + expect(Schema.decodeSync(codec)(Schema.encodeSync(codec)(response.message))).toEqual(response.message) +}) + +test("compaction requires exactly one typed representation", () => { + const provider = ProviderID.make("anthropic") + expect(CompactionPart.make({ provider, text: null })).toEqual({ type: "compaction", provider, text: null }) + const decode = Schema.decodeUnknownSync(CompactionPart) + expect(() => decode({ type: "compaction", provider })).toThrow() + expect(() => decode({ type: "compaction", provider, text: "summary", encrypted: "opaque" })).toThrow() +}) + +test("tagged content and event guards accept both checkpoint representations", () => { + for (const part of [ + CompactionPart.make({ provider: ProviderID.make("openai"), encrypted: "opaque" }), + CompactionPart.make({ provider: ProviderID.make("anthropic"), text: "summary" }), + CompactionPart.make({ provider: ProviderID.make("anthropic"), text: null }), + ]) { + expect(LLMEvent.is.compaction(part)).toBe(true) + expect(LLMEvent.guards.compaction(part)).toBe(true) + const codec = Schema.fromJsonString(Message) + const message = Message.assistant(part) + expect(Schema.decodeSync(codec)(Schema.encodeSync(codec)(message))).toEqual(message) + } +}) diff --git a/packages/ai/test/provider-options/compaction.types.ts b/packages/ai/test/provider-options/compaction.types.ts new file mode 100644 index 00000000000..a3e83b0c201 --- /dev/null +++ b/packages/ai/test/provider-options/compaction.types.ts @@ -0,0 +1,69 @@ +import { Effect } from "effect" +import { CompactionPart, LLM, LLMClient, LLMEvent, Message, ProviderID } from "../../src/index.js" +import { OpenAI, Anthropic, AmazonBedrock } from "../../src/providers.js" + +const openai = OpenAI.configure({ + apiKey: "test", + providerOptions: { contextManagement: [{ type: "compaction", compactThreshold: 100000 }] }, +}).responses("gpt-5.3-codex") +LLMClient.compact(LLM.request({ model: openai, prompt: "hello" })) + +const checkpoint = CompactionPart.make({ provider: ProviderID.make("openai"), id: "cmp_1", encrypted: "opaque" }) +const provider = ProviderID.make("anthropic") +CompactionPart.make({ provider, text: "summary" }) +CompactionPart.make({ provider, text: null }) +// @ts-expect-error A checkpoint must have a representation. +CompactionPart.make({ provider }) +// @ts-expect-error Encrypted and summary representations are mutually exclusive. +CompactionPart.make({ provider, encrypted: "opaque", text: "summary" }) +// @ts-expect-error A failed summary cannot also carry encrypted content. +LLMEvent.compaction({ provider, encrypted: "opaque", text: null }) +// @ts-expect-error The canonical message type also enforces the invariant. +Message.assistant({ type: "compaction", provider }) +if (checkpoint.encrypted !== undefined) { + checkpoint.encrypted satisfies string + checkpoint.text satisfies undefined +} +if (checkpoint.text !== undefined) { + checkpoint.text satisfies string | null + checkpoint.encrypted satisfies undefined +} +checkpoint.encrypted +// @ts-expect-error Compaction parts do not contain a generic provider payload. +checkpoint.value +LLMClient.compact(LLM.request({ model: openai, prompt: "hello" })).pipe( + Effect.map((result) => { + result.messages + // @ts-expect-error Compaction returns replacement history, not a synthetic assistant message. + result.message + }), +) +LLM.request({ + model: openai, + providerOptions: { + // @ts-expect-error A token threshold is numeric. + contextManagement: [{ type: "compaction", compactThreshold: "100000" }], + }, +}) +for (const model of [ + Anthropic.configure().model("claude-opus-4-6"), + AmazonBedrock.configure().messages("anthropic.claude-opus-4-6-v1"), +]) { + LLM.request({ + model, + providerOptions: { + contextManagement: { + edits: [ + { type: "compact_20260112", pauseAfterCompaction: true, instructions: "Summarize without using tools" }, + ], + }, + }, + }) + LLM.request({ + model, + providerOptions: { + // @ts-expect-error A pause setting is boolean. + contextManagement: { edits: [{ type: "compact_20260112", pauseAfterCompaction: "yes" }] }, + }, + }) +} diff --git a/packages/ai/test/provider/anthropic-compaction.test.ts b/packages/ai/test/provider/anthropic-compaction.test.ts new file mode 100644 index 00000000000..04ce6e719cb --- /dev/null +++ b/packages/ai/test/provider/anthropic-compaction.test.ts @@ -0,0 +1,170 @@ +import { expect } from "bun:test" +import { Effect, Schema } from "effect" +import { LLM, LLMRequest, Message } from "../../src/index.js" +import { LLMClient } from "../../src/route/client.js" +import { Anthropic, GoogleVertexMessages } from "../../src/providers/index.js" +import { testEffect } from "../lib/effect.js" +import { dynamicResponse, fixedResponse } from "../lib/http.js" +import { sseEvents } from "../lib/sse.js" + +for (const fixture of [ + { + name: "empty iterations fall back to top-level usage", + usage: { input_tokens: 2, output_tokens: 3, cache_read_input_tokens: null, iterations: [] }, + expected: { inputTokens: 2, outputTokens: 3, totalTokens: 5, contextTokens: undefined }, + }, + { + name: "compaction-only usage has no post-compaction context size", + usage: { + input_tokens: 0, + output_tokens: 0, + iterations: [{ type: "compaction", input_tokens: 7, cache_read_input_tokens: 3, output_tokens: 2 }], + }, + expected: { inputTokens: 10, outputTokens: 2, totalTokens: 12, contextTokens: undefined }, + }, + { + name: "partially reported iterations preserve known totals", + usage: { + iterations: [ + { type: "compaction", input_tokens: 7, cache_creation_input_tokens: 2 }, + { type: "message", output_tokens: 3 }, + ], + }, + expected: { inputTokens: 9, outputTokens: 3, totalTokens: 12, contextTokens: undefined }, + }, + { + name: "missing counters remain unknown rather than zero", + usage: { iterations: [{ type: "message" }] }, + expected: { inputTokens: undefined, outputTokens: undefined, totalTokens: undefined, contextTokens: undefined }, + }, +]) { + testEffect( + fixedResponse( + sseEvents( + { type: "message_start", message: { usage: fixture.usage } }, + { type: "message_delta", delta: { stop_reason: "end_turn" } }, + { type: "message_stop" }, + ), + ), + ).effect(fixture.name, () => + Effect.gen(function* () { + const result = yield* LLMClient.generate( + LLM.request({ + model: Anthropic.configure({ apiKey: "test" }).model("claude-opus-4-6"), + prompt: "hello", + }), + ) + expect(result.usage).toMatchObject(fixture.expected) + }), + ) +} + +for (const model of [ + Anthropic.configure({ apiKey: "test" }).model("claude-opus-4-6"), + GoogleVertexMessages.configure({ accessToken: "test", project: "test" }).model("claude-opus-4-6"), +]) { + for (const summary of ["Summary of the conversation", null]) { + const block = { type: "compaction", content: summary } + testEffect( + dynamicResponse(({ request, text, respond }) => + Effect.sync(() => { + const body = JSON.parse(text) + expect(request.headers["anthropic-beta"]).toBe("existing-beta,compact-2026-01-12") + if (body.messages.length === 1) { + expect(body.context_management.edits).toEqual([ + { + type: "compact_20260112", + trigger: { type: "input_tokens", value: 50000 }, + pause_after_compaction: true, + }, + ]) + } + if (body.messages.length > 1) { + expect(body.messages[1].content).toEqual([block]) + expect(body.context_management).toBeUndefined() + } + return respond( + sseEvents( + { type: "message_start", message: { usage: { input_tokens: 50000, output_tokens: 0 } } }, + { type: "content_block_start", index: 0, content_block: { type: "compaction", content: null } }, + { type: "content_block_delta", index: 0, delta: { type: "compaction_delta", content: summary } }, + { type: "content_block_stop", index: 0 }, + { + type: "message_delta", + delta: { stop_reason: "compaction" }, + usage: { + input_tokens: 1000, + output_tokens: 5, + iterations: [ + { type: "compaction", input_tokens: 50000, output_tokens: 1000, cache_read_input_tokens: 10 }, + { type: "message", input_tokens: 1000, output_tokens: 5 }, + ], + }, + }, + { type: "message_stop" }, + ), + { headers: { "content-type": "text/event-stream" } }, + ) + }), + ), + ).effect( + `${model.provider} replays ${summary === null ? "failed" : "successful"} compaction with billing and context usage`, + () => + Effect.gen(function* () { + const request = LLM.request({ + model, + prompt: "hello", + http: { headers: { "anthropic-beta": "existing-beta" } }, + providerOptions: { + contextManagement: { + edits: [ + { + type: "compact_20260112", + trigger: { type: "input_tokens", value: 50000 }, + pauseAfterCompaction: true, + }, + ], + }, + }, + }) + const first = yield* LLMClient.generate(request) + expect(first.finishReason.raw).toBe("compaction") + expect(first.message.content).toEqual([{ type: "compaction", provider: model.provider, text: summary }]) + expect(first.text).toBe("") + expect(first.usage?.inputTokens).toBe(51010) + expect(first.usage?.outputTokens).toBe(1005) + expect(first.usage?.totalTokens).toBe(52015) + expect(first.usage?.contextTokens).toBe(1000) + const codec = Schema.fromJsonString(Message) + const message = Schema.decodeSync(codec)(Schema.encodeSync(codec)(first.message)) + yield* LLMClient.generate( + LLMRequest.update(request, { + providerOptions: {}, + messages: [...request.messages, message, Message.user("continue")], + }), + ) + }), + ) + } +} + +for (const events of [ + [{ type: "content_block_start", index: 0, content_block: { type: "compaction", content: 42 } }], + [{ type: "content_block_delta", index: 0, delta: { type: "compaction_delta", content: "no start" } }], + [ + { type: "content_block_start", index: 0, content_block: { type: "compaction", content: null } }, + { type: "message_stop" }, + ], +]) { + testEffect(fixedResponse(sseEvents(...events))).effect( + `rejects malformed compaction lifecycle: ${JSON.stringify(events)}`, + () => + Effect.gen(function* () { + const error = yield* LLMClient.generate( + LLM.request({ model: Anthropic.configure({ apiKey: "test" }).model("claude-opus-4-6"), prompt: "hello" }), + ).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidProviderOutput") + expect(error.reason.http?.status).toBe(200) + }), + ) +} diff --git a/packages/ai/test/provider/bedrock-compaction.test.ts b/packages/ai/test/provider/bedrock-compaction.test.ts new file mode 100644 index 00000000000..bbde555cd6a --- /dev/null +++ b/packages/ai/test/provider/bedrock-compaction.test.ts @@ -0,0 +1,69 @@ +import { EventStreamCodec } from "@smithy/eventstream-codec" +import { fromUtf8, toUtf8 } from "@smithy/util-utf8" +import { expect } from "bun:test" +import { Effect } from "effect" +import { LLM, LLMRequest, Message } from "../../src/index.js" +import { LLMClient } from "../../src/route/client.js" +import { AmazonBedrock } from "../../src/providers/index.js" +import { testEffect } from "../lib/effect.js" +import { dynamicResponse } from "../lib/http.js" + +const codec = new EventStreamCodec(toUtf8, fromUtf8) +const frame = (event: object) => + codec.encode({ + headers: { ":message-type": { type: "string", value: "event" }, ":event-type": { type: "string", value: "chunk" } }, + body: new TextEncoder().encode(JSON.stringify({ bytes: Buffer.from(JSON.stringify(event)).toString("base64") })), + }) +const response = Buffer.concat( + [ + { type: "message_start", message: { usage: { input_tokens: 60000 } } }, + { type: "content_block_start", index: 0, content_block: { type: "compaction", content: null } }, + { type: "content_block_delta", index: 0, delta: { type: "compaction_delta", content: "Summary" } }, + { type: "content_block_stop", index: 0 }, + { type: "content_block_start", index: 1, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 1, delta: { type: "text_delta", text: "Hello" } }, + { type: "content_block_stop", index: 1 }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 10 } }, + { type: "message_stop" }, + ].map(frame), +) + +for (const auth of [ + { apiKey: "test" }, + { credentials: { accessKeyId: "test", secretAccessKey: "test", region: "us-west-2" } }, +]) { + testEffect( + dynamicResponse(({ request, text, respond }) => + Effect.sync(() => { + expect(request.url).toBe( + "https://bedrock-runtime.us-west-2.amazonaws.com/model/anthropic.claude-opus-4-6-v1%3A0/invoke-with-response-stream", + ) + expect(request.headers.authorization).toStartWith(auth.apiKey ? "Bearer test" : "AWS4-HMAC-SHA256") + const body = JSON.parse(text) + expect(body.model).toBeUndefined() + expect(body.stream).toBeUndefined() + expect(body.anthropic_version).toBe("bedrock-2023-05-31") + expect(body.anthropic_beta).toEqual(["compact-2026-01-12"]) + expect(body.context_management.edits).toEqual([{ type: "compact_20260112" }]) + if (body.messages.length > 1) + expect(body.messages[1].content[0]).toEqual({ type: "compaction", content: "Summary" }) + return respond(response, { headers: { "content-type": "application/vnd.amazon.eventstream" } }) + }), + ), + ).effect(`Bedrock Messages compaction round trip with ${auth.apiKey ? "bearer" : "SigV4"} authentication`, () => + Effect.gen(function* () { + const model = AmazonBedrock.configure({ ...auth, region: "us-west-2" }).messages("anthropic.claude-opus-4-6-v1:0") + const request = LLM.request({ + model, + prompt: "hello", + providerOptions: { contextManagement: { edits: [{ type: "compact_20260112" }] } }, + }) + const first = yield* LLMClient.generate(request) + expect(first.text).toBe("Hello") + expect(first.message.content.map((part) => part.type)).toEqual(["compaction", "text"]) + yield* LLMClient.generate( + LLMRequest.update(request, { messages: [...request.messages, first.message, Message.user("continue")] }), + ) + }), + ) +} diff --git a/packages/ai/test/provider/compaction-websocket.test.ts b/packages/ai/test/provider/compaction-websocket.test.ts new file mode 100644 index 00000000000..34121df3dfd --- /dev/null +++ b/packages/ai/test/provider/compaction-websocket.test.ts @@ -0,0 +1,50 @@ +import { expect } from "bun:test" +import { Effect, Stream } from "effect" +import { LLM, LLMRequest, Message } from "../../src/index.js" +import { LLMClient, WebSocketTransport } from "../../src/route.js" +import { OpenAI } from "../../src/providers.js" +import { testEffect } from "../lib/effect.js" +import { fixedResponse } from "../lib/http.js" + +testEffect(fixedResponse("unexpected HTTP fallback")).effect( + "WebSocket responses preserve compaction options and replay state", + () => + Effect.gen(function* () { + const checkpoint = { type: "compaction", id: "cmp_ws", encrypted_content: "opaque" } + const sent: unknown[] = [] + const webSocket = WebSocketTransport.makeDirect({ + open: () => + Effect.succeed({ + sendText: (message) => + Effect.sync(() => { + const body = JSON.parse(message) + expect(body.context_management).toEqual([{ type: "compaction", compact_threshold: 100000 }]) + expect(body.stream).toBeUndefined() + if (sent.length) expect(body.input[1]).toEqual(checkpoint) + sent.push(body) + }), + messages: Stream.fromIterable( + [ + { type: "response.created", response: { id: "resp_ws" } }, + { type: "response.output_item.done", item: checkpoint }, + { type: "response.completed", response: { id: "resp_ws", output: [checkpoint] } }, + ].map((event) => JSON.stringify(event)), + ), + close: Effect.void, + }), + }) + const request = LLM.request({ + model: OpenAI.configure({ apiKey: "test" }).responses("gpt-5.3-codex"), + prompt: "hello", + providerOptions: { contextManagement: [{ type: "compaction", compactThreshold: 100000 }] }, + }) + const first = yield* LLMClient.generate(request, { webSocket }) + expect(first.message.content).toHaveLength(1) + expect(first.message.content[0]?.type).toBe("compaction") + yield* LLMClient.generate( + LLMRequest.update(request, { messages: [...request.messages, first.message, Message.user("continue")] }), + { webSocket }, + ) + expect(sent).toHaveLength(2) + }), +) diff --git a/packages/ai/test/provider/compaction.recorded.test.ts b/packages/ai/test/provider/compaction.recorded.test.ts new file mode 100644 index 00000000000..0bec86b42e1 --- /dev/null +++ b/packages/ai/test/provider/compaction.recorded.test.ts @@ -0,0 +1,91 @@ +import { expect } from "bun:test" +import { Effect } from "effect" +import { LLM, LLMRequest, Message } from "../../src/index.js" +import { LLMClient } from "../../src/route/client.js" +import { OpenAI, XAI, Anthropic } from "../../src/providers.js" +import { recordedTests } from "../recorded-test.js" + +const history = [ + Message.user("Remember the project codename COPPER-ORBIT-42."), + Message.assistant( + "The project codename is COPPER-ORBIT-42. " + "We reviewed the implementation and tests. ".repeat(1000), + ), +] + +for (const provider of [ + { + id: "openai", + key: "OPENAI_API_KEY", + model: OpenAI.configure({ apiKey: process.env.OPENAI_API_KEY ?? "fixture" }).responses("gpt-5.3-codex"), + }, + { + id: "xai", + key: "XAI_API_KEY", + model: XAI.configure({ apiKey: process.env.XAI_API_KEY ?? "fixture" }).responses("grok-4.6"), + }, +]) { + recordedTests({ prefix: `${provider.id}-compaction`, provider: provider.id, requires: [provider.key] }).effect( + "compacts and continues with the provider checkpoint", + () => + Effect.gen(function* () { + const request = LLM.request({ model: provider.model, messages: history, generation: { maxTokens: 1024 } }) + const compacted = yield* LLMClient.compact(request) + const result = yield* LLMClient.generate( + LLMRequest.update(request, { + messages: [ + ...compacted.messages, + Message.user("What is the project codename? Reply only with the codename."), + ], + }), + ) + expect(result.text).toContain("COPPER-ORBIT-42") + }), + 120000, + ) +} + +recordedTests({ + prefix: "anthropic-compaction", + provider: "anthropic", + requires: ["ANTHROPIC_API_KEY"], + options: { redact: { allowRequestHeaders: ["anthropic-version", "anthropic-beta"] } }, +}).effect( + "automatically compacts and continues after a pause", + () => + Effect.gen(function* () { + const model = Anthropic.configure({ apiKey: process.env.ANTHROPIC_API_KEY ?? "fixture" }).model( + "claude-sonnet-4-6", + ) + const request = LLM.request({ + model, + messages: [ + Message.user( + "Remember the project codename COPPER-ORBIT-42. " + + "The implementation and tests were reviewed. ".repeat(10000), + ), + ], + generation: { maxTokens: 4096 }, + providerOptions: { + contextManagement: { + edits: [ + { type: "compact_20260112", trigger: { type: "input_tokens", value: 50000 }, pauseAfterCompaction: true }, + ], + }, + }, + }) + const first = yield* LLMClient.generate(request) + expect(first.finishReason.raw).toBe("compaction") + expect(first.message.content.some((part) => part.type === "compaction")).toBe(true) + const result = yield* LLMClient.generate( + LLMRequest.update(request, { + messages: [ + ...request.messages, + first.message, + Message.user("What is the project codename? Reply only with the codename."), + ], + }), + ) + expect(result.text).toContain("COPPER-ORBIT-42") + }), + 120000, +) diff --git a/packages/ai/test/provider/compaction.test.ts b/packages/ai/test/provider/compaction.test.ts new file mode 100644 index 00000000000..5f1a5411986 --- /dev/null +++ b/packages/ai/test/provider/compaction.test.ts @@ -0,0 +1,136 @@ +import { expect } from "bun:test" +import { Effect, Schema } from "effect" +import { LLM, LLMRequest, Message } from "../../src/index.js" +import { LLMClient } from "../../src/route/client.js" +import { OpenAI, Azure, XAI } from "../../src/providers/index.js" +import { testEffect } from "../lib/effect.js" +import { dynamicResponse, fixedResponse } from "../lib/http.js" +import { sseEvents } from "../lib/sse.js" + +const checkpoint = { type: "compaction", id: "cmp_1", encrypted_content: "opaque" } +const response = sseEvents( + { type: "response.output_item.done", item: checkpoint }, + { + type: "response.completed", + response: { id: "resp_1", output: [checkpoint], usage: { input_tokens: 10, output_tokens: 2, total_tokens: 12 } }, + }, +) + +for (const model of [ + OpenAI.configure({ apiKey: "test" }).responses("gpt-5.3-codex"), + Azure.configure({ apiKey: "test", resourceName: "test" }).responses("deployment"), +]) { + testEffect( + dynamicResponse(({ text, respond }) => + Effect.sync(() => { + const body = JSON.parse(text) + expect(body.context_management).toEqual([{ type: "compaction", compact_threshold: 100000 }]) + expect(body.store).toBe(false) + if (body.input.length > 1) expect(body.input[1]).toEqual(checkpoint) + return respond(response, { headers: { "content-type": "text/event-stream" } }) + }), + ), + ).effect(`${model.provider} compaction survives generation, serialization, and a second request`, () => + Effect.gen(function* () { + const request = LLM.request({ + model, + prompt: "hello", + providerOptions: { contextManagement: [{ type: "compaction", compactThreshold: 100000 }] }, + }) + const first = yield* LLMClient.generate(request) + expect(first.message.content).toHaveLength(1) + expect(first.message.content[0]?.type).toBe("compaction") + expect(first.text).toBe("") + const codec = Schema.fromJsonString(Message) + const message = Schema.decodeSync(codec)(Schema.encodeSync(codec)(first.message)) + yield* LLMClient.generate( + LLMRequest.update(request, { messages: [...request.messages, message, Message.user("continue")] }), + ) + const rejected = yield* LLMClient.generate( + LLMRequest.update(request, { + model: XAI.configure({ apiKey: "test" }).responses("grok-4.6"), + providerOptions: {}, + messages: [message], + }), + ).pipe(Effect.flip) + expect(rejected.reason._tag).toBe("InvalidRequest") + }), + ) +} + +testEffect(fixedResponse(sseEvents({ type: "response.completed", response: { output: [checkpoint] } }))).effect( + "recovers compaction from the terminal output when item completion is absent", + () => + Effect.gen(function* () { + const result = yield* LLMClient.generate( + LLM.request({ model: OpenAI.configure({ apiKey: "test" }).responses("gpt-5.3-codex"), prompt: "hello" }), + ) + expect(result.message.content).toHaveLength(1) + expect(result.message.content[0]?.type).toBe("compaction") + }), +) + +testEffect( + fixedResponse(sseEvents({ type: "response.output_item.done", item: { type: "compaction", id: "cmp_bad" } })), +).effect("rejects incomplete compaction payloads without publishing a checkpoint", () => + Effect.gen(function* () { + const error = yield* LLMClient.generate( + LLM.request({ model: OpenAI.configure({ apiKey: "test" }).responses("gpt-5.3-codex"), prompt: "hello" }), + ).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidProviderOutput") + expect(error.reason.body).toContain("cmp_bad") + }), +) + +const textItem = { + type: "message", + id: "msg_after", + role: "assistant", + content: [{ type: "output_text", text: "After checkpoint" }], +} + +for (const completed of [false, true]) { + testEffect( + fixedResponse( + sseEvents( + { type: "response.output_item.added", output_index: 0, item: { type: "compaction", id: checkpoint.id } }, + ...(completed ? [{ type: "response.output_item.done", output_index: 0, item: checkpoint }] : []), + { type: "response.output_item.added", output_index: 1, item: textItem }, + { type: "response.output_text.delta", output_index: 1, item_id: textItem.id, delta: "After checkpoint" }, + { type: "response.output_item.done", output_index: 1, item: textItem }, + { type: "response.completed", response: { id: "resp_1", output: [checkpoint, textItem] } }, + ), + ), + ).effect(completed ? "keeps streamed checkpoints before later text" : "rejects order-unsafe terminal recovery", () => + Effect.gen(function* () { + const request = LLM.request({ model: OpenAI.configure({ apiKey: "test" }).responses("fixture"), prompt: "hello" }) + if (completed) { + const response = yield* LLMClient.generate(request) + expect(response.message.content.map((part) => part.type)).toEqual(["compaction", "text"]) + return + } + const error = yield* LLMClient.generate(request).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidProviderOutput") + expect(error.message).toContain("Cannot recover a compaction checkpoint") + expect(error.reason.body).toContain("response.completed") + expect(error.reason.http?.status).toBe(200) + }), + ) +} + +testEffect( + fixedResponse( + sseEvents({ + type: "response.completed", + response: { output: [{ type: "compaction", encrypted_content: "opaque" }] }, + }), + ), +).effect("rejects terminal checkpoints missing an id", () => + Effect.gen(function* () { + const error = yield* LLMClient.generate( + LLM.request({ model: OpenAI.configure({ apiKey: "test" }).responses("fixture"), prompt: "hello" }), + ).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidProviderOutput") + expect(error.message).toContain("missing its id") + }), +) diff --git a/packages/ai/test/provider/explicit-compaction.test.ts b/packages/ai/test/provider/explicit-compaction.test.ts new file mode 100644 index 00000000000..efd79108a64 --- /dev/null +++ b/packages/ai/test/provider/explicit-compaction.test.ts @@ -0,0 +1,377 @@ +import { expect } from "bun:test" +import { Effect, Schema } from "effect" +import { LLM, LLMRequest, Message } from "../../src/index.js" +import { LLMClient, Route } from "../../src/route/client.js" +import { Auth } from "../../src/route/auth.js" +import { Endpoint } from "../../src/route/endpoint.js" +import { OpenAIResponses } from "../../src/protocols/openai-responses.js" +import { OpenAI, Azure, XAI, Anthropic, AmazonBedrockMantle } from "../../src/providers/index.js" +import { testEffect } from "../lib/effect.js" +import { dynamicResponse, fixedResponse } from "../lib/http.js" +import { sseEvents } from "../lib/sse.js" + +const checkpoint = { type: "compaction", id: "cmp_1", encrypted_content: "opaque" } +const retained = { + type: "message", + role: "user", + id: "msg_1", + status: "completed", + content: [{ type: "input_text", text: "retained" }], +} +const output = [retained, checkpoint] + +testEffect( + dynamicResponse(({ request, text, respond }) => + Effect.sync(() => { + expect(request.headers["x-deployment"]).toBe("fixture") + expect(request.headers["x-override"]).toBe("request") + expect(request.headers["x-default"]).toBe("configured") + expect(request.headers.authorization).toBe("Bearer test") + expect(new URL(request.url).searchParams.get("api-version")).toBe("fixture") + expect(new URL(request.url).searchParams.get("trace")).toBe("request") + if (new URL(request.url).pathname.endsWith("/compact")) { + expect(JSON.parse(text)).toEqual({ + model: "overlaid", + input: [{ role: "user", content: [{ type: "input_text", text: "hello" }] }], + instructions: "request instructions", + previous_response_id: "resp_previous", + }) + return respond(JSON.stringify({ object: "response.compaction", output })) + } + return respond(sseEvents({ type: "response.completed", response: { id: "resp_1" } })) + }), + ), +).effect("generation and compaction share deployment headers, defaults, auth, query, and middleware", () => + Effect.gen(function* () { + const headers: string[] = [] + const middleware: string[] = [] + const route = Route.make({ + id: "compaction-headers", + provider: "openai", + protocol: OpenAIResponses.protocol, + compact: OpenAIResponses.route.compact, + transport: OpenAIResponses.httpTransport, + endpoint: Endpoint.path(({ body }) => `/${body.model}/responses`, { + baseURL: "https://example.com", + query: { "api-version": "fixture" }, + }), + auth: Auth.bearer("test"), + headers: ({ request }) => { + expect(request.providerOptions?.store).toBe(false) + headers.push(String(request.model.id)) + return { "x-deployment": "fixture", "x-override": "route" } + }, + defaults: { + headers: { "x-default": "configured", "x-override": "configured" }, + providerOptions: { store: false }, + http: { body: { instructions: "default instructions" } }, + }, + }) + const request = LLM.request({ + model: route.model({ id: "fixture" }), + prompt: "hello", + system: "system instructions", + http: { + headers: { "x-override": "request" }, + query: { trace: "request" }, + body: { + model: "overlaid", + instructions: "request instructions", + previous_response_id: "resp_previous", + store: false, + stream: true, + }, + }, + }) + const options: Parameters[1] = { + http: (request, next) => { + middleware.push(new URL(request.url).pathname) + return next(request) + }, + } + yield* LLMClient.generate(request, options) + yield* LLMClient.compact(request, options) + expect(headers).toEqual(["fixture", "fixture"]) + expect(middleware).toEqual(["/fixture/responses", "/fixture/responses/compact"]) + }), +) + +for (const model of [ + OpenAI.configure({ apiKey: "test" }).responses("fixture"), + Azure.configure({ apiKey: "test", resourceName: "test" }).responses("fixture"), + XAI.configure({ apiKey: "test" }).responses("fixture"), +]) { + const item = { + type: model.provider === "xai" ? "x_search_call" : "computer_call", + id: "hosted_1", + status: "completed", + } + testEffect( + dynamicResponse(({ request, text, respond }) => + Effect.sync(() => { + expect(new URL(request.url).pathname).toEndWith("/responses/compact") + expect(JSON.parse(text)).toEqual({ model: "fixture", input: [item], instructions: "Keep the context" }) + return respond(JSON.stringify({ object: "response.compaction", output: [checkpoint] })) + }), + ), + ).effect(`${model.provider} compacts provider-specific history without lowering generation settings`, () => + Effect.gen(function* () { + const request = LLM.request({ + model, + system: "Keep the context", + messages: [ + Message.assistant({ + type: "tool-result", + id: item.id, + name: item.type, + result: { type: "json", value: item }, + providerExecuted: true, + providerMetadata: { [model.route.providerMetadataKey ?? model.provider]: { itemId: item.id } }, + }), + ], + }) + for (const candidate of [ + LLMRequest.update(request, { + tools: [ + { name: "unsupported", description: "Generation only", inputSchema: {}, native: { unsupported: {} } }, + ], + }), + LLMRequest.update(request, { providerOptions: { contextManagement: "invalid-generation-option" } }), + ]) { + const error = yield* LLMClient.generate(candidate).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidRequest") + const response = yield* LLMClient.compact(candidate) + expect(response.messages[0]?.content[0]?.type).toBe("compaction") + } + }), + ) +} + +const retainedItems = [ + retained, + { + type: "message", + id: "msg_assistant", + role: "assistant", + status: "completed", + phase: "commentary", + content: [ + { type: "output_text", text: "First" }, + { type: "output_text", text: "Second" }, + ], + }, + { + type: "reasoning", + id: "rs_1", + summary: [ + { type: "summary_text", text: "Thinking" }, + { type: "summary_text", text: "More thinking" }, + ], + encrypted_content: "reasoning-state", + }, + { type: "reasoning", id: "rs_2", summary: [], encrypted_content: "hidden-reasoning" }, + { + type: "message", + id: "msg_media", + role: "user", + content: [ + { type: "input_image", image_url: "https://example.com/image.png" }, + { type: "input_file", filename: "report.pdf", file_data: "data:application/pdf;base64,cGRm" }, + { type: "input_file", filename: "other.pdf", file_url: "https://example.com/report.pdf" }, + ], + }, + checkpoint, +] + +testEffect( + dynamicResponse(({ request, text, respond }) => + Effect.sync(() => { + if (new URL(request.url).pathname.endsWith("/compact")) + return respond(JSON.stringify({ object: "response.compaction", output: retainedItems })) + expect(JSON.parse(text).input).toEqual(retainedItems) + return respond(sseEvents({ type: "response.completed", response: { id: "resp_1" } }), { + headers: { "content-type": "text/event-stream" }, + }) + }), + ), +).effect("retained messages, reasoning, and media are ordinary typed conversation parts", () => + Effect.gen(function* () { + const request = LLM.request({ + model: OpenAI.configure({ apiKey: "test" }).responses("gpt-5.3-codex"), + prompt: "hello", + }) + const compacted = yield* LLMClient.compact(request) + expect(compacted.messages.map((message) => message.role)).toEqual([ + "user", + "assistant", + "assistant", + "assistant", + "user", + "assistant", + ]) + expect(compacted.messages[1]?.content).toEqual([ + { type: "text", text: "First" }, + { type: "text", text: "Second" }, + ]) + expect(compacted.messages[2]?.content.map((part) => part.type)).toEqual(["reasoning", "reasoning"]) + expect(compacted.messages[4]?.content.map((part) => part.type)).toEqual(["media", "media", "media"]) + const codec = Schema.fromJsonString(Schema.Array(Message)) + const messages = Schema.decodeSync(codec)(Schema.encodeSync(codec)(compacted.messages)) + yield* LLMClient.generate(LLMRequest.update(request, { messages })) + }), +) + +for (const item of [ + { type: "unknown_provider_item", data: "do not hide in a compaction part" }, + { + type: "message", + role: "user", + content: [{ type: "input_image", image_url: "https://example.com/image.png", detail: 42 }], + }, + { type: "message", role: "user", content: [] }, + { + type: "message", + role: "assistant", + content: [{ type: "input_image", image_url: "https://example.com/image.png" }], + }, + { type: "message", role: "user", content: [{ type: "input_file", filename: "missing.pdf" }] }, +]) { + testEffect(fixedResponse(JSON.stringify({ object: "response.compaction", output: [item, checkpoint] }))).effect( + `rejects unsupported compact output: ${JSON.stringify(item)}`, + () => + Effect.gen(function* () { + const error = yield* LLMClient.compact( + LLM.request({ model: OpenAI.configure({ apiKey: "test" }).responses("gpt-5.3-codex"), prompt: "hello" }), + ).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidProviderOutput") + expect(error.reason.body).toContain(JSON.stringify(item)) + expect(error.reason.http?.status).toBe(200) + }), + ) +} + +for (const model of [ + OpenAI.configure({ apiKey: "test" }).responses("fixture"), + Azure.configure({ apiKey: "test", resourceName: "test" }).responses("fixture"), + XAI.configure({ apiKey: "test" }).responses("fixture"), +]) { + const images = [undefined, "low", "high", "auto"].map((detail) => ({ + type: "input_image", + image_url: "https://example.com/image.png", + ...(detail === undefined ? {} : { detail }), + })) + testEffect( + dynamicResponse(({ request, text, respond }) => + Effect.sync(() => { + if (new URL(request.url).pathname.endsWith("/compact")) + return respond( + JSON.stringify({ + object: "response.compaction", + output: [{ type: "message", role: "user", content: images }, checkpoint], + }), + ) + expect(JSON.parse(text).input[0].content).toEqual(images) + return respond(sseEvents({ type: "response.completed", response: { id: "resp_1" } })) + }), + ), + ).effect(`${model.provider} preserves retained image detail through serialization and replay`, () => + Effect.gen(function* () { + const request = LLM.request({ model, prompt: "hello" }) + const compacted = yield* LLMClient.compact(request) + const codec = Schema.fromJsonString(Schema.Array(Message)) + const messages = Schema.decodeSync(codec)(Schema.encodeSync(codec)(compacted.messages)) + yield* LLMClient.generate(LLMRequest.update(request, { messages })) + }), + ) +} + +testEffect(fixedResponse("must not execute")).effect("xAI rejects automatic compaction options", () => + Effect.gen(function* () { + const request = LLMRequest.update( + LLM.request({ model: XAI.configure({ apiKey: "test" }).responses("grok-4.6"), prompt: "hello" }), + { providerOptions: { contextManagement: [{ type: "compaction" }] } }, + ) + const error = yield* LLMClient.generate(request).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidRequest") + expect(error.message).toContain("LLMClient.compact") + }), +) + +for (const model of [ + OpenAI.configure({ apiKey: "test" }).responses("gpt-5.3-codex"), + Azure.configure({ apiKey: "test", resourceName: "test" }).responses("deployment"), + XAI.configure({ apiKey: "test" }).responses("grok-4.6"), +]) { + testEffect( + dynamicResponse(({ request, text, respond }) => + Effect.sync(() => { + const body = JSON.parse(text) + expect(request.method).toBe("POST") + expect(request.headers[model.provider === "azure" ? "api-key" : "authorization"]).toBe( + model.provider === "azure" ? "test" : "Bearer test", + ) + if (new URL(request.url).pathname.endsWith("/responses/compact")) { + expect(body).toEqual({ + model: model.id, + input: [{ role: "user", content: [{ type: "input_text", text: "original" }] }], + instructions: "system", + }) + return respond( + JSON.stringify({ + object: "response.compaction", + output, + usage: { input_tokens: 1000, output_tokens: 10, total_tokens: 1010 }, + }), + { headers: { "content-type": "application/json" } }, + ) + } + expect(new URL(request.url).pathname.endsWith("/responses")).toBe(true) + expect(body.input).toEqual([...output, { role: "user", content: [{ type: "input_text", text: "continue" }] }]) + return respond(sseEvents({ type: "response.completed", response: { id: "resp_1", output: [] } }), { + headers: { "content-type": "text/event-stream" }, + }) + }), + ), + ).effect(`${model.provider} explicitly compacts and replays the entire canonical window`, () => + Effect.gen(function* () { + const request = LLM.request({ model, prompt: "original", system: "system", http: { body: { store: false } } }) + const compacted = yield* LLMClient.compact(request) + expect(compacted.usage?.totalTokens).toBe(1010) + expect(compacted.messages.map((message) => message.role)).toEqual(["user", "assistant"]) + expect(compacted.messages[0]?.content).toEqual([{ type: "text", text: "retained" }]) + expect(compacted.messages[1]?.content).toEqual([ + { type: "compaction", provider: model.provider, id: "cmp_1", encrypted: "opaque" }, + ]) + const codec = Schema.fromJsonString(Schema.Array(Message)) + const messages = Schema.decodeSync(codec)(Schema.encodeSync(codec)(compacted.messages)) + yield* LLMClient.generate(LLMRequest.update(request, { messages: [...messages, Message.user("continue")] })) + }), + ) +} + +for (const model of [ + Anthropic.configure({ apiKey: "test" }).model("claude-opus-4-6"), + AmazonBedrockMantle.configure({ apiKey: "test" }).responses("model"), +]) { + testEffect(fixedResponse("must not execute")).effect( + `${model.route.id} does not inherit an unsupported compact endpoint`, + () => + Effect.gen(function* () { + const error = yield* LLMClient.compact(LLM.request({ model, prompt: "hello" })).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidRequest") + }), + ) +} + +testEffect( + fixedResponse(JSON.stringify({ object: "response.compaction", output: [retained], debug: "original payload" })), +).effect("invalid explicit compaction preserves the original response and HTTP context", () => + Effect.gen(function* () { + const error = yield* LLMClient.compact( + LLM.request({ model: OpenAI.configure({ apiKey: "test" }).responses("gpt-5.3-codex"), prompt: "hello" }), + ).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidProviderOutput") + expect(error.reason.body).toContain("original payload") + expect(error.reason.http?.status).toBe(200) + }), +) diff --git a/packages/core/src/aisdk.ts b/packages/core/src/aisdk.ts index 841ca6a5172..efd8dfcdd7e 100644 --- a/packages/core/src/aisdk.ts +++ b/packages/core/src/aisdk.ts @@ -327,7 +327,12 @@ function modelFromLanguage(info: Info, language: LanguageModelV3) { }, body: { schema: Schema.Unknown, - from: (request) => Effect.succeed(callOptions(request, packageName, info.modelID ?? info.id, optionKey)), + from: (request) => + Effect.try({ + try: () => callOptions(request, packageName, info.modelID ?? info.id, optionKey), + catch: (cause) => + cause instanceof AIError ? cause : ProviderShared.invalidRequest("Invalid AI SDK request", cause), + }), }, with: () => route, model: (input) => @@ -512,6 +517,8 @@ function userPart(part: ContentPart): UserContent { function assistantPart(part: ContentPart): AssistantContent { switch (part.type) { + case "compaction": + throw ProviderShared.invalidRequest("AI SDK routes cannot replay native provider compaction state") case "text": return [{ type: "text", text: part.text, providerOptions: metadataProviderOptions(part.providerMetadata) }] case "media": diff --git a/packages/core/test/aisdk.test.ts b/packages/core/test/aisdk.test.ts index 436a56533ae..55a5a38eb87 100644 --- a/packages/core/test/aisdk.test.ts +++ b/packages/core/test/aisdk.test.ts @@ -10,6 +10,8 @@ import { Provider } from "@opencode-ai/core/provider" import { LLM, AIError, + CompactionPart, + ProviderID, HttpContext, LLMEvent, Message, @@ -68,6 +70,31 @@ const client = LLMClient.layer.pipe( ), ) +it.effect("rejects native provider compaction rather than silently dropping replay state", () => + Effect.gen(function* () { + const aisdk = yield* AISDK.Service + yield* aisdk.hook.sdk((event) => { + event.sdk = { languageModel: () => streamModel([]) } + }) + const resolved = yield* aisdk.model(model("@ai-sdk/openai")) + const error = yield* compileRequest( + LLM.request({ + model: resolved, + messages: [ + Message.assistant( + CompactionPart.make({ + provider: ProviderID.make("test-provider"), + encrypted: "opaque", + }), + ), + ], + }), + ).pipe(Effect.flip) + expect(error.reason._tag).toBe("InvalidRequest") + expect(error.message).toContain("cannot replay") + }), +) + it.effect("keys language models by package and flattened overlays", () => Effect.gen(function* () { const aisdk = yield* AISDK.Service