From 12d5060c4fd4ca4647ccd9ec47dea8df5e7522c2 Mon Sep 17 00:00:00 2001 From: kelvinwww <1970138194@qq.com> Date: Thu, 20 Aug 2026 14:59:21 +0800 Subject: [PATCH] fix(opencode): recover stale encrypted reasoning state --- packages/opencode/src/session/processor.ts | 113 ++++++++++- .../test/session/processor-effect.test.ts | 185 +++++++++++++++++- 2 files changed, 293 insertions(+), 5 deletions(-) diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 20aa8a8404d8..9c94da19212c 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -25,10 +25,51 @@ import { isRecord } from "@/util/record" import { EventV2Bridge } from "@/event-v2-bridge" import { Database } from "@opencode-ai/core/database/database" import { Usage, type LLMEvent } from "@opencode-ai/llm" +import type { ModelMessage } from "ai" const DOOM_LOOP_THRESHOLD = 3 export type Result = "compact" | "stop" | "continue" +function containsInvalidEncryptedContent(value: unknown) { + return typeof value === "string" && value.toLowerCase().includes("invalid_encrypted_content") +} + +class InvalidEncryptedContentError extends Error { + constructor(message: string) { + super(message) + this.name = "InvalidEncryptedContentError" + } +} + +function clearReasoningState(messages: ModelMessage[]): ModelMessage[] { + return messages.map((message) => { + if (message.role !== "assistant" || !Array.isArray(message.content)) return message + + return { + ...message, + content: message.content.map((part) => { + if (part.type !== "reasoning") return part + + const clear = (value: typeof part.providerOptions) => { + if (!isRecord(value) || !isRecord(value.openai)) return value + if (!("itemId" in value.openai) && !("reasoningEncryptedContent" in value.openai)) return value + + const openai = { ...value.openai } + delete openai.itemId + delete openai.reasoningEncryptedContent + return { ...value, openai } + } + + const providerOptions = clear(part.providerOptions) + return { + ...part, + ...(providerOptions === part.providerOptions ? {} : { providerOptions }), + } + }), + } + }) +} + export interface Handle { readonly message: SessionV1.Assistant readonly updateToolCall: ( @@ -120,6 +161,40 @@ const layer = Layer.effect( aborted, }) + const isInvalidEncryptedContent = (error: unknown) => { + if (error instanceof InvalidEncryptedContentError) return true + + const parsed = parse(error) + if (SessionV1.APIError.isInstance(parsed)) { + return ( + containsInvalidEncryptedContent(parsed.data.message) || + containsInvalidEncryptedContent(parsed.data.responseBody) + ) + } + if (isRecord(parsed.data)) return containsInvalidEncryptedContent(parsed.data.message) + return false + } + + const clearStoredReasoningState = Effect.fn("SessionProcessor.clearStoredReasoningState")(function* () { + const messages = yield* session.messages({ sessionID: ctx.sessionID }) + for (const message of messages) { + if (message.info.id === ctx.assistantMessage.id) continue + + for (const part of message.parts) { + if (part.type !== "reasoning" || !isRecord(part.metadata) || !isRecord(part.metadata.openai)) continue + if (!("itemId" in part.metadata.openai) && !("reasoningEncryptedContent" in part.metadata.openai)) continue + + const openai = { ...part.metadata.openai } + delete openai.itemId + delete openai.reasoningEncryptedContent + yield* session.updatePart({ + ...part, + metadata: { ...part.metadata, openai }, + }) + } + } + }) + const settleToolCall = Effect.fn("SessionProcessor.settleToolCall")(function* (toolCallID: string) { const done = ctx.toolcalls[toolCallID]?.done delete ctx.toolcalls[toolCallID] @@ -418,8 +493,12 @@ const layer = Layer.effect( return } - case "provider-error": + case "provider-error": { + if (containsInvalidEncryptedContent(value.message)) { + throw new InvalidEncryptedContentError(value.message) + } throw new Error(value.message) + } case "step-start": if (!ctx.snapshot) ctx.snapshot = yield* snapshot.track() @@ -631,16 +710,29 @@ const layer = Layer.effect( }) ctx.needsCompaction = false ctx.shouldBreak = (yield* config.get()).experimental?.continue_loop_on_deny !== true + let current = streamInput + let recoveredReasoning = false + let emitted = false return yield* Effect.gen(function* () { - yield* Effect.gen(function* () { + const runAttempt = Effect.gen(function* () { ctx.currentText = undefined ctx.reasoningMap = {} yield* status.set(ctx.sessionID, { type: "busy" }) - const stream = llm.stream(streamInput) + const stream = llm.stream(current) yield* stream.pipe( - Stream.tap((event) => handleEvent(event)), + Stream.tap((event) => { + if ( + event.type === "text-start" || + event.type === "reasoning-start" || + event.type === "tool-call" || + event.type === "tool-input-start" + ) { + emitted = true + } + return handleEvent(event) + }), Stream.takeUntil(() => ctx.needsCompaction), Stream.runDrain, ) @@ -657,6 +749,19 @@ const layer = Layer.effect( (cause) => !Cause.hasInterruptsOnly(cause), (cause) => Effect.fail(Cause.squash(cause)), ), + ) + + yield* runAttempt.pipe( + Effect.catchIf( + (error) => isInvalidEncryptedContent(error) && !emitted && !recoveredReasoning, + () => + Effect.gen(function* () { + recoveredReasoning = true + yield* clearStoredReasoningState() + current = { ...current, messages: clearReasoningState(current.messages) } + yield* runAttempt + }), + ), Effect.retry( SessionRetry.policy({ provider: input.model.providerID, diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 052477d0a2e7..7fe4dd91ad47 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -3,7 +3,7 @@ import { Database } from "@opencode-ai/core/database/database" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { EventV2Bridge } from "@/event-v2-bridge" import { expect } from "bun:test" -import { tool } from "ai" +import { APICallError, tool } from "ai" import { Cause, Effect, Exit, Fiber, Layer, Stream } from "effect" import path from "path" import z from "zod" @@ -226,6 +226,44 @@ const fragmentFailureLLM = Layer.succeed( const fragmentFailureEnv = LayerNode.compile(root, [...replacements, [LLM.node, fragmentFailureLLM]]) const itFragmentFailure = testEffect(fragmentFailureEnv) +let encryptedRecoveryCalls = 0 +let encryptedRecoveryMode: "success" | "fail" | "stream-failure" = "success" +let encryptedRecoveryInputs: LLM.StreamInput[] = [] +const encryptedRecoveryLLM = Layer.succeed( + LLM.Service, + LLM.Service.of({ + stream: (input) => { + encryptedRecoveryInputs.push(input) + encryptedRecoveryCalls += 1 + if (encryptedRecoveryMode === "stream-failure" && encryptedRecoveryCalls === 1) { + return Stream.fail( + new APICallError({ + message: "Upstream request failed: [invalid_encrypted_content] stale reasoning", + url: "https://example.com", + requestBodyValues: {}, + statusCode: 400, + responseHeaders: { "content-type": "application/json" }, + responseBody: JSON.stringify({ error: { code: "invalid_encrypted_content" } }), + isRetryable: false, + }), + ) + } + if (encryptedRecoveryMode === "fail" || encryptedRecoveryCalls === 1) { + return Stream.make(LLMEvent.providerError({ message: "invalid_encrypted_content: stale reasoning" })) + } + return Stream.make( + LLMEvent.textStart({ id: "text-1" }), + LLMEvent.textDelta({ id: "text-1", text: "recovered" }), + LLMEvent.textEnd({ id: "text-1" }), + LLMEvent.stepFinish({ index: 0, reason: "stop" }), + LLMEvent.finish({ reason: "stop" }), + ) + }, + }), +) +const encryptedRecoveryEnv = LayerNode.compile(root, [...replacements, [LLM.node, encryptedRecoveryLLM]]) +const itEncryptedRecovery = testEffect(encryptedRecoveryEnv) + const boot = Effect.fn("test.boot")(function* () { const processors = yield* SessionProcessor.Service const session = yield* Session.Service @@ -237,6 +275,151 @@ const boot = Effect.fn("test.boot")(function* () { // Tests // --------------------------------------------------------------------------- +itEncryptedRecovery.live("session.processor recovers from invalid encrypted reasoning stream failure", () => + provideTmpdirInstance( + (dir) => + Effect.gen(function* () { + encryptedRecoveryCalls = 0 + encryptedRecoveryMode = "stream-failure" + encryptedRecoveryInputs = [] + + const { processors, session, provider } = yield* boot() + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "stale reasoning") + const historical = yield* assistant(chat.id, parent.id, path.resolve(dir)) + yield* session.updatePart({ + id: PartID.ascending(), + messageID: historical.id, + sessionID: chat.id, + type: "reasoning", + text: "visible reasoning", + time: { start: Date.now() }, + metadata: { + openai: { + itemId: "item-1", + reasoningEncryptedContent: "encrypted-1", + keep: "yes", + }, + }, + }) + const currentParent = yield* user(chat.id, "continue") + const msg = yield* assistant(chat.id, currentParent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl }) + + const value = yield* handle.process({ + user: { + id: currentParent.id, + sessionID: chat.id, + role: "user", + time: currentParent.time, + agent: currentParent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies SessionV1.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [ + { + role: "assistant", + content: [ + { + type: "reasoning", + text: "visible reasoning", + providerOptions: { + openai: { + itemId: "item-1", + reasoningEncryptedContent: "encrypted-1", + keep: "yes", + }, + }, + }, + { type: "text", text: "previous answer" }, + ], + }, + { role: "user", content: "continue" }, + ], + tools: {}, + }) + + const stored = yield* session.messages({ sessionID: chat.id }) + const storedReasoning = stored + .flatMap((item) => item.parts) + .find((part): part is SessionV1.ReasoningPart => part.type === "reasoning" && part.messageID === historical.id) + + expect(value).toBe("continue") + expect(encryptedRecoveryCalls).toBe(2) + expect(encryptedRecoveryInputs[0]?.messages[0]).toMatchObject({ + role: "assistant", + content: [ + { + type: "reasoning", + providerOptions: { + openai: { itemId: "item-1", reasoningEncryptedContent: "encrypted-1", keep: "yes" }, + }, + }, + { type: "text", text: "previous answer" }, + ], + }) + expect(encryptedRecoveryInputs[1]?.messages[0]).toMatchObject({ + role: "assistant", + content: [ + { + type: "reasoning", + text: "visible reasoning", + providerOptions: { openai: { keep: "yes" } }, + }, + { type: "text", text: "previous answer" }, + ], + }) + expect(storedReasoning?.text).toBe("visible reasoning") + expect(storedReasoning?.metadata).toEqual({ openai: { keep: "yes" } }) + }), + { config: cfg }, + ), +) + +itEncryptedRecovery.live("session.processor only recovers invalid encrypted reasoning content once", () => + provideTmpdirInstance( + (dir) => + Effect.gen(function* () { + encryptedRecoveryCalls = 0 + encryptedRecoveryMode = "fail" + encryptedRecoveryInputs = [] + + const { processors, session, provider } = yield* boot() + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "stale reasoning") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl }) + + const value = yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies SessionV1.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "stale reasoning" }], + tools: {}, + }) + + expect(value).toBe("stop") + expect(encryptedRecoveryCalls).toBe(2) + expect(JSON.stringify(handle.message.error?.data)).toContain("invalid_encrypted_content") + }), + { config: cfg }, + ), +) + it.live("session.processor effect tests capture llm input cleanly", () => provideTmpdirServer( ({ dir, llm }) =>