luna/oh-my-opencode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(session-recovery): recover interrupted idle tool turns

Browse Source
This commit is contained in:
YeonGyu-Kim
2026-05-17 15:08:39 +09:00
parent fbec112bc2
commit f43effb842
11 changed files with 531 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/features/background-agent/parent-wake-notifier.ts
+13
View File
@@ -34,6 +34,9 @@ type ParentWakeSessionMessage = {
type?: string
text?: string
content?: unknown
state?: {
status?: unknown
}
}>
}
@@ -323,6 +326,15 @@ export class ParentWakeNotifier {
return undefined
}
private parentWakePartIsWaitingOnTool(part: NonNullable<ParentWakeSessionMessage["parts"]>[number]): boolean {
if (part.type !== "tool" && part.type !== "tool_use") {
return false
}
const status = part.state?.status
return status === "pending" || status === "running"
}
private latestAssistantTurnIsWaitingOnTools(messages: ParentWakeSessionMessage[]): boolean {
for (let index = messages.length - 1; index >= 0; index--) {
const message = messages[index]
@@ -332,6 +344,7 @@ export class ParentWakeNotifier {
const role = this.getParentWakeMessageRole(message)
if (role === "assistant") {
return this.getParentWakeMessageFinish(message) === "tool-calls"
|| message.parts?.some((part) => this.parentWakePartIsWaitingOnTool(part)) === true
}
if (role === "user") {
return false
src/features/background-agent/task-completion-cleanup.test.ts
+39 -1
View File
@@ -24,7 +24,7 @@ type SessionMessageForTest = {
finish?: string
time?: { created?: number }
}
parts?: Array<{ type?: string }>
parts?: Array<{ type?: string; state?: { status?: string } }>
}
type FakeTimers = {
@@ -508,6 +508,44 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => {
expect(promptAsyncCalls).toHaveLength(0)
})
test("#when parent status is idle but latest assistant turn has running tool state without finish #then background completion does not fork a reply", async () => {
// given
const sessionStatuses: Record<string, { type: string }> = {
"parent-1": { type: "idle" },
}
const sessionMessages: SessionMessageForTest[] = [
{
info: { role: "user", time: { created: 1778819814009 } },
parts: [{ type: "text" }],
},
{
info: { role: "assistant", time: { created: 1778819997535 } },
parts: [
{ type: "tool", state: { status: "running" } },
{ type: "tool", state: { status: "pending" } },
],
},
]
const { manager, promptAsyncCalls } = createManager(true, sessionStatuses, undefined, sessionMessages)
managerUnderTest = manager
const task = createTask({
id: "task-a",
parentSessionId: "parent-1",
description: "task A",
status: "completed",
completedAt: new Date("2026-05-17T05:25:01.000Z"),
})
getTasks(manager).set(task.id, task)
getPendingByParent(manager).set(task.parentSessionId, new Set([task.id]))
// when
await notifyParentSessionForTest(manager, task)
await waitForCoalescedFlush()
// then
expect(promptAsyncCalls).toHaveLength(0)
})
test("#when stale tool-call history keeps blocking an all-complete wake #then completion eventually wakes the parent", async () => {
// given
const sessionStatuses: Record<string, { type: string }> = {
src/hooks/anthropic-context-window-limit-recovery/storage.test.ts
+3
View File
@@ -11,7 +11,10 @@ const findToolResultsBySize = mock<(_: string) => ToolResultInfo[]>(() => [])
const truncateToolResult = mock<(_: string) => TruncateToolResult>(() => ({ success: false }))
mock.module("./tool-result-storage", () => ({
countTruncatedResults: () => 0,
findLargestToolResult: () => null,
findToolResultsBySize,
getTotalToolOutputSize: () => 0,
truncateToolResult,
}))
src/hooks/session-recovery/hook.test.ts
+102 -1
View File
@@ -1,8 +1,26 @@
import { describe, expect, test } from "bun:test"
import { afterEach, describe, expect, test } from "bun:test"
import { createSessionRecoveryHook } from "./hook"
import { releaseAllPromptAsyncReservationsForTesting } from "../../shared/prompt-async-gate"
type RecoverableInfo = Parameters<ReturnType<typeof createSessionRecoveryHook>["handleSessionRecovery"]>[0]
type PromptAsyncCall = {
path: { id: string }
body: {
parts: Array<{
toolUseId?: string
content?: Array<{ text?: string }>
}>
agent?: string
model?: { providerID: string; modelID: string }
variant?: string
}
}
afterEach(() => {
releaseAllPromptAsyncReservationsForTesting()
})
function createPrefillErrorInfo(): RecoverableInfo {
return {
id: "msg_failed_prefill",
@@ -84,3 +102,86 @@ describe("session-recovery hook persistent dedupe", () => {
expect(counts.abort).toBe(1)
})
})
describe("session-recovery hook interrupted idle recovery", () => {
test("#given idle session has an unfinished assistant turn with pending tool parts #when idle recovery runs #then it injects only interrupted tool results once", async () => {
// given
const promptAsyncCalls: PromptAsyncCall[] = []
const ctx = {
client: {
session: {
status: async () => ({ data: { ses_idle_interrupted: { type: "idle" } } }),
messages: async () => ({
data: [
{
info: {
id: "msg_user",
role: "user",
agent: "Sisyphus",
model: { providerID: "anthropic", modelID: "claude-opus-4-7", variant: "max" },
},
parts: [{ type: "text", text: "run /init-deep ultrafucking deep" }],
},
{
info: {
id: "msg_assistant_unfinished",
role: "assistant",
sessionID: "ses_idle_interrupted",
time: { created: 1778995446058 },
},
parts: [
{
type: "tool",
callID: "call_completed",
name: "bash",
input: {},
state: { status: "completed" },
},
{
type: "tool_use",
id: "toolu_running",
name: "bash",
input: {},
state: { status: "running" },
},
{
type: "tool_use",
id: "toolu_pending",
name: "task",
input: {},
state: { status: "pending" },
},
],
},
],
}),
promptAsync: async (call: PromptAsyncCall) => {
promptAsyncCalls.push(call)
return {}
},
},
},
directory: "/tmp/session-recovery-idle-test",
}
const hook = createSessionRecoveryHook(ctx as never)
// when
const firstResult = await hook.handleInterruptedToolResultsOnIdle("ses_idle_interrupted")
const secondResult = await hook.handleInterruptedToolResultsOnIdle("ses_idle_interrupted")
// then
expect(firstResult).toBe(true)
expect(secondResult).toBe(false)
expect(promptAsyncCalls).toHaveLength(1)
expect(promptAsyncCalls[0]?.body.parts.map((part) => part.toolUseId)).toEqual([
"toolu_running",
"toolu_pending",
])
expect(promptAsyncCalls[0]?.body.parts[0]?.content?.[0]?.text).toBe(
"Tool execution was interrupted before producing a result.",
)
expect(promptAsyncCalls[0]?.body.agent).toBe("Sisyphus")
expect(promptAsyncCalls[0]?.body.model).toEqual({ providerID: "anthropic", modelID: "claude-opus-4-7" })
expect(promptAsyncCalls[0]?.body.variant).toBe("max")
})
})
src/hooks/session-recovery/hook.ts
+94
View File
@@ -4,6 +4,7 @@ import { log } from "../../shared/logger"
import { detectErrorType } from "./detect-error-type"
import type { RecoveryErrorType } from "./detect-error-type"
import type { MessageData } from "./types"
import { normalizeSDKResponse } from "../../shared"
import { recoverToolResultMissing } from "./recover-tool-result-missing"
import { recoverUnavailableTool } from "./recover-unavailable-tool"
import { recoverThinkingBlockOrder } from "./recover-thinking-block-order"
@@ -24,6 +25,7 @@ export interface SessionRecoveryOptions {
export interface SessionRecoveryHook {
handleSessionRecovery: (info: MessageInfo) => Promise<boolean>
handleInterruptedToolResultsOnIdle: (sessionID: string) => Promise<boolean>
isRecoverableError: (error: unknown) => boolean
setOnAbortCallback: (callback: (sessionID: string) => void) => void
setOnRecoveryCompleteCallback: (callback: (sessionID: string) => void) => void
@@ -31,6 +33,7 @@ export interface SessionRecoveryHook {
export function createSessionRecoveryHook(ctx: PluginInput, options?: SessionRecoveryOptions): SessionRecoveryHook {
const processingErrors = new Set<string>()
const processingInterruptedToolMessages = new Set<string>()
const experimental = options?.experimental
let onAbortCallback: ((sessionID: string) => void) | null = null
let onRecoveryCompleteCallback: ((sessionID: string) => void) | null = null
@@ -47,6 +50,96 @@ export function createSessionRecoveryHook(ctx: PluginInput, options?: SessionRec
return detectErrorType(error) !== null
}
const assistantMessageIsFinished = (message: MessageData): boolean => {
if (message.info?.error) {
return true
}
const finish = message.info?.finish
if ((typeof finish === "string" && finish.length > 0) || finish === true) {
return true
}
const completed = message.info?.time?.completed
if (typeof completed === "number" && Number.isFinite(completed)) {
return true
}
return typeof completed === "string" && completed.length > 0
}
const messageHasInterruptedToolResults = (message: MessageData): boolean => {
return message.parts?.some((part) =>
(part.type === "tool" || part.type === "tool_use")
&& (part.state?.status === "pending" || part.state?.status === "running")
&& typeof (part.callID ?? part.id) === "string"
&& /^(toolu_|call_)/.test(part.callID ?? part.id ?? "")
) === true
}
const findLatestAssistantMessage = (messages: MessageData[]): MessageData | undefined => {
for (let index = messages.length - 1; index >= 0; index--) {
const message = messages[index]
if (message?.info?.role === "assistant") {
return message
}
}
return undefined
}
const handleInterruptedToolResultsOnIdle = async (sessionID: string): Promise<boolean> => {
let recoveryStarted = false
let assistantMessageIDForRecovery: string | undefined
try {
const messagesResp = await ctx.client.session.messages({
path: { id: sessionID },
query: { directory: ctx.directory },
})
const messages = normalizeSDKResponse(messagesResp, [] as MessageData[])
const latestAssistant = findLatestAssistantMessage(messages)
if (!latestAssistant?.info?.id) {
return false
}
if (assistantMessageIsFinished(latestAssistant) || !messageHasInterruptedToolResults(latestAssistant)) {
return false
}
const assistantMessageID = latestAssistant.info.id
if (processingInterruptedToolMessages.has(assistantMessageID)) {
return false
}
processingInterruptedToolMessages.add(assistantMessageID)
assistantMessageIDForRecovery = assistantMessageID
if (onAbortCallback) {
onAbortCallback(sessionID)
}
recoveryStarted = true
const lastUser = findLastUserMessage(messages)
const resumeConfig = extractResumeConfig(lastUser, sessionID)
const success = await recoverToolResultMissing(ctx.client, sessionID, latestAssistant, resumeConfig, {
recoverStatuses: new Set(["pending", "running"]),
resultText: "Tool execution was interrupted before producing a result.",
source: "session-recovery-interrupted-tool-results",
})
if (!success) {
processingInterruptedToolMessages.delete(assistantMessageID)
}
return success
} catch (err) {
if (assistantMessageIDForRecovery) {
processingInterruptedToolMessages.delete(assistantMessageIDForRecovery)
}
log("[session-recovery] Interrupted tool result recovery failed:", { sessionID, error: err })
return false
} finally {
if (recoveryStarted && onRecoveryCompleteCallback) {
onRecoveryCompleteCallback(sessionID)
}
}
}
const handleSessionRecovery = async (info: MessageInfo): Promise<boolean> => {
if (!info || info.role !== "assistant" || !info.error) return false
@@ -175,6 +268,7 @@ export function createSessionRecoveryHook(ctx: PluginInput, options?: SessionRec
return {
handleSessionRecovery,
handleInterruptedToolResultsOnIdle,
isRecoverableError,
setOnAbortCallback,
setOnRecoveryCompleteCallback,
src/hooks/session-recovery/recover-tool-result-missing.test.ts
+56 -1
View File
@@ -9,8 +9,9 @@ mock.module("../../shared/opencode-storage-detection", () => ({
isSqliteBackend: () => sqliteBackend,
}))
mock.module("./storage", () => ({
mock.module("./storage/parts-reader", () => ({
readParts: () => storedParts,
readPartsFromSDK: () => storedParts,
}))
const { recoverToolResultMissing } = await import("./recover-tool-result-missing")
@@ -91,6 +92,60 @@ describe("recoverToolResultMissing", () => {
})
})
it("sends only interrupted sqlite tool results when recoverStatuses is provided", async () => {
//#given
sqliteBackend = true
const { client, promptAsync } = createMockClient([
{
info: { id: "msg_failed", role: "assistant" },
parts: [
{
type: "tool",
id: "prt_completed_call",
callID: "call_completed",
name: "bash",
input: {},
state: { status: "completed" },
},
{
type: "tool",
id: "prt_running_call",
callID: "call_running",
name: "bash",
input: {},
state: { status: "running" },
},
{
type: "tool",
id: "prt_pending_call",
callID: "toolu_pending",
name: "task",
input: {},
state: { status: "pending" },
},
],
},
])
//#when
const result = await recoverToolResultMissing(client, "ses_1", failedAssistantMsg, undefined, {
recoverStatuses: new Set(["pending", "running"]),
resultText: "Tool execution was interrupted before producing a result.",
source: "session-recovery-interrupted-tool-results",
})
//#then
expect(result).toBe(true)
expect(promptAsync).toHaveBeenCalledTimes(1)
const call = promptAsync.mock.calls[0]?.[0] as {
body: {
parts: Array<{ toolUseId: string; content: Array<{ text: string }> }>
}
}
expect(call.body.parts.map((part) => part.toolUseId)).toEqual(["call_running", "toolu_pending"])
expect(call.body.parts[0]?.content[0]?.text).toBe("Tool execution was interrupted before producing a result.")
})
it("returns false for stored parts when tool part has no valid callID", async () => {
//#given
storedParts = [{ type: "tool", id: "prt_stored_missing_call", tool: "bash", state: { input: {} } }]
src/hooks/session-recovery/recover-tool-result-missing.ts
+39 -11
View File
@@ -1,6 +1,6 @@
import type { createOpencodeClient } from "@opencode-ai/sdk"
import type { MessageData, ResumeConfig } from "./types"
import { readParts } from "./storage"
import { readParts } from "./storage/parts-reader"
import { isSqliteBackend } from "../../shared/opencode-storage-detection"
import { normalizeSDKResponse } from "../../shared"
import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate"
@@ -21,6 +21,12 @@ type ClientWithPromptAsync = {
}
}
export type RecoverToolResultMissingOptions = {
recoverStatuses?: ReadonlySet<string>
resultText?: string
source?: string
}
function hasPromptAsync(client: Client): client is Client & ClientWithPromptAsync {
return "promptAsync" in client.session && typeof client.session.promptAsync === "function"
}
@@ -36,21 +42,26 @@ interface ToolUsePart {
interface MessagePart {
type: string
id?: string
state?: {
status?: unknown
}
}
function isValidToolUseID(id: string | undefined): id is string {
return typeof id === "string" && /^(toolu_|call_)/.test(id)
}
function normalizeMessagePart(part: { type: string; id?: string; callID?: string }): MessagePart | null {
function normalizeMessagePart(part: { type: string; id?: string; callID?: string; state?: { status?: unknown } }): MessagePart | null {
if (part.type === "tool" || part.type === "tool_use") {
if (!isValidToolUseID(part.callID)) {
const toolUseID = part.callID ?? part.id
if (!isValidToolUseID(toolUseID)) {
return null
}
return {
type: "tool_use",
id: part.callID,
id: toolUseID,
state: part.state,
}
}
@@ -60,8 +71,21 @@ function normalizeMessagePart(part: { type: string; id?: string; callID?: string
}
}
function extractToolUseIds(parts: MessagePart[]): string[] {
return parts.filter((part): part is ToolUsePart => part.type === "tool_use" && isValidToolUseID(part.id)).map((part) => part.id)
function shouldRecoverToolUsePart(part: MessagePart, recoverStatuses: ReadonlySet<string> | undefined): boolean {
if (part.type !== "tool_use" || !isValidToolUseID(part.id)) {
return false
}
if (!recoverStatuses) {
return true
}
const status = part.state?.status
return typeof status === "string" && recoverStatuses.has(status)
}
function extractToolUseIds(parts: MessagePart[], recoverStatuses?: ReadonlySet<string>): string[] {
return parts
.filter((part): part is ToolUsePart => shouldRecoverToolUsePart(part, recoverStatuses))
.map((part) => part.id)
}
async function readPartsFromSDKFallback(
@@ -85,9 +109,12 @@ export async function recoverToolResultMissing(
client: Client,
sessionID: string,
failedAssistantMsg: MessageData,
resumeConfig?: ResumeConfig
resumeConfig?: ResumeConfig,
options?: RecoverToolResultMissingOptions,
): Promise<boolean> {
let parts = failedAssistantMsg.parts || []
let parts = (failedAssistantMsg.parts || [])
.map((part) => normalizeMessagePart(part))
.filter((part): part is MessagePart => part !== null)
if (parts.length === 0 && failedAssistantMsg.info?.id) {
if (isSqliteBackend()) {
parts = await readPartsFromSDKFallback(client, sessionID, failedAssistantMsg.info.id)
@@ -97,17 +124,18 @@ export async function recoverToolResultMissing(
}
}
const toolUseIds = extractToolUseIds(parts)
const toolUseIds = extractToolUseIds(parts, options?.recoverStatuses)
if (toolUseIds.length === 0) {
return false
}
const resultText = options?.resultText ?? "Operation cancelled by user (ESC pressed)"
const toolResultParts = toolUseIds.map((id) => ({
type: "tool_result" as const,
toolUseId: id,
tool_use_id: id,
isError: true,
content: [{ type: "text" as const, text: "Operation cancelled by user (ESC pressed)" }],
content: [{ type: "text" as const, text: resultText }],
}))
const launchAgent = resumeConfig?.agent
@@ -134,7 +162,7 @@ export async function recoverToolResultMissing(
const promptResult = await promptAsyncAfterSessionIdle({
client,
sessionID,
source: "session-recovery-tool-result-missing",
source: options?.source ?? "session-recovery-tool-result-missing",
input: promptInput,
})
src/hooks/session-recovery/types.ts
+11
View File
@@ -69,6 +69,11 @@ export interface MessageData {
sessionID?: string
parentID?: string
error?: unknown
finish?: unknown
time?: {
created?: unknown
completed?: unknown
}
agent?: string
model?: {
providerID: string
@@ -87,6 +92,12 @@ export interface MessageData {
name?: string
input?: Record<string, unknown>
callID?: string
state?: {
status?: unknown
input?: Record<string, unknown>
output?: unknown
error?: unknown
}
}>
}
src/plugin/event.test.ts
+35
View File
@@ -373,6 +373,41 @@ describe("createEventHandler - idle deduplication", () => {
expect((dispatchCalls[1]?.event.properties as { sessionID?: string } | undefined)?.sessionID).toBe(sessionId)
})
it("#given idle recovery handles an interrupted tool turn #when session.idle arrives #then later idle hooks are skipped for that event", async () => {
const callOrder: string[] = []
const eventHandler = createEventHandler({
ctx: asEventHandlerContext({ directory: "/tmp" }),
pluginConfig: asPluginConfig({}),
firstMessageVariantGate: {
markSessionCreated: () => {},
clear: () => {},
},
managers: createEventHandlerManagers(),
hooks: createEventHandlerHooks({
sessionRecovery: {
handleInterruptedToolResultsOnIdle: async () => {
callOrder.push("sessionRecovery")
return true
},
},
todoContinuationEnforcer: {
handler: async () => {
callOrder.push("todoContinuationEnforcer")
},
},
}),
})
await eventHandler(asEventHandlerInput({
event: {
type: "session.idle",
properties: { sessionID: "ses_interrupted_idle" },
},
}))
expect(callOrder).toEqual(["sessionRecovery"])
})
it("keeps other session dedup state untouched when bypassing synthetic-idle for current session", async () => {
//#given
const originalDateNow = Date.now
src/plugin/event.ts
+10
View File
@@ -571,6 +571,16 @@ export function createEventHandler(args: {
}
}
if (input.event.type === "session.idle") {
const sessionID = getEventSessionID(input);
if (sessionID && hooks.sessionRecovery?.handleInterruptedToolResultsOnIdle) {
const recovered = await hooks.sessionRecovery.handleInterruptedToolResultsOnIdle(sessionID);
if (recovered) {
return;
}
}
}
await dispatchToHooks(input);
const syntheticIdle = normalizeSessionStatusToIdle(input);