diff --git a/packages/opencode/src/provider/transform.ts b/packages/opencode/src/provider/transform.ts index b388297aee80..4769a8d27d75 100644 --- a/packages/opencode/src/provider/transform.ts +++ b/packages/opencode/src/provider/transform.ts @@ -97,6 +97,26 @@ function sdkKey(npm: string): string | undefined { return undefined } +function isMistral(model: Provider.Model) { + const id = model.api.id.toLowerCase() + return ( + model.providerID === "mistral" || + ["mistral", "devstral", "codestral", "pixtral", "mixtral"].some((family) => id.includes(family)) + ) +} + +function mistralBridge( + model: Provider.Model, + previous: ModelMessage | undefined, + next: ModelMessage | undefined, +): ModelMessage | undefined { + if (!isMistral(model) || previous?.role !== "tool" || next?.role !== "user") return + return { + role: "assistant", + content: [{ type: "text", text: "Done." }], + } +} + // TODO: fix this stupid inefficient dogshit function function normalizeMessages( msgs: ModelMessage[], @@ -251,11 +271,7 @@ function normalizeMessages( }) } - const modelID = model.api.id.toLowerCase() - if ( - model.providerID === "mistral" || - ["mistral", "devstral", "codestral", "pixtral", "mixtral"].some((family) => modelID.includes(family)) - ) { + if (isMistral(model)) { const scrub = (id: string) => { return id .replace(/[^a-zA-Z0-9]/g, "") // Remove non-alphanumeric characters @@ -286,17 +302,8 @@ function normalizeMessages( result.push(msg) // Fix message sequence: tool messages cannot be followed by user messages - if (msg.role === "tool" && nextMsg?.role === "user") { - result.push({ - role: "assistant", - content: [ - { - type: "text", - text: "Done.", - }, - ], - }) - } + const bridge = mistralBridge(model, msg, nextMsg) + if (bridge) result.push(bridge) } return result } @@ -463,9 +470,18 @@ function mapProviderOptions( }) } -export function message(msgs: ModelMessage[], model: Provider.Model, options: Record) { +export type MessageSuffix = readonly ModelMessage[] + +export function message( + msgs: ModelMessage[], + model: Provider.Model, + options: Record, + suffix?: MessageSuffix, +) { msgs = unsupportedParts(msgs, model) msgs = normalizeMessages(msgs, model, options) + const tail = suffix?.length ? normalizeMessages(unsupportedParts([...suffix], model), model, options) : [] + const bridge = mistralBridge(model, msgs.at(-1), tail[0]) const usesAnthropicAutomaticCaching = options.cacheControl !== undefined && (model.api.npm === "@ai-sdk/anthropic" || model.api.npm === "@ai-sdk/google-vertex/anthropic") @@ -483,6 +499,7 @@ export function message(msgs: ModelMessage[], model: Provider.Model, options: Re ) { msgs = applyCaching(msgs, model) } + if (bridge || tail.length) msgs = [...msgs, ...(bridge ? [bridge] : []), ...tail] // Remap providerOptions keys from stored providerID to expected SDK key const key = sdkKey(model.api.npm) diff --git a/packages/opencode/src/session/llm.ts b/packages/opencode/src/session/llm.ts index a99f8acff20c..4fe1c6c37dab 100644 --- a/packages/opencode/src/session/llm.ts +++ b/packages/opencode/src/session/llm.ts @@ -41,6 +41,7 @@ export type StreamInput = { permission?: PermissionV1.Ruleset system: string[] messages: ModelMessage[] + messageSuffix?: readonly ModelMessage[] small?: boolean tools: Record retries?: number @@ -230,6 +231,7 @@ const live: Layer.Layer< auth: info, llmClient, messages: prepared.messages, + messageSuffix: prepared.messageSuffix, tools: prepared.tools, toolChoice: input.toolChoice, temperature: prepared.params.temperature, @@ -334,6 +336,7 @@ const live: Layer.Layer< args.params.prompt, input.model, prepared.messageTransformOptions, + prepared.messageSuffix, ) } return args.params diff --git a/packages/opencode/src/session/llm/native-runtime.ts b/packages/opencode/src/session/llm/native-runtime.ts index bac385c59137..db440851eb34 100644 --- a/packages/opencode/src/session/llm/native-runtime.ts +++ b/packages/opencode/src/session/llm/native-runtime.ts @@ -32,6 +32,7 @@ type StreamInput = { readonly auth: Auth.Info | undefined readonly llmClient: LLMClientShape readonly messages: ModelMessage[] + readonly messageSuffix?: readonly ModelMessage[] readonly tools: Record readonly toolChoice?: "auto" | "required" | "none" readonly temperature?: number @@ -91,7 +92,7 @@ export function stream(input: StreamInput): StreamResult { model: input.model, apiKey: current.apiKey, baseURL: current.baseURL, - messages: ProviderTransform.message(input.messages, input.model, input.providerOptions ?? {}), + messages: ProviderTransform.message(input.messages, input.model, input.providerOptions ?? {}, input.messageSuffix), toolChoice: input.toolChoice, temperature: input.temperature, topP: input.topP, diff --git a/packages/opencode/src/session/llm/request.ts b/packages/opencode/src/session/llm/request.ts index 4f93411107df..f4d7052e23af 100644 --- a/packages/opencode/src/session/llm/request.ts +++ b/packages/opencode/src/session/llm/request.ts @@ -26,6 +26,7 @@ type PrepareInput = { readonly permission?: PermissionV1.Ruleset readonly system: string[] readonly messages: ModelMessage[] + readonly messageSuffix?: readonly ModelMessage[] readonly small?: boolean readonly tools: Record readonly provider: Provider.Info @@ -38,6 +39,7 @@ type PrepareInput = { export type Prepared = { readonly system: string[] readonly messages: ModelMessage[] + readonly messageSuffix?: readonly ModelMessage[] readonly tools: Record readonly params: { readonly temperature?: number @@ -181,6 +183,7 @@ export const prepare = Effect.fn("LLMRequestPrep.prepare")(function* (input: Pre return { system, messages, + messageSuffix: input.messageSuffix, tools: Object.fromEntries(Object.entries(tools).toSorted(([a], [b]) => a.localeCompare(b))), params, messageTransformOptions: options, diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index 9b3f2c46f405..8e203609804a 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -128,13 +128,40 @@ function providerMeta(metadata: Record | undefined) { return Object.keys(rest).length > 0 ? rest : undefined } -export const toModelMessagesEffect = Effect.fnUntraced(function* ( +export function appendedTailCount(before: readonly string[], after: readonly WithParts[]): number { + const anchor = before.at(-1) + if (!anchor) return 0 + if (new Set(before).size !== before.length) return 0 + + const ids = after.map((message) => message.info.id) + if (new Set(ids).size !== ids.length) return 0 + const anchorIndexes = ids.flatMap((id, index) => (id === anchor ? [index] : [])) + if (anchorIndexes.length !== 1) return 0 + + const anchorIndex = anchorIndexes[0]! + const rank = new Map(before.map((id, index) => [id, index])) + const surviving = ids.slice(0, anchorIndex + 1).flatMap((id) => (rank.has(id) ? [rank.get(id)!] : [])) + if (surviving.some((index, offset) => offset > 0 && index <= surviving[offset - 1]!)) return 0 + + const prior = new Set(before) + const suffix = ids.slice(anchorIndex + 1) + if (suffix.some((id) => prior.has(id))) return 0 + return suffix.length +} + +export const toModelMessagesSplitEffect = Effect.fnUntraced(function* ( input: WithParts[], model: Provider.Model, - options?: { stripMedia?: boolean; toolOutputMaxChars?: number }, + options?: { stripMedia?: boolean; toolOutputMaxChars?: number; requestOnlyTailCount?: number }, ) { const result: UIMessage[] = [] const toolNames = new Set() + const requestOnlyStart = input.length - Math.min(Math.max(options?.requestOnlyTailCount ?? 0, 0), input.length) + const requestOnly = new WeakSet() + const emit = (message: UIMessage, sourceIndex: number) => { + result.push(message) + if (sourceIndex >= requestOnlyStart) requestOnly.add(message) + } // Track media from tool results that need to be injected as user messages // for providers that don't support that media type in tool results. // @@ -192,7 +219,7 @@ export const toModelMessagesEffect = Effect.fnUntraced(function* ( return { type: "json", value: output as never } } - for (const msg of input) { + for (const [sourceIndex, msg] of input.entries()) { if (msg.parts.length === 0) continue if (msg.info.role === "user") { @@ -238,7 +265,7 @@ export const toModelMessagesEffect = Effect.fnUntraced(function* ( }) } } - if (userMessage.parts.length > 0) result.push(userMessage) + if (userMessage.parts.length > 0) emit(userMessage, sourceIndex) } if (msg.info.role === "assistant") { @@ -376,26 +403,29 @@ export const toModelMessagesEffect = Effect.fnUntraced(function* ( } } if (assistantMessage.parts.length > 0) { - result.push(assistantMessage) + emit(assistantMessage, sourceIndex) // Inject pending media as a user message for providers that don't support // media (images, PDFs) in tool results if (media.length > 0) { - result.push({ - id: MessageID.ascending(), - role: "user", - parts: [ - { - type: "text" as const, - text: SYNTHETIC_ATTACHMENT_PROMPT, - }, - ...media.map((attachment) => ({ - type: "file" as const, - url: attachment.url, - mediaType: attachment.mime, - filename: attachment.filename, - })), - ], - }) + emit( + { + id: MessageID.ascending(), + role: "user", + parts: [ + { + type: "text" as const, + text: SYNTHETIC_ATTACHMENT_PROMPT, + }, + ...media.map((attachment) => ({ + type: "file" as const, + url: attachment.url, + mediaType: attachment.mime, + filename: attachment.filename, + })), + ], + }, + sourceIndex, + ) } } } @@ -403,15 +433,46 @@ export const toModelMessagesEffect = Effect.fnUntraced(function* ( const tools = Object.fromEntries(Array.from(toolNames).map((toolName) => [toolName, { toModelOutput }])) - return yield* Effect.promise(() => - convertToModelMessages( - result.filter((msg) => msg.parts.some((part) => part.type !== "step-start")), - { + const convert = (messages: UIMessage[]) => + Effect.promise(() => + convertToModelMessages(messages, { //@ts-expect-error (convertToModelMessages expects a ToolSet but only actually needs tools[name]?.toModelOutput) tools, - }, - ), - ) + }), + ) + + const filtered = result.filter((msg) => msg.parts.some((part) => part.type !== "step-start")) + const split = filtered.findIndex((message) => requestOnly.has(message)) + if (split < 0 || !filtered.slice(split).every((message) => requestOnly.has(message))) { + return { messages: yield* convert(filtered), tail: [] } + } + + // The AI SDK lowers ModelMessage into a provider prompt before middleware runs. The suffix joins + // inside that middleware, so only text (plus the step boundary that converts into text messages) + // is already in the same shape on both sides of that lowering. Keep richer plugin appends in the + // main array rather than risk changing files, tool results, or other structured content in flight. + const tail = filtered.slice(split) + if ( + !tail.every((message) => + message.parts.every((part) => part.type === "step-start" || (part.type === "text" && part.text !== "")), + ) + ) { + return { messages: yield* convert(filtered), tail: [] } + } + + return { + messages: yield* convert(filtered.slice(0, split)), + tail: yield* convert(tail), + } +}) + +export const toModelMessagesEffect = Effect.fnUntraced(function* ( + input: WithParts[], + model: Provider.Model, + options?: { stripMedia?: boolean; toolOutputMaxChars?: number }, +) { + const result = yield* toModelMessagesSplitEffect(input, model, options) + return result.messages }) export function toModelMessages( diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 22b1d7d99a2a..99ff507b79ee 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -1252,14 +1252,16 @@ const layer = Layer.effect( if (step === 1) yield* summary.summarize({ sessionID, messageID: lastUser.id }).pipe(Effect.ignore, Effect.forkIn(scope)) + const beforeTransform = msgs.map((message) => message.info.id) yield* plugin.trigger("experimental.chat.messages.transform", {}, { messages: msgs }) + const requestOnlyTailCount = MessageV2.appendedTailCount(beforeTransform, msgs) const [skills, env, instructions, mcpInstructions, modelMsgs] = yield* Effect.all([ sys.skills(agent), sys.environment(model), instruction.system().pipe(Effect.orDie), sys.mcp(agent, session.permission), - MessageV2.toModelMessagesEffect(msgs, model), + MessageV2.toModelMessagesSplitEffect(msgs, model, { requestOnlyTailCount }), ]) const system = [ ...env, @@ -1276,9 +1278,17 @@ const layer = Layer.effect( sessionID, parentSessionID: session.parentID, system, - messages: [ - ...modelMsgs, - ...(isLastStep ? [{ role: "assistant" as const, content: MAX_STEPS_PROMPT }] : []), + messages: modelMsgs.messages, + messageSuffix: [ + ...modelMsgs.tail, + ...(isLastStep + ? [ + { + role: "assistant" as const, + content: [{ type: "text" as const, text: MAX_STEPS_PROMPT }], + }, + ] + : []), ], tools, model, diff --git a/packages/opencode/test/session/cache-breakpoints-wire.test.ts b/packages/opencode/test/session/cache-breakpoints-wire.test.ts new file mode 100644 index 000000000000..6af6242f547a --- /dev/null +++ b/packages/opencode/test/session/cache-breakpoints-wire.test.ts @@ -0,0 +1,519 @@ +import { describe, expect, test } from "bun:test" +import { createAnthropic } from "@ai-sdk/anthropic" +import { jsonSchema, streamText, tool, wrapLanguageModel, type ModelMessage } from "ai" +import { ProviderTransform } from "@/provider/transform" +import type { Provider } from "@/provider/provider" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { ModelV2 } from "@opencode-ai/core/model" +import { MessageV2 } from "@/session/message-v2" +import { SessionV1 } from "@opencode-ai/core/v1/session" +import { SessionID, MessageID, PartID } from "@/session/schema" +import { Effect } from "effect" + +const sessionID = SessionID.make("session") + +function makeModel(input: { providerID: string; apiID: string; npm: string; id?: string }): Provider.Model { + return { + id: ModelV2.ID.make(input.id ?? input.apiID), + providerID: ProviderV2.ID.make(input.providerID), + api: { id: input.apiID, url: "https://example.com", npm: input.npm }, + name: "Probe", + capabilities: { + temperature: true, + reasoning: false, + attachment: false, + toolcall: true, + input: { text: true, audio: false, image: false, video: false, pdf: false }, + output: { text: true, audio: false, image: false, video: false, pdf: false }, + interleaved: false, + }, + cost: { input: 0, output: 0, cache: { read: 0, write: 0 } }, + limit: { context: 100000, input: 0, output: 8000 }, + status: "active", + options: {}, + headers: {}, + release_date: "2026-01-01", + } as unknown as Provider.Model +} + +const anthropicModel = () => makeModel({ providerID: "anthropic", apiID: "claude-opus-5", npm: "@ai-sdk/anthropic" }) + +const deepseekModel = () => + makeModel({ providerID: "deepseek", apiID: "deepseek-reasoner", npm: "@ai-sdk/openai-compatible" }) + +const mistralModel = () => makeModel({ providerID: "mistral", apiID: "mistral-large-latest", npm: "@ai-sdk/mistral" }) + +const ineligibleModel = () => makeModel({ providerID: "openai", apiID: "gpt-5", npm: "@ai-sdk/openai" }) + +function sse(events: Record[]) { + return events.map((event) => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join("") +} + +function textTurn(text: string) { + return sse([ + { + type: "message_start", + message: { + id: "msg_probe", + type: "message", + role: "assistant", + model: "claude-opus-5", + content: [], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 10, output_tokens: 1, cache_creation_input_tokens: 0, cache_read_input_tokens: 0 }, + }, + }, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text } }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ]) +} + +function toolTurn(id: string, input: { x: number }) { + return sse([ + { + type: "message_start", + message: { + id: `msg_${id}`, + type: "message", + role: "assistant", + model: "claude-opus-5", + content: [], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 10, output_tokens: 1, cache_creation_input_tokens: 0, cache_read_input_tokens: 0 }, + }, + }, + { type: "content_block_start", index: 0, content_block: { type: "tool_use", id, name: "probe_tool", input: {} } }, + { + type: "content_block_delta", + index: 0, + delta: { type: "input_json_delta", partial_json: JSON.stringify(input) }, + }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "tool_use", stop_sequence: null }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ]) +} + +const probeTool = tool({ + description: "probe", + inputSchema: jsonSchema({ type: "object", properties: { x: { type: "number" } } }), + execute: async () => "tool output for the probe", +}) + +type WireBody = { messages: Array<{ role: string; content: Array> }> } + +async function request( + messages: ModelMessage[], + options?: { + suffix?: ProviderTransform.MessageSuffix + model?: Provider.Model + scripted?: string + }, +): Promise<{ body: WireBody; produced: ModelMessage[]; fetches: number }> { + const model = options?.model ?? anthropicModel() + const captured: { body?: string } = {} + let fetches = 0 + const fetch = (async (_url: unknown, init: RequestInit | undefined) => { + fetches++ + captured.body = typeof init?.body === "string" ? init.body : JSON.stringify(init?.body) + return new Response(options?.scripted ?? textTurn("done"), { + status: 200, + headers: { "content-type": "text/event-stream" }, + }) + }) as typeof globalThis.fetch + const anthropic = createAnthropic({ apiKey: "probe", fetch }) + const wrapped = wrapLanguageModel({ + model: anthropic("claude-opus-5") as never, + middleware: [ + { + specificationVersion: "v3" as const, + async transformParams(args) { + if (args.type === "stream") { + // @ts-expect-error The middleware prompt is the runtime shape transformed in production. + args.params.prompt = ProviderTransform.message(args.params.prompt, model, {}, options?.suffix) + } + return args.params + }, + }, + ], + }) + const result = streamText({ + model: wrapped as never, + messages, + tools: { probe_tool: probeTool }, + maxRetries: 0, + }) + for await (const _ of result.fullStream) { + } + const response = await result.response + return { + body: JSON.parse(captured.body!) as WireBody, + produced: response.messages as ModelMessage[], + fetches, + } +} + +async function wire(messages: ModelMessage[], options?: Parameters[1]): Promise { + return (await request(messages, options)).body +} + +function textPart(messageID: string, partID: string, text: string) { + return { + id: PartID.make(partID), + sessionID, + messageID: MessageID.make(messageID), + type: "text", + text, + } +} + +function userSource(id: string, text: string): SessionV1.WithParts { + return { + info: { + id, + sessionID, + role: "user", + time: { created: 0 }, + agent: "user", + model: { providerID: ProviderV2.ID.make("anthropic"), modelID: ModelV2.ID.make("claude-opus-5") }, + tools: {}, + mode: "", + } as unknown as SessionV1.User, + parts: [textPart(id, `prt_${id}`, text)] as SessionV1.Part[], + } +} + +function assistantSource(id: string, parentID: string, text: string): SessionV1.WithParts { + return { + info: { + id, + sessionID, + role: "assistant", + time: { created: 0 }, + parentID, + modelID: "claude-opus-5", + providerID: "anthropic", + mode: "", + agent: "agent", + path: { cwd: "/", root: "/" }, + cost: 0, + } as unknown as SessionV1.Assistant, + parts: [textPart(id, `prt_${id}`, text)] as SessionV1.Part[], + } +} + +function conversation(): SessionV1.WithParts[] { + return [ + userSource("msg_durable_user", "durable question"), + assistantSource("msg_durable_assistant", "msg_durable_user", "durable answer"), + userSource("msg_request_status", "request status"), + userSource("msg_request_policy", "request policy"), + ] +} + +const separatedConversation = (model: Provider.Model = anthropicModel()) => + Effect.runPromise(MessageV2.toModelMessagesSplitEffect(conversation(), model, { requestOnlyTailCount: 2 })) + +function breakpoints(body: WireBody): string[] { + return body.messages.flatMap((message) => + message.content.flatMap((block) => (block["cache_control"] ? [String(block["text"] ?? block["type"])] : [])), + ) +} + +describe("cache breakpoints on the provider wire", () => { + test("breakpoints land on durable content while appended messages carry none", async () => { + const converted = await separatedConversation() + const body = await wire(converted.messages, { suffix: converted.tail }) + const text = body.messages.flatMap((message) => message.content.map((block) => block["text"])) + + expect(text).toEqual(["durable question", "durable answer", "request status", "request policy"]) + expect(breakpoints(body)).toEqual(["durable question", "durable answer"]) + }) + + test("flattening appended messages back into the conversation reproduces the defect", async () => { + const converted = await separatedConversation() + const body = await wire([...converted.messages, ...converted.tail]) + + expect(breakpoints(body)).toEqual(["request status", "request policy"]) + }) + + test("the negative control produces a different provider request", async () => { + const converted = await separatedConversation() + const separated = await wire(converted.messages, { suffix: converted.tail }) + const flattened = await wire([...converted.messages, ...converted.tail]) + + expect(JSON.stringify(separated)).not.toBe(JSON.stringify(flattened)) + }) + + test("plain appended messages are structurally excluded from selection", async () => { + const messages = await MessageV2.toModelMessages(conversation().slice(0, 2), anthropicModel()) + const body = await wire(messages, { + suffix: [ + { role: "user", content: [{ type: "text", text: "plain suffix one" }] }, + { role: "user", content: [{ type: "text", text: "plain suffix two" }] }, + ], + }) + + expect(breakpoints(body)).toEqual(["durable question", "durable answer"]) + }) + + test("an empty suffix is byte-identical to no suffix on the provider wire", async () => { + const messages = await MessageV2.toModelMessages(conversation().slice(0, 2), anthropicModel()) + const without = await wire(messages) + const empty = await wire(await MessageV2.toModelMessages(conversation().slice(0, 2), anthropicModel()), { + suffix: [], + }) + + expect(JSON.stringify(empty)).toBe(JSON.stringify(without)) + }) + + test("conversion returns the conversation and appended messages separately", async () => { + const converted = await separatedConversation() + expect(converted.messages).toEqual([ + { role: "user", content: [{ type: "text", text: "durable question" }] }, + { role: "assistant", content: [{ type: "text", text: "durable answer" }] }, + ]) + expect(converted.tail).toEqual([ + { role: "user", content: [{ type: "text", text: "request status" }] }, + { role: "user", content: [{ type: "text", text: "request policy" }] }, + ]) + }) + + test("an ineligible model receives no breakpoints", async () => { + const converted = await separatedConversation(ineligibleModel()) + const body = await wire(converted.messages, { suffix: converted.tail, model: ineligibleModel() }) + + expect(breakpoints(body)).toEqual([]) + }) + + test("breakpoints stay within the provider's four-marker cap", async () => { + const converted = await separatedConversation() + const body = await wire(converted.messages, { suffix: converted.tail }) + + expect(breakpoints(body).length).toBeLessThanOrEqual(4) + }) + + test("every request in the OpenCode-shaped outer loop selects the last two durable messages", async () => { + const durable: ModelMessage[] = [ + { role: "system", content: "stable system" }, + { role: "user", content: "u1: first user turn" }, + { role: "assistant", content: "a1: first assistant turn" }, + { role: "user", content: "u2: the real current prompt" }, + ] + const scripts = [toolTurn("toolu_1", { x: 1 }), toolTurn("toolu_2", { x: 2 }), textTurn("done")] + const expected = [ + ["a1: first assistant turn", "u2: the real current prompt"], + ["tool_use", "tool_result"], + ["tool_use", "tool_result"], + ] + + for (const [step, scripted] of scripts.entries()) { + const suffix: ModelMessage[] = [ + { role: "user", content: [{ type: "text", text: "request status" }] }, + { role: "user", content: [{ type: "text", text: "request policy" }] }, + ] + const result = await request(durable, { suffix, scripted }) + const blocks = result.body.messages.flatMap((message) => message.content) + + expect(result.fetches).toBe(1) + expect(breakpoints(result.body)).toEqual(expected[step]) + expect(blocks.filter((block) => block["text"] === "request status")).toHaveLength(1) + expect(blocks.filter((block) => block["text"] === "request policy")).toHaveLength(1) + expect( + blocks + .filter((block) => ["request status", "request policy"].includes(String(block["text"]))) + .every((block) => block["cache_control"] === undefined), + ).toBe(true) + durable.push(...result.produced) + } + }) +}) + +type NormalizationFixture = { messages: ModelMessage[]; tail: ModelMessage[] } + +function textFixture(): NormalizationFixture { + return { + messages: [ + { role: "system", content: "stable system" }, + { role: "user", content: [{ type: "text", text: "durable user" }] }, + { role: "assistant", content: [{ type: "text", text: "durable assistant" }] }, + { role: "user", content: [{ type: "text", text: "current prompt" }] }, + ], + tail: [ + { role: "assistant", content: [{ type: "text", text: "assistant notice" }] }, + { role: "user", content: [{ type: "text", text: "user notice" }] }, + ], + } +} + +function mistralFixture(): NormalizationFixture { + return { + messages: [ + { role: "system", content: "stable system" }, + { role: "assistant", content: "durable assistant" }, + { + role: "tool", + content: [ + { + type: "tool-result", + toolCallId: "call_main", + toolName: "probe", + output: { type: "text", value: "durable tool output" }, + }, + ], + }, + ], + tail: [ + { role: "user", content: [{ type: "text", text: "request-only seam notice" }] }, + { + role: "tool", + content: [ + { + type: "tool-result", + toolCallId: "call_suffix", + toolName: "probe", + output: { type: "text", value: "request-only tool output" }, + }, + ], + }, + { role: "user", content: [{ type: "text", text: "request-only user notice" }] }, + ], + } +} + +function expectNormalizationIdentity(model: Provider.Model, fixture: () => NormalizationFixture) { + const combined = fixture() + const previous = ProviderTransform.message([...combined.messages, ...combined.tail], model, { cacheControl: true }) + const separated = fixture() + const redesigned = ProviderTransform.message(separated.messages, model, { cacheControl: true }, separated.tail) + expect(JSON.stringify(redesigned)).toBe(JSON.stringify(previous)) + return redesigned +} + +function expectNoSuffixIdentity(model: Provider.Model, fixture: () => NormalizationFixture) { + const without = ProviderTransform.message(fixture().messages, model, {}) + const empty = ProviderTransform.message(fixture().messages, model, {}, []) + expect(JSON.stringify(empty)).toBe(JSON.stringify(without)) +} + +describe("appended message normalization", () => { + test("Anthropic normalization is byte-identical with appended messages", () => { + expectNormalizationIdentity(anthropicModel(), textFixture) + }) + + test("DeepSeek normalization is byte-identical with appended messages", () => { + const output = expectNormalizationIdentity(deepseekModel(), textFixture) + const assistant = output.filter((message) => message.role === "assistant") + expect(assistant).toHaveLength(2) + expect( + assistant.every( + (message) => Array.isArray(message.content) && message.content.some((part) => part.type === "reasoning"), + ), + ).toBe(true) + }) + + test("Mistral normalization is byte-identical with appended messages", () => { + const output = expectNormalizationIdentity(mistralModel(), mistralFixture) + const bridges = output.filter( + (message) => + message.role === "assistant" && + Array.isArray(message.content) && + message.content.some((part) => part.type === "text" && part.text === "Done."), + ) + expect(bridges).toHaveLength(2) + }) + + test("Anthropic is byte-identical with no appended messages", () => + expectNoSuffixIdentity(anthropicModel(), textFixture)) + + test("DeepSeek is byte-identical with no appended messages", () => + expectNoSuffixIdentity(deepseekModel(), textFixture)) + + test("Mistral is byte-identical with no appended messages", () => + expectNoSuffixIdentity(mistralModel(), mistralFixture)) + + test("appended messages receive provider-key remapping and itemId stripping", () => { + const model = makeModel({ providerID: "custom-openai", apiID: "gpt-5", npm: "@ai-sdk/openai" }) + const output = ProviderTransform.message([{ role: "user", content: "durable" }], model, { store: false }, [ + { + role: "user", + content: [{ type: "text", text: "request-only" }], + providerOptions: { "custom-openai": { itemId: "item_suffix", keep: "yes" } }, + }, + ]) + + expect(output.at(-1)?.providerOptions).toEqual({ openai: { keep: "yes" } }) + }) +}) + +function marked(messages: ModelMessage[]): number { + return messages.filter((message) => { + if (message.providerOptions?.["anthropic"]) return true + if (!Array.isArray(message.content)) return false + return message.content.some( + (part) => + typeof part === "object" && part !== null && "providerOptions" in part && part.providerOptions?.["anthropic"], + ) + }).length +} + +describe("cache selection invariants", () => { + test("normalizes before selection so a deleted empty message cannot consume a breakpoint", () => { + const output = ProviderTransform.message( + [ + { role: "user", content: [{ type: "text", text: "durable one" }] }, + { role: "assistant", content: [{ type: "text", text: "durable two" }] }, + { role: "user", content: [{ type: "text", text: "" }] }, + ], + anthropicModel(), + {}, + ) + + expect(output).toHaveLength(2) + expect(output[0]!.providerOptions?.["anthropic"]).toBeDefined() + expect(output[1]!.providerOptions?.["anthropic"]).toBeDefined() + expect(marked(output)).toBe(2) + }) + + test("repeated selection over the same input does not accumulate breakpoints", () => { + const input: ModelMessage[] = [ + { role: "user", content: [{ type: "text", text: "one" }] }, + { role: "assistant", content: [{ type: "text", text: "two" }] }, + ] + const first = ProviderTransform.message(input, anthropicModel(), {}) + const second = ProviderTransform.message(input, anthropicModel(), {}) + const third = ProviderTransform.message(input, anthropicModel(), {}) + + expect(marked(first)).toBe(2) + expect(marked(second)).toBe(2) + expect(marked(third)).toBe(2) + expect(JSON.stringify(second)).toBe(JSON.stringify(first)) + expect(JSON.stringify(third)).toBe(JSON.stringify(first)) + }) + + test("appended messages remain unmarked after the transformed arrays are joined", () => { + const output = ProviderTransform.message( + [ + { role: "user", content: [{ type: "text", text: "durable one" }] }, + { role: "assistant", content: [{ type: "text", text: "durable two" }] }, + ], + anthropicModel(), + {}, + [ + { role: "user", content: [{ type: "text", text: "notice one" }] }, + { role: "user", content: [{ type: "text", text: "notice two" }] }, + ], + ) + + expect(marked(output)).toBe(2) + expect(output[0]!.providerOptions?.["anthropic"]).toBeDefined() + expect(output[1]!.providerOptions?.["anthropic"]).toBeDefined() + expect(output[2]!.providerOptions?.["anthropic"]).toBeUndefined() + expect(output[3]!.providerOptions?.["anthropic"]).toBeUndefined() + }) +}) diff --git a/packages/opencode/test/session/llm.test.ts b/packages/opencode/test/session/llm.test.ts index 3bfc722e2bec..f619a77321ac 100644 --- a/packages/opencode/test/session/llm.test.ts +++ b/packages/opencode/test/session/llm.test.ts @@ -1410,6 +1410,88 @@ describe("session.llm.stream", () => { { config: () => openAIConfig(loadFixture("openai", "gpt-5.2").model, `${state.server!.url.origin}/v1`) }, ) + it.instance( + "preserves appended message order through AI SDK and native routes", + () => + Effect.gen(function* () { + const model = loadFixture("openai", "gpt-5.2").model + const chunks = [ + { type: "response.created", response: { id: "resp-suffix" } }, + { + type: "response.output_item.added", + output_index: 0, + item: { type: "message", id: "item-suffix", status: "in_progress" }, + }, + { + type: "response.content_part.added", + item_id: "item-suffix", + output_index: 0, + content_index: 0, + part: { type: "output_text", text: "", annotations: [] }, + }, + { + type: "response.output_text.delta", + item_id: "item-suffix", + output_index: 0, + content_index: 0, + delta: "done", + logprobs: null, + }, + { + type: "response.completed", + response: { incomplete_details: null, usage: { input_tokens: 1, output_tokens: 1 } }, + }, + ] + const aiRequest = waitRequest("/responses", createEventResponse(chunks, true)) + const nativeRequest = waitRequest("/responses", createEventResponse(chunks, true)) + const resolved = yield* Provider.use.getModel(ProviderV2.ID.openai, ModelV2.ID.make(model.id)) + const agent = { + name: "test", + mode: "primary", + options: {}, + permission: [{ permission: "*", pattern: "*", action: "allow" }], + } satisfies Agent.Info + const input = (route: string): LLM.StreamInput => ({ + user: { + id: MessageID.make(`msg_user-suffix-${route}`), + sessionID: SessionID.make(`session-suffix-${route}`), + role: "user", + time: { created: Date.now() }, + agent: agent.name, + model: { providerID: ProviderV2.ID.openai, modelID: resolved.id, variant: "high" }, + }, + sessionID: SessionID.make(`session-suffix-${route}`), + model: resolved, + agent, + system: ["stable system prefix"], + messages: [{ role: "user", content: "durable history" }], + messageSuffix: [ + { role: "user", content: [{ type: "text", text: "plugin appended message" }] }, + { role: "assistant", content: [{ type: "text", text: "request limit message" }] }, + ], + tools: {}, + }) + + yield* drainWith(llmLayerWithExecutor({ flags: { experimentalNativeLlm: false } }), input("ai")) + yield* drainWith(llmLayerWithExecutor({ flags: { experimentalNativeLlm: true } }), input("native")) + + const ai = yield* Effect.promise(() => aiRequest) + const native = yield* Effect.promise(() => nativeRequest) + for (const capture of [ai, native]) { + const body = JSON.stringify(capture.body.input) + const durable = body.indexOf("durable history") + const plugin = body.indexOf("plugin appended message") + const limit = body.indexOf("request limit message") + + expect(capture.url.pathname.endsWith("/responses")).toBe(true) + expect(durable).toBeGreaterThanOrEqual(0) + expect(plugin).toBeGreaterThan(durable) + expect(limit).toBeGreaterThan(plugin) + } + }), + { config: () => openAIConfig(loadFixture("openai", "gpt-5.2").model, `${state.server!.url.origin}/v1`) }, + ) + it.instance( "uses injected native request executor for tool calls", () => diff --git a/packages/opencode/test/session/message-v2.test.ts b/packages/opencode/test/session/message-v2.test.ts index 734a30e42454..7fef0c85fe04 100644 --- a/packages/opencode/test/session/message-v2.test.ts +++ b/packages/opencode/test/session/message-v2.test.ts @@ -9,6 +9,7 @@ import { SessionID, MessageID, PartID } from "../../src/session/schema" import { Question } from "../../src/question" import { ProviderV2 } from "@opencode-ai/core/provider" import { ModelV2 } from "@opencode-ai/core/model" +import { Effect } from "effect" const sessionID = SessionID.make("session") const providerID = ProviderV2.ID.make("test") @@ -1727,3 +1728,195 @@ describe("session.message-v2.latest", () => { expect(state.tasks[0]).toMatchObject({ type: "subtask", prompt: "inspect" }) }) }) + +describe("session.message-v2 request-only partition", () => { + function partitionInput(): SessionV1.WithParts[] { + return [ + { + info: userInfo("m-durable-user"), + parts: [{ ...basePart("m-durable-user", "p1"), type: "text", text: "durable question" }] as SessionV1.Part[], + }, + { + info: assistantInfo("m-durable-assistant", "m-durable-user"), + parts: [{ ...basePart("m-durable-assistant", "p2"), type: "text", text: "durable answer" }] as SessionV1.Part[], + }, + { + info: userInfo("m-tail-one"), + parts: [{ ...basePart("m-tail-one", "p3"), type: "text", text: "request-only one" }] as SessionV1.Part[], + }, + { + info: userInfo("m-tail-two"), + parts: [{ ...basePart("m-tail-two", "p4"), type: "text", text: "request-only two" }] as SessionV1.Part[], + }, + ] + } + + const separated = (requestOnlyTailCount?: number) => + Effect.runPromise(MessageV2.toModelMessagesSplitEffect(partitionInput(), model, { requestOnlyTailCount })) + + test("partitioned conversion is byte-identical to bulk conversion at every split point", async () => { + const bulk = await MessageV2.toModelMessages(partitionInput(), model) + expect(bulk.length).toBeGreaterThan(0) + for (const requestOnlyTailCount of [0, 1, 2, 3, 4, 99]) { + const partitioned = await separated(requestOnlyTailCount) + expect(JSON.stringify([...partitioned.messages, ...partitioned.tail])).toBe(JSON.stringify(bulk)) + } + }) + + test("a negative or absent count converts in bulk", async () => { + const bulk = await MessageV2.toModelMessages(partitionInput(), model) + for (const requestOnlyTailCount of [undefined, -1, -99]) { + const output = await separated(requestOnlyTailCount) + expect(JSON.stringify(output.messages)).toBe(JSON.stringify(bulk)) + expect(output.tail).toEqual([]) + } + }) + + test("returns the converted request-only suffix separately", async () => { + const output = await separated(2) + + expect(output.messages).toEqual([ + { role: "user", content: [{ type: "text", text: "durable question" }] }, + { role: "assistant", content: [{ type: "text", text: "durable answer" }] }, + ]) + expect(output.tail).toEqual([ + { role: "user", content: [{ type: "text", text: "request-only one" }] }, + { role: "user", content: [{ type: "text", text: "request-only two" }] }, + ]) + }) + + test("the returned suffix has the requested contiguous source cardinality", async () => { + for (const requestOnlyTailCount of [1, 2, 3]) { + const output = await separated(requestOnlyTailCount) + expect(output.messages).toHaveLength(4 - requestOnlyTailCount) + expect(output.tail).toHaveLength(requestOnlyTailCount) + } + }) + + test("keeps every provider message expanded from a request-only source in the suffix", async () => { + const assistantID = "m-tail-split" + const input: SessionV1.WithParts[] = [ + { + info: userInfo("m-durable"), + parts: [{ ...basePart("m-durable", "durable"), type: "text", text: "durable" }] as SessionV1.Part[], + }, + { + info: assistantInfo(assistantID, "m-durable"), + parts: [ + { ...basePart(assistantID, "first"), type: "text", text: "request-only first" }, + { ...basePart(assistantID, "step"), type: "step-start" }, + { ...basePart(assistantID, "second"), type: "text", text: "request-only second" }, + ] as SessionV1.Part[], + }, + ] + + const output = await Effect.runPromise( + MessageV2.toModelMessagesSplitEffect(input, model, { requestOnlyTailCount: 1 }), + ) + expect(output.messages).toEqual([{ role: "user", content: [{ type: "text", text: "durable" }] }]) + expect(output.tail.map((message) => message.role)).toEqual(["assistant", "assistant"]) + expect(JSON.stringify(output.tail)).toContain("request-only first") + expect(JSON.stringify(output.tail)).toContain("request-only second") + }) + + test("keeps structured request-only content in the normal AI SDK conversion path", async () => { + const input: SessionV1.WithParts[] = [ + { + info: userInfo("m-durable"), + parts: [{ ...basePart("m-durable", "durable"), type: "text", text: "durable" }] as SessionV1.Part[], + }, + { + info: userInfo("m-tail-file"), + parts: [ + { + ...basePart("m-tail-file", "file"), + type: "file", + mime: "image/png", + filename: "image.png", + url: "https://example.com/image.png", + }, + ] as SessionV1.Part[], + }, + ] + + const bulk = await MessageV2.toModelMessages(input, model) + const output = await Effect.runPromise( + MessageV2.toModelMessagesSplitEffect(input, model, { requestOnlyTailCount: 1 }), + ) + + expect(output.messages).toEqual(bulk) + expect(output.tail).toEqual([]) + }) + + test("keeps empty request-only text in the normal AI SDK conversion path", async () => { + const assistantID = "m-tail-empty" + const input: SessionV1.WithParts[] = [ + { + info: userInfo("m-durable"), + parts: [{ ...basePart("m-durable", "durable"), type: "text", text: "durable" }] as SessionV1.Part[], + }, + { + info: assistantInfo(assistantID, "m-durable"), + parts: [{ ...basePart(assistantID, "empty"), type: "text", text: "" }] as SessionV1.Part[], + }, + ] + + const bulk = await MessageV2.toModelMessages(input, model) + const output = await Effect.runPromise( + MessageV2.toModelMessagesSplitEffect(input, model, { requestOnlyTailCount: 1 }), + ) + + expect(output.messages).toEqual(bulk) + expect(output.tail).toEqual([]) + }) +}) + +describe("session.message-v2.appendedTailCount", () => { + function after(...list: string[]): SessionV1.WithParts[] { + return list.map((id) => ({ info: { id }, parts: [] }) as unknown as SessionV1.WithParts) + } + + test("counts a fresh-ID suffix appended after the surviving pre-hook anchor", () => { + expect(MessageV2.appendedTailCount(["a", "b"], after("a", "b", "c"))).toBe(1) + expect(MessageV2.appendedTailCount(["a", "b"], after("a", "b", "c", "d"))).toBe(2) + }) + + test("returns zero when nothing was appended", () => { + expect(MessageV2.appendedTailCount(["a", "b"], after("a", "b"))).toBe(0) + }) + + test("returns zero when there is no pre-hook anchor", () => { + expect(MessageV2.appendedTailCount([], after("a"))).toBe(0) + }) + + test("returns zero on duplicate identities before or after the hook", () => { + expect(MessageV2.appendedTailCount(["a", "a"], after("a", "a", "b"))).toBe(0) + expect(MessageV2.appendedTailCount(["a", "b"], after("a", "b", "c", "c"))).toBe(0) + }) + + test("returns zero when the anchor is missing or duplicated after the hook", () => { + expect(MessageV2.appendedTailCount(["a", "b"], after("a", "c"))).toBe(0) + expect(MessageV2.appendedTailCount(["a", "b"], after("a", "b", "b"))).toBe(0) + }) + + test("returns zero when surviving pre-hook messages are reordered before the anchor", () => { + expect(MessageV2.appendedTailCount(["a", "b", "c"], after("b", "a", "c", "d"))).toBe(0) + }) + + test("returns zero when a pre-hook identity is moved or reused after the anchor", () => { + expect(MessageV2.appendedTailCount(["a", "b"], after("b", "a"))).toBe(0) + }) + + test("tolerates prefix insertion and interior removal", () => { + expect(MessageV2.appendedTailCount(["a", "b"], after("x", "a", "b", "c"))).toBe(1) + expect(MessageV2.appendedTailCount(["a", "b", "c"], after("a", "c", "d"))).toBe(1) + }) + + test("never reports a count larger than the messages actually appended", () => { + for (const count of [0, 1, 2, 3]) { + const tail = Array.from({ length: count }, (_, index) => `new${index}`) + const result = MessageV2.appendedTailCount(["a", "b"], after("a", "b", ...tail)) + expect(result).toBe(count) + } + }) +}) diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index 5a0176abc9b0..892f318806cd 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -6,7 +6,7 @@ import { SessionProjector } from "@opencode-ai/core/session/projector" import { eq } from "drizzle-orm" import { EventV2Bridge } from "@/event-v2-bridge" import { expect } from "bun:test" -import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer } from "effect" +import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer, Stream } from "effect" import path from "path" import { fileURLToPath } from "url" import { NamedError } from "@opencode-ai/core/util/error" @@ -57,6 +57,8 @@ import { RuntimeFlags } from "@/effect/runtime-flags" import { ProviderV2 } from "@opencode-ai/core/provider" import { ModelV2 } from "@opencode-ai/core/model" import { LocationServiceMap, locationServiceMapLayer } from "@opencode-ai/core/location-services" +import { LLMEvent, Usage } from "@opencode-ai/llm" +import { MAX_STEPS_PROMPT } from "@opencode-ai/core/session/runner/max-steps" const summary = Layer.succeed( SessionSummary.Service, @@ -239,6 +241,70 @@ function makeHttpNoLLMServer(input?: { mcpInstructions?: MCP.ServerInstructions[ return makePrompt(input) } +const requestOnlyInputs: LLM.StreamInput[] = [] +const requestOnlyPlugin = Layer.mock(Plugin.Service)({ + trigger: (name: Name, _input: Input, output: Output) => { + if (name !== "experimental.chat.messages.transform") return Effect.succeed(output) + return Effect.sync(() => { + const messages = (output as { messages: SessionV1.WithParts[] }).messages + const sessionID = messages.at(-1)!.info.sessionID + for (const text of ["request-only status", "request-only policy"]) { + const messageID = MessageID.ascending() + messages.push({ + info: { + id: messageID, + sessionID, + role: "user", + time: { created: Date.now() }, + agent: "build", + model: ref, + tools: {}, + mode: "", + } as unknown as SessionV1.User, + parts: [ + { + id: PartID.ascending(), + sessionID, + messageID, + type: "text", + text, + }, + ], + }) + } + return output + }) + }, + list: () => Effect.succeed([]), + init: () => Effect.void, +}) +const requestOnlyLLM = Layer.succeed( + LLM.Service, + LLM.Service.of({ + stream: (input) => { + requestOnlyInputs.push(input) + const usage = new Usage({ inputTokens: 1, outputTokens: 1, totalTokens: 2 }) + return Stream.make( + LLMEvent.textStart({ id: "txt-request-only" }), + LLMEvent.textDelta({ id: "txt-request-only", text: "done" }), + LLMEvent.textEnd({ id: "txt-request-only" }), + LLMEvent.stepFinish({ index: 0, reason: "stop", usage }), + LLMEvent.finish({ reason: "stop", usage }), + ) + }, + }), +) +const requestOnlyPrompt = testEffect( + LayerNode.compile(promptRoot, [ + [SessionSummary.node, summary], + [LSP.node, lsp], + [MCP.node, makeMcp()], + [RuntimeFlags.node, runtimeFlags], + [Plugin.node, requestOnlyPlugin], + [LLM.node, requestOnlyLLM], + ]), +) + const it = testEffect(makeHttp()) const noLLMServer = testEffect(makeHttpNoLLMServer()) const raceNoLLMServer = testEffect(makeHttpNoLLMServer({ processor: "blocking" })) @@ -554,6 +620,54 @@ it.instance("loop calls LLM and returns assistant message", () => }), ) +requestOnlyPrompt.instance( + "loop separates plugin and max-step request-only messages", + () => + Effect.gen(function* () { + requestOnlyInputs.length = 0 + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const chat = yield* sessions.create({ + title: "Appended messages", + permission: [{ permission: "*", pattern: "*", action: "allow" }], + }) + yield* prompt.prompt({ + sessionID: chat.id, + agent: "build", + noReply: true, + parts: [{ type: "text", text: "durable prompt" }], + }) + + yield* prompt.loop({ sessionID: chat.id }) + + expect(requestOnlyInputs).toHaveLength(1) + const input = requestOnlyInputs[0]! + const messages = JSON.stringify(input.messages) + const suffix = JSON.stringify(input.messageSuffix) + const suffixText = input.messageSuffix?.map((message) => + typeof message.content === "string" + ? message.content + : message.content.find((part) => part.type === "text")?.text, + ) + expect(messages).toContain("durable prompt") + expect(messages).not.toContain("request-only status") + expect(messages).not.toContain("request-only policy") + expect(suffix).toContain("request-only status") + expect(suffix).toContain("request-only policy") + expect(suffixText).toEqual(["request-only status", "request-only policy", MAX_STEPS_PROMPT]) + expect(input.messageSuffix?.at(-1)).toEqual({ + role: "assistant", + content: [{ type: "text", text: MAX_STEPS_PROMPT }], + }) + expect(input.messages).not.toContainEqual({ + role: "assistant", + content: [{ type: "text", text: MAX_STEPS_PROMPT }], + }) + expect(suffix.indexOf("request-only status")).toBeLessThan(suffix.indexOf("request-only policy")) + }), + { config: { agent: { build: { steps: 1 } } } }, +) + withMcpInstructions.instance( "loop includes MCP instructions in model system context", () =>