fix(session-recovery): reuse signed thinking blocks safely
Reuse signed Anthropic thinking blocks only when they can still sort before the target message's parts, otherwise skip recovery instead of reintroducing invalid loops. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
YeonGyu-Kim
@@ -1,5 +1,10 @@
|
||||
import { describe, expect, it } from "bun:test"
|
||||
import { existsSync, readFileSync, rmSync } from "node:fs"
|
||||
import { join } from "node:path"
|
||||
import { detectErrorType } from "./index"
|
||||
import { prependThinkingPart, prependThinkingPartAsync } from "./storage/thinking-prepend"
|
||||
import { PART_STORAGE } from "../../shared/opencode-storage-paths"
|
||||
|
||||
const { describe, expect, it, mock } = require("bun:test")
|
||||
|
||||
describe("detectErrorType", () => {
|
||||
describe("thinking_block_order errors", () => {
|
||||
@@ -278,3 +283,249 @@ describe("detectErrorType", () => {
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
type StoredPartRecord = {
|
||||
id: string
|
||||
sessionID: string
|
||||
messageID: string
|
||||
type: string
|
||||
signature?: string
|
||||
thinking?: string
|
||||
text?: string
|
||||
}
|
||||
|
||||
function cleanupParts(messageID: string): void {
|
||||
rmSync(join(PART_STORAGE, messageID), { recursive: true, force: true })
|
||||
}
|
||||
|
||||
describe("thinking-prepend", () => {
|
||||
it("writes the original signed thinking part verbatim for file-backed recovery", () => {
|
||||
const sessionID = "ses_thinking_prepend_sync"
|
||||
const targetMessageID = "msg_target_signed"
|
||||
const originalPart = {
|
||||
id: "prt_prev_signed",
|
||||
sessionID,
|
||||
messageID: "msg_prev_signed",
|
||||
type: "thinking",
|
||||
thinking: "prior reasoning",
|
||||
signature: "sig_prev",
|
||||
} as const satisfies StoredPartRecord
|
||||
|
||||
const result = prependThinkingPart(sessionID, targetMessageID, {
|
||||
isSqliteBackend: () => false,
|
||||
patchPart: async () => true,
|
||||
log: mock(() => {}),
|
||||
findLastThinkingPart: () => originalPart,
|
||||
findLastThinkingPartFromSDK: async () => null,
|
||||
readTargetPartIDs: () => ["prt_target_text"],
|
||||
readTargetPartIDsFromSDK: async () => [],
|
||||
})
|
||||
|
||||
expect(result).toBe(true)
|
||||
const writtenPath = join(PART_STORAGE, targetMessageID, `${originalPart.id}.json`)
|
||||
expect(existsSync(writtenPath)).toBe(true)
|
||||
expect(JSON.parse(readFileSync(writtenPath, "utf-8"))).toEqual(originalPart)
|
||||
|
||||
cleanupParts(targetMessageID)
|
||||
})
|
||||
|
||||
it("returns false without writing when no signed thinking part exists in history", () => {
|
||||
const sessionID = "ses_thinking_prepend_sync_missing"
|
||||
const targetMessageID = "msg_target_missing"
|
||||
|
||||
const result = prependThinkingPart(sessionID, targetMessageID, {
|
||||
isSqliteBackend: () => false,
|
||||
patchPart: async () => true,
|
||||
log: mock(() => {}),
|
||||
findLastThinkingPart: () => null,
|
||||
findLastThinkingPartFromSDK: async () => null,
|
||||
readTargetPartIDs: () => [],
|
||||
readTargetPartIDsFromSDK: async () => [],
|
||||
})
|
||||
|
||||
expect(result).toBe(false)
|
||||
expect(existsSync(join(PART_STORAGE, targetMessageID))).toBe(false)
|
||||
|
||||
cleanupParts(targetMessageID)
|
||||
})
|
||||
|
||||
it("returns false immediately when sqlite backend is active", () => {
|
||||
const result = prependThinkingPart("ses_sqlite", "msg_sqlite", {
|
||||
isSqliteBackend: () => true,
|
||||
patchPart: async () => true,
|
||||
log: mock(() => {}),
|
||||
findLastThinkingPart: () => null,
|
||||
findLastThinkingPartFromSDK: async () => null,
|
||||
readTargetPartIDs: () => [],
|
||||
readTargetPartIDsFromSDK: async () => [],
|
||||
})
|
||||
|
||||
expect(result).toBe(false)
|
||||
})
|
||||
|
||||
it("returns false when the reused signed thinking part would not sort before target parts", () => {
|
||||
const sessionID = "ses_thinking_prepend_sync_out_of_order"
|
||||
const targetMessageID = "msg_target_out_of_order"
|
||||
const originalPart = {
|
||||
id: "prt_z_reused",
|
||||
sessionID,
|
||||
messageID: "msg_prev_signed",
|
||||
type: "thinking",
|
||||
thinking: "prior reasoning",
|
||||
signature: "sig_prev",
|
||||
} as const satisfies StoredPartRecord
|
||||
|
||||
const result = prependThinkingPart(sessionID, targetMessageID, {
|
||||
isSqliteBackend: () => false,
|
||||
patchPart: async () => true,
|
||||
log: mock(() => {}),
|
||||
findLastThinkingPart: () => originalPart,
|
||||
findLastThinkingPartFromSDK: async () => null,
|
||||
readTargetPartIDs: () => ["prt_a_target"],
|
||||
readTargetPartIDsFromSDK: async () => [],
|
||||
})
|
||||
|
||||
expect(result).toBe(false)
|
||||
expect(existsSync(join(PART_STORAGE, targetMessageID))).toBe(false)
|
||||
})
|
||||
|
||||
it("patches the original signed thinking part verbatim for sdk-backed recovery", async () => {
|
||||
const prependThinkingPartAsyncUntyped = Reflect.get(
|
||||
{ prependThinkingPartAsync },
|
||||
"prependThinkingPartAsync"
|
||||
)
|
||||
const sessionID = "ses_thinking_prepend_async"
|
||||
const targetMessageID = "msg_target_async"
|
||||
const patchPartMock = mock(async () => true)
|
||||
const originalPart = {
|
||||
id: "prt_prev_async",
|
||||
type: "thinking",
|
||||
thinking: "prior reasoning",
|
||||
signature: "sig_async",
|
||||
} as const
|
||||
const client = {
|
||||
session: {
|
||||
messages: async () => ({
|
||||
data: [
|
||||
{
|
||||
info: { id: "msg_prev_async", role: "assistant" },
|
||||
parts: [originalPart],
|
||||
},
|
||||
{
|
||||
info: { id: targetMessageID, role: "assistant" },
|
||||
parts: [{ id: "prt_target_text", type: "text", text: "tool result" }],
|
||||
},
|
||||
],
|
||||
}),
|
||||
},
|
||||
}
|
||||
|
||||
const result = await Reflect.apply(prependThinkingPartAsyncUntyped, undefined, [
|
||||
client,
|
||||
sessionID,
|
||||
targetMessageID,
|
||||
{
|
||||
isSqliteBackend: () => false,
|
||||
patchPart: patchPartMock,
|
||||
log: mock(() => {}),
|
||||
findLastThinkingPart: () => null,
|
||||
findLastThinkingPartFromSDK: async () => originalPart,
|
||||
readTargetPartIDs: () => [],
|
||||
readTargetPartIDsFromSDK: async () => ["prt_target_text"],
|
||||
},
|
||||
])
|
||||
|
||||
expect(result).toBe(true)
|
||||
expect(patchPartMock).toHaveBeenCalledTimes(1)
|
||||
expect(patchPartMock.mock.calls[0]).toEqual([
|
||||
client,
|
||||
sessionID,
|
||||
targetMessageID,
|
||||
"prt_prev_async",
|
||||
originalPart,
|
||||
])
|
||||
})
|
||||
|
||||
it("returns false without patching when sdk history has no signed thinking part", async () => {
|
||||
const prependThinkingPartAsyncUntyped = Reflect.get(
|
||||
{ prependThinkingPartAsync },
|
||||
"prependThinkingPartAsync"
|
||||
)
|
||||
const sessionID = "ses_thinking_prepend_async_missing"
|
||||
const targetMessageID = "msg_target_async_missing"
|
||||
const patchPartMock = mock(async () => true)
|
||||
const client = {
|
||||
session: {
|
||||
messages: async () => ({
|
||||
data: [
|
||||
{
|
||||
info: { id: "msg_prev_async", role: "assistant" },
|
||||
parts: [{ id: "prt_prev_reasoning", type: "reasoning", text: "unsigned reasoning" }],
|
||||
},
|
||||
{
|
||||
info: { id: targetMessageID, role: "assistant" },
|
||||
parts: [{ id: "prt_target_text", type: "text", text: "tool result" }],
|
||||
},
|
||||
],
|
||||
}),
|
||||
},
|
||||
}
|
||||
|
||||
const result = await Reflect.apply(prependThinkingPartAsyncUntyped, undefined, [
|
||||
client,
|
||||
sessionID,
|
||||
targetMessageID,
|
||||
{
|
||||
isSqliteBackend: () => false,
|
||||
patchPart: patchPartMock,
|
||||
log: mock(() => {}),
|
||||
findLastThinkingPart: () => null,
|
||||
findLastThinkingPartFromSDK: async () => null,
|
||||
readTargetPartIDs: () => [],
|
||||
readTargetPartIDsFromSDK: async () => ["prt_target_text"],
|
||||
},
|
||||
])
|
||||
|
||||
expect(result).toBe(false)
|
||||
expect(patchPartMock).toHaveBeenCalledTimes(0)
|
||||
})
|
||||
|
||||
it("returns false when the sdk reused signed thinking part would not sort before target parts", async () => {
|
||||
const prependThinkingPartAsyncUntyped = Reflect.get(
|
||||
{ prependThinkingPartAsync },
|
||||
"prependThinkingPartAsync"
|
||||
)
|
||||
const sessionID = "ses_thinking_prepend_async_out_of_order"
|
||||
const targetMessageID = "msg_target_async_out_of_order"
|
||||
const patchPartMock = mock(async () => true)
|
||||
const originalPart = {
|
||||
id: "prt_z_reused",
|
||||
type: "thinking",
|
||||
thinking: "prior reasoning",
|
||||
signature: "sig_async",
|
||||
} as const
|
||||
const client = {
|
||||
session: {
|
||||
messages: async () => ({ data: [] }),
|
||||
},
|
||||
}
|
||||
|
||||
const result = await Reflect.apply(prependThinkingPartAsyncUntyped, undefined, [
|
||||
client,
|
||||
sessionID,
|
||||
targetMessageID,
|
||||
{
|
||||
isSqliteBackend: () => false,
|
||||
patchPart: patchPartMock,
|
||||
log: mock(() => {}),
|
||||
findLastThinkingPart: () => null,
|
||||
findLastThinkingPartFromSDK: async () => originalPart,
|
||||
readTargetPartIDs: () => [],
|
||||
readTargetPartIDsFromSDK: async () => ["prt_a_target"],
|
||||
},
|
||||
])
|
||||
|
||||
expect(result).toBe(false)
|
||||
expect(patchPartMock).toHaveBeenCalledTimes(0)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -2,19 +2,115 @@ import { existsSync, mkdirSync, writeFileSync } from "node:fs"
|
||||
import { join } from "node:path"
|
||||
import type { PluginInput } from "@opencode-ai/plugin"
|
||||
import { PART_STORAGE, THINKING_TYPES } from "../constants"
|
||||
import type { MessageData } from "../types"
|
||||
import type { MessageData, StoredPart } from "../types"
|
||||
import { readMessages } from "./messages-reader"
|
||||
import { readParts } from "./parts-reader"
|
||||
import { log, isSqliteBackend, patchPart } from "../../../shared"
|
||||
import { normalizeSDKResponse } from "../../../shared"
|
||||
|
||||
type OpencodeClient = PluginInput["client"]
|
||||
type StoredSignedThinkingPart = StoredPart & {
|
||||
type: "thinking" | "redacted_thinking"
|
||||
signature: string
|
||||
}
|
||||
type SDKMessagePart = NonNullable<MessageData["parts"]>[number]
|
||||
type SDKSignedThinkingPart = SDKMessagePart & {
|
||||
id: string
|
||||
type: "thinking" | "redacted_thinking"
|
||||
signature: string
|
||||
}
|
||||
|
||||
function findLastThinkingContent(sessionID: string, beforeMessageID: string): string {
|
||||
type ThinkingPrependDeps = {
|
||||
isSqliteBackend: typeof isSqliteBackend
|
||||
patchPart: typeof patchPart
|
||||
log: typeof log
|
||||
findLastThinkingPart: typeof findLastThinkingPart
|
||||
findLastThinkingPartFromSDK: typeof findLastThinkingPartFromSDK
|
||||
readTargetPartIDs: typeof readTargetPartIDs
|
||||
readTargetPartIDsFromSDK: typeof readTargetPartIDsFromSDK
|
||||
}
|
||||
|
||||
const thinkingPrependDeps: ThinkingPrependDeps = {
|
||||
isSqliteBackend,
|
||||
patchPart,
|
||||
log,
|
||||
findLastThinkingPart,
|
||||
findLastThinkingPartFromSDK,
|
||||
readTargetPartIDs,
|
||||
readTargetPartIDsFromSDK,
|
||||
}
|
||||
|
||||
function readTargetPartIDs(messageID: string): string[] {
|
||||
return readParts(messageID)
|
||||
.map((part) => part.id)
|
||||
.filter((id): id is string => typeof id === "string")
|
||||
}
|
||||
|
||||
async function readTargetPartIDsFromSDK(
|
||||
client: OpencodeClient,
|
||||
sessionID: string,
|
||||
messageID: string
|
||||
): Promise<string[]> {
|
||||
try {
|
||||
const response = await client.session.messages({ path: { id: sessionID } })
|
||||
const messages = normalizeSDKResponse(response, [] as MessageData[], { preferResponseOnMissingData: true })
|
||||
const targetMessage = messages.find((message) => message.info?.id === messageID)
|
||||
if (!targetMessage?.parts) {
|
||||
return []
|
||||
}
|
||||
|
||||
return targetMessage.parts
|
||||
.map((part) => part.id)
|
||||
.filter((id): id is string => typeof id === "string")
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
}
|
||||
|
||||
function canPrependBeforeTargetParts(partID: string, targetPartIDs: string[]): boolean {
|
||||
const firstTargetPartID = [...targetPartIDs].sort((left, right) => left.localeCompare(right))[0]
|
||||
return !firstTargetPartID || partID.localeCompare(firstTargetPartID) < 0
|
||||
}
|
||||
|
||||
function isStoredSignedThinkingPart(part: StoredPart): part is StoredSignedThinkingPart {
|
||||
if (!THINKING_TYPES.has(part.type)) {
|
||||
return false
|
||||
}
|
||||
|
||||
if (part.type === "reasoning") {
|
||||
return false
|
||||
}
|
||||
|
||||
const signature = Reflect.get(part, "signature")
|
||||
return typeof signature === "string" && signature.length > 0
|
||||
}
|
||||
|
||||
function isSDKSignedThinkingPart(part: SDKMessagePart): part is SDKSignedThinkingPart {
|
||||
if (!part.type || !THINKING_TYPES.has(part.type)) {
|
||||
return false
|
||||
}
|
||||
|
||||
if (part.type === "reasoning") {
|
||||
return false
|
||||
}
|
||||
|
||||
return typeof part.id === "string"
|
||||
&& typeof (part as { signature?: unknown }).signature === "string"
|
||||
&& ((part as { signature?: string }).signature?.length ?? 0) > 0
|
||||
}
|
||||
|
||||
function toPatchBody(part: SDKSignedThinkingPart): Record<string, unknown> {
|
||||
return { ...part }
|
||||
}
|
||||
|
||||
function findLastThinkingPart(
|
||||
sessionID: string,
|
||||
beforeMessageID: string
|
||||
): StoredSignedThinkingPart | null {
|
||||
const messages = readMessages(sessionID)
|
||||
|
||||
const currentIndex = messages.findIndex((message) => message.id === beforeMessageID)
|
||||
if (currentIndex === -1) return ""
|
||||
if (currentIndex === -1) return null
|
||||
|
||||
for (let i = currentIndex - 1; i >= 0; i--) {
|
||||
const message = messages[i]
|
||||
@@ -22,63 +118,62 @@ function findLastThinkingContent(sessionID: string, beforeMessageID: string): st
|
||||
|
||||
const parts = readParts(message.id)
|
||||
for (const part of parts) {
|
||||
if (THINKING_TYPES.has(part.type)) {
|
||||
const thinking = (part as { thinking?: string; text?: string }).thinking
|
||||
const reasoning = (part as { thinking?: string; text?: string }).text
|
||||
const content = thinking || reasoning
|
||||
if (content && content.trim().length > 0) {
|
||||
return content
|
||||
}
|
||||
if (isStoredSignedThinkingPart(part)) {
|
||||
return part
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return ""
|
||||
return null
|
||||
}
|
||||
|
||||
export function prependThinkingPart(sessionID: string, messageID: string): boolean {
|
||||
if (isSqliteBackend()) {
|
||||
export function prependThinkingPart(
|
||||
sessionID: string,
|
||||
messageID: string,
|
||||
deps: ThinkingPrependDeps = thinkingPrependDeps
|
||||
): boolean {
|
||||
if (deps.isSqliteBackend()) {
|
||||
log("[session-recovery] Disabled on SQLite backend: prependThinkingPart (use async variant)")
|
||||
return false
|
||||
}
|
||||
|
||||
const previousThinkingPart = deps.findLastThinkingPart(sessionID, messageID)
|
||||
if (!previousThinkingPart) {
|
||||
return false
|
||||
}
|
||||
|
||||
if (!canPrependBeforeTargetParts(previousThinkingPart.id, deps.readTargetPartIDs(messageID))) {
|
||||
return false
|
||||
}
|
||||
|
||||
const partDir = join(PART_STORAGE, messageID)
|
||||
|
||||
if (!existsSync(partDir)) {
|
||||
mkdirSync(partDir, { recursive: true })
|
||||
}
|
||||
|
||||
const previousThinking = findLastThinkingContent(sessionID, messageID)
|
||||
|
||||
const partId = `prt_0000000000_${messageID}_thinking`
|
||||
const part = {
|
||||
id: partId,
|
||||
sessionID,
|
||||
messageID,
|
||||
type: "thinking",
|
||||
thinking: previousThinking || "[Continuing from previous reasoning]",
|
||||
synthetic: true,
|
||||
}
|
||||
|
||||
try {
|
||||
writeFileSync(join(partDir, `${partId}.json`), JSON.stringify(part, null, 2))
|
||||
writeFileSync(
|
||||
join(partDir, `${previousThinkingPart.id}.json`),
|
||||
JSON.stringify(previousThinkingPart, null, 2)
|
||||
)
|
||||
return true
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
async function findLastThinkingContentFromSDK(
|
||||
async function findLastThinkingPartFromSDK(
|
||||
client: OpencodeClient,
|
||||
sessionID: string,
|
||||
beforeMessageID: string
|
||||
): Promise<string> {
|
||||
): Promise<SDKSignedThinkingPart | null> {
|
||||
try {
|
||||
const response = await client.session.messages({ path: { id: sessionID } })
|
||||
const messages = normalizeSDKResponse(response, [] as MessageData[], { preferResponseOnMissingData: true })
|
||||
|
||||
const currentIndex = messages.findIndex((m) => m.info?.id === beforeMessageID)
|
||||
if (currentIndex === -1) return ""
|
||||
if (currentIndex === -1) return null
|
||||
|
||||
for (let i = currentIndex - 1; i >= 0; i--) {
|
||||
const msg = messages[i]
|
||||
@@ -86,39 +181,43 @@ async function findLastThinkingContentFromSDK(
|
||||
if (!msg.parts) continue
|
||||
|
||||
for (const part of msg.parts) {
|
||||
if (part.type && THINKING_TYPES.has(part.type)) {
|
||||
const content = part.thinking || part.text
|
||||
if (content && content.trim().length > 0) return content
|
||||
if (isSDKSignedThinkingPart(part)) {
|
||||
return part
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return ""
|
||||
return null
|
||||
}
|
||||
return ""
|
||||
return null
|
||||
}
|
||||
|
||||
export async function prependThinkingPartAsync(
|
||||
client: OpencodeClient,
|
||||
sessionID: string,
|
||||
messageID: string
|
||||
messageID: string,
|
||||
deps: ThinkingPrependDeps = thinkingPrependDeps
|
||||
): Promise<boolean> {
|
||||
const previousThinking = await findLastThinkingContentFromSDK(client, sessionID, messageID)
|
||||
const previousThinkingPart = await deps.findLastThinkingPartFromSDK(client, sessionID, messageID)
|
||||
if (!previousThinkingPart) {
|
||||
return false
|
||||
}
|
||||
|
||||
const partId = `prt_0000000000_${messageID}_thinking`
|
||||
const part: Record<string, unknown> = {
|
||||
id: partId,
|
||||
sessionID,
|
||||
messageID,
|
||||
type: "thinking",
|
||||
thinking: previousThinking || "[Continuing from previous reasoning]",
|
||||
synthetic: true,
|
||||
const targetPartIDs = await deps.readTargetPartIDsFromSDK(client, sessionID, messageID)
|
||||
if (!canPrependBeforeTargetParts(previousThinkingPart.id, targetPartIDs)) {
|
||||
return false
|
||||
}
|
||||
|
||||
try {
|
||||
return await patchPart(client, sessionID, messageID, partId, part)
|
||||
return await deps.patchPart(
|
||||
client,
|
||||
sessionID,
|
||||
messageID,
|
||||
previousThinkingPart.id,
|
||||
toPatchBody(previousThinkingPart)
|
||||
)
|
||||
} catch (error) {
|
||||
log("[session-recovery] prependThinkingPartAsync failed", { error: String(error) })
|
||||
deps.log("[session-recovery] prependThinkingPartAsync failed", { error: String(error) })
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user