Skip to content

fix(opencode): recover stale encrypted reasoning state #43595

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wang-kaopu wants to merge 1 commit into
base: dev
Choose a base branch
from
+293 −5
Open

fix(opencode): recover stale encrypted reasoning state #43595

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 109 additions & 4 deletions packages/opencode/src/session/processor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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: (
Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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,
)
Expand All @@ -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,
Expand Down
185 changes: 184 additions & 1 deletion packages/opencode/test/session/processor-effect.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand All @@ -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 }) =>
Expand Down
Loading