Merge pull request #4106 from code-yeongyu/code-yeongyu/fix-stale-tool-hang
fix(session-recovery): recover interrupted idle tool turns
This commit is contained in:
YeonGyu-Kim
@@ -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
|
||||
|
||||
@@ -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 }> = {
|
||||
|
||||
@@ -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,
|
||||
}))
|
||||
|
||||
|
||||
@@ -1,8 +1,28 @@
|
||||
import { describe, expect, test } from "bun:test"
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
import { createSessionRecoveryHook } from "./hook"
|
||||
import { _setInterruptedIdleMessagesFetchTimeoutMsForTesting } from "./interrupted-idle-message-fetch-timeout"
|
||||
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()
|
||||
_setInterruptedIdleMessagesFetchTimeoutMsForTesting(undefined)
|
||||
})
|
||||
|
||||
function createPrefillErrorInfo(): RecoverableInfo {
|
||||
return {
|
||||
id: "msg_failed_prefill",
|
||||
@@ -84,3 +104,110 @@ 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",
|
||||
finish: "tool-calls",
|
||||
time: { created: 1778995446058, completed: 1778995447058 },
|
||||
},
|
||||
parts: [
|
||||
{
|
||||
type: "tool",
|
||||
callID: "call_completed",
|
||||
name: "bash",
|
||||
input: {},
|
||||
state: { status: "completed" },
|
||||
},
|
||||
{
|
||||
type: "tool_use",
|
||||
id: "toolu_running",
|
||||
callID: "prt_not_a_tool_use_id",
|
||||
name: "bash",
|
||||
input: {},
|
||||
state: { status: "running" },
|
||||
},
|
||||
{
|
||||
type: "tool_use",
|
||||
id: "toolu_pending",
|
||||
callID: "also_not_a_tool_use_id",
|
||||
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")
|
||||
})
|
||||
|
||||
test("#given session.messages hangs during idle recovery #when timeout elapses #then idle recovery returns false", async () => {
|
||||
// given
|
||||
_setInterruptedIdleMessagesFetchTimeoutMsForTesting(5)
|
||||
const ctx = {
|
||||
client: {
|
||||
session: {
|
||||
messages: async () => new Promise(() => {}),
|
||||
promptAsync: async () => ({}),
|
||||
},
|
||||
},
|
||||
directory: "/tmp/session-recovery-timeout-test",
|
||||
}
|
||||
const hook = createSessionRecoveryHook(ctx as never)
|
||||
|
||||
// when
|
||||
const result = await hook.handleInterruptedToolResultsOnIdle("ses_messages_hangs")
|
||||
|
||||
// then
|
||||
expect(result).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -4,6 +4,11 @@ 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 {
|
||||
getInterruptedIdleMessagesFetchTimeoutMs,
|
||||
withInterruptedIdleMessagesFetchTimeout,
|
||||
} from "./interrupted-idle-message-fetch-timeout"
|
||||
import { recoverToolResultMissing } from "./recover-tool-result-missing"
|
||||
import { recoverUnavailableTool } from "./recover-unavailable-tool"
|
||||
import { recoverThinkingBlockOrder } from "./recover-thinking-block-order"
|
||||
@@ -24,6 +29,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 +37,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 +54,111 @@ 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 (finish === "tool-calls") {
|
||||
return false
|
||||
}
|
||||
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 partHasValidToolUseID = (part: NonNullable<MessageData["parts"]>[number]): boolean => {
|
||||
const callID = part.callID
|
||||
if (typeof callID === "string" && /^(toolu_|call_)/.test(callID)) {
|
||||
return true
|
||||
}
|
||||
|
||||
const id = part.id
|
||||
return typeof id === "string" && /^(toolu_|call_)/.test(id)
|
||||
}
|
||||
|
||||
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")
|
||||
&& partHasValidToolUseID(part)
|
||||
) === 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 withInterruptedIdleMessagesFetchTimeout(
|
||||
ctx.client.session.messages({
|
||||
path: { id: sessionID },
|
||||
query: { directory: ctx.directory },
|
||||
}),
|
||||
getInterruptedIdleMessagesFetchTimeoutMs(),
|
||||
)
|
||||
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 +287,7 @@ export function createSessionRecoveryHook(ctx: PluginInput, options?: SessionRec
|
||||
|
||||
return {
|
||||
handleSessionRecovery,
|
||||
handleInterruptedToolResultsOnIdle,
|
||||
isRecoverableError,
|
||||
setOnAbortCallback,
|
||||
setOnRecoveryCompleteCallback,
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
export const DEFAULT_INTERRUPTED_IDLE_MESSAGES_FETCH_TIMEOUT_MS = 5_000
|
||||
|
||||
let interruptedIdleMessagesFetchTimeoutMsForTesting: number | undefined
|
||||
|
||||
export function _setInterruptedIdleMessagesFetchTimeoutMsForTesting(value: number | undefined): void {
|
||||
interruptedIdleMessagesFetchTimeoutMsForTesting = value
|
||||
}
|
||||
|
||||
export function getInterruptedIdleMessagesFetchTimeoutMs(): number {
|
||||
return interruptedIdleMessagesFetchTimeoutMsForTesting ?? DEFAULT_INTERRUPTED_IDLE_MESSAGES_FETCH_TIMEOUT_MS
|
||||
}
|
||||
|
||||
export class InterruptedIdleMessagesFetchTimeoutError extends Error {
|
||||
constructor(timeoutMs: number) {
|
||||
super(`[session-recovery] session.messages timed out after ${timeoutMs}ms while checking interrupted idle tools`)
|
||||
this.name = "InterruptedIdleMessagesFetchTimeoutError"
|
||||
}
|
||||
}
|
||||
|
||||
export function withInterruptedIdleMessagesFetchTimeout<T>(operation: Promise<T>, timeoutMs: number): Promise<T> {
|
||||
if (timeoutMs <= 0) {
|
||||
return operation
|
||||
}
|
||||
|
||||
let timeoutID: ReturnType<typeof globalThis.setTimeout> | undefined
|
||||
const timeoutPromise = new Promise<never>((_, reject) => {
|
||||
timeoutID = globalThis.setTimeout(
|
||||
() => reject(new InterruptedIdleMessagesFetchTimeoutError(timeoutMs)),
|
||||
timeoutMs,
|
||||
)
|
||||
})
|
||||
|
||||
return Promise.race([operation, timeoutPromise]).finally(() => {
|
||||
if (timeoutID !== undefined) {
|
||||
globalThis.clearTimeout(timeoutID)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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,89 @@ describe("recoverToolResultMissing", () => {
|
||||
})
|
||||
})
|
||||
|
||||
it("falls back to a valid id when callID is malformed", async () => {
|
||||
//#given
|
||||
const { client, promptAsync } = createMockClient()
|
||||
const failedAssistantWithMalformedCallID: MessageData = {
|
||||
info: { id: "msg_failed", role: "assistant" },
|
||||
parts: [{
|
||||
type: "tool_use",
|
||||
id: "toolu_recovered_from_id",
|
||||
callID: "prt_not_a_tool_use_id",
|
||||
state: { status: "running" },
|
||||
}],
|
||||
}
|
||||
|
||||
//#when
|
||||
const result = await recoverToolResultMissing(client, "ses_1", failedAssistantWithMalformedCallID, undefined, {
|
||||
recoverStatuses: new Set(["pending", "running"]),
|
||||
})
|
||||
|
||||
//#then
|
||||
expect(result).toBe(true)
|
||||
expect(promptAsync).toHaveBeenCalledTimes(1)
|
||||
const call = promptAsync.mock.calls[0]?.[0] as {
|
||||
body: {
|
||||
parts: Array<{ toolUseId: string }>
|
||||
}
|
||||
}
|
||||
expect(call.body.parts.map((part) => part.toolUseId)).toEqual(["toolu_recovered_from_id"])
|
||||
})
|
||||
|
||||
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: {} } }]
|
||||
|
||||
@@ -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,36 @@ 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 selectValidToolUseID(part: { id?: string; callID?: string }): string | undefined {
|
||||
if (isValidToolUseID(part.callID)) {
|
||||
return part.callID
|
||||
}
|
||||
if (isValidToolUseID(part.id)) {
|
||||
return part.id
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
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 = selectValidToolUseID(part)
|
||||
if (!toolUseID) {
|
||||
return null
|
||||
}
|
||||
|
||||
return {
|
||||
type: "tool_use",
|
||||
id: part.callID,
|
||||
id: toolUseID,
|
||||
state: part.state,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -60,8 +81,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 +119,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 +134,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,8 +172,9 @@ 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,
|
||||
checkToolState: false,
|
||||
})
|
||||
|
||||
return promptResult.status === "dispatched"
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}>
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
|
||||
import {
|
||||
_setPromptGateMessagesFetchTimeoutMsForTesting,
|
||||
promptAfterSessionIdle,
|
||||
promptAsyncAfterSessionIdle,
|
||||
releaseAllPromptAsyncReservationsForTesting,
|
||||
@@ -148,6 +149,109 @@ describe("promptAsyncAfterSessionIdle", () => {
|
||||
expect(promptCalls).toBe(0)
|
||||
})
|
||||
|
||||
test("#given latest assistant turn is waiting on tools #when an internal promptAsync is requested #then no prompt is sent", async () => {
|
||||
// given
|
||||
let promptCalls = 0
|
||||
const client = {
|
||||
session: {
|
||||
status: async () => ({ data: { ses_waiting_tools: { type: "idle" } } }),
|
||||
messages: async () => ({
|
||||
data: [
|
||||
{
|
||||
info: { id: "msg_user", role: "user" },
|
||||
parts: [{ type: "text", text: "run work" }],
|
||||
},
|
||||
{
|
||||
info: { id: "msg_assistant", role: "assistant", finish: "tool-calls" },
|
||||
parts: [{ type: "tool_use", id: "toolu_pending", state: { status: "pending" } }],
|
||||
},
|
||||
],
|
||||
}),
|
||||
promptAsync: async () => {
|
||||
promptCalls += 1
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
// when
|
||||
const result = await promptAsyncAfterSessionIdle({
|
||||
client,
|
||||
sessionID: "ses_waiting_tools",
|
||||
input: { path: { id: "ses_waiting_tools" }, body: { parts: [] } },
|
||||
source: "test:waiting-tools",
|
||||
settleMs: 0,
|
||||
postDispatchHoldMs: 0,
|
||||
})
|
||||
|
||||
// then
|
||||
expect(result.status).toBe("active")
|
||||
expect(promptCalls).toBe(0)
|
||||
})
|
||||
|
||||
test("#given latest assistant turn is waiting on tools #when tool-state check is disabled #then promptAsync is sent", async () => {
|
||||
// given
|
||||
let promptCalls = 0
|
||||
const client = {
|
||||
session: {
|
||||
status: async () => ({ data: { ses_recovery_tools: { type: "idle" } } }),
|
||||
messages: async () => ({
|
||||
data: [{
|
||||
info: { id: "msg_assistant", role: "assistant", finish: "tool-calls" },
|
||||
parts: [{ type: "tool_use", id: "toolu_pending", state: { status: "pending" } }],
|
||||
}],
|
||||
}),
|
||||
promptAsync: async () => {
|
||||
promptCalls += 1
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
// when
|
||||
const result = await promptAsyncAfterSessionIdle({
|
||||
client,
|
||||
sessionID: "ses_recovery_tools",
|
||||
input: { path: { id: "ses_recovery_tools" }, body: { parts: [] } },
|
||||
source: "test:recovery-tools",
|
||||
settleMs: 0,
|
||||
postDispatchHoldMs: 0,
|
||||
checkToolState: false,
|
||||
})
|
||||
|
||||
// then
|
||||
expect(result.status).toBe("dispatched")
|
||||
expect(promptCalls).toBe(1)
|
||||
})
|
||||
|
||||
test("#given latest-message fetch hangs #when an internal promptAsync is requested #then the tool-state check times out and dispatch continues", async () => {
|
||||
// given
|
||||
_setPromptGateMessagesFetchTimeoutMsForTesting(5)
|
||||
let promptCalls = 0
|
||||
const client = {
|
||||
session: {
|
||||
status: async () => ({ data: { ses_messages_hang: { type: "idle" } } }),
|
||||
messages: async () => new Promise(() => {}),
|
||||
promptAsync: async () => {
|
||||
promptCalls += 1
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
// when
|
||||
const result = await promptAsyncAfterSessionIdle({
|
||||
client,
|
||||
sessionID: "ses_messages_hang",
|
||||
input: { path: { id: "ses_messages_hang" }, body: { parts: [] } },
|
||||
source: "test:messages-hang",
|
||||
settleMs: 0,
|
||||
postDispatchHoldMs: 0,
|
||||
dispatchTimeoutMs: 50,
|
||||
})
|
||||
|
||||
// then
|
||||
expect(result.status).toBe("dispatched")
|
||||
expect(promptCalls).toBe(1)
|
||||
})
|
||||
|
||||
test("#given dispatch hold has expired #when the same session prompts again #then the next promptAsync is accepted", async () => {
|
||||
// given
|
||||
let promptCalls = 0
|
||||
|
||||
@@ -373,6 +373,81 @@ 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("#given idle recovery handles an interrupted tool turn #when session.status normalizes to idle #then synthetic idle hooks are skipped", 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 (input: EventInput) => {
|
||||
if (input.event.type === "session.idle") {
|
||||
callOrder.push("todoContinuationEnforcer")
|
||||
}
|
||||
},
|
||||
},
|
||||
}),
|
||||
})
|
||||
|
||||
await eventHandler(asEventHandlerInput({
|
||||
event: {
|
||||
type: "session.status",
|
||||
properties: {
|
||||
sessionID: "ses_interrupted_status_idle",
|
||||
status: { type: "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
|
||||
|
||||
+34
-11
@@ -377,6 +377,19 @@ export function createEventHandler(args: {
|
||||
return true;
|
||||
};
|
||||
|
||||
const recoverInterruptedToolResultsOnIdleEvent = async (input: EventInput): Promise<boolean> => {
|
||||
if (input.event.type !== "session.idle") {
|
||||
return false;
|
||||
}
|
||||
|
||||
const sessionID = getEventSessionID(input);
|
||||
if (!sessionID || !hooks.sessionRecovery?.handleInterruptedToolResultsOnIdle) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return hooks.sessionRecovery.handleInterruptedToolResultsOnIdle(sessionID);
|
||||
};
|
||||
|
||||
const getFallbackContinuationKeys = (fallbackContext?: FallbackContinuationContext): FallbackContinuationDedupeKeys => {
|
||||
const agentKey = fallbackContext?.agentName
|
||||
? getAgentConfigKey(fallbackContext.agentName).trim().toLowerCase()
|
||||
@@ -571,6 +584,13 @@ export function createEventHandler(args: {
|
||||
}
|
||||
}
|
||||
|
||||
if (input.event.type === "session.idle") {
|
||||
const recovered = await recoverInterruptedToolResultsOnIdleEvent(input);
|
||||
if (recovered) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
await dispatchToHooks(input);
|
||||
|
||||
const syntheticIdle = normalizeSessionStatusToIdle(input);
|
||||
@@ -586,17 +606,20 @@ export function createEventHandler(args: {
|
||||
if (!shouldDispatchIdleEvent(sessionID, now)) {
|
||||
return;
|
||||
}
|
||||
await dispatchToHooks(syntheticIdle as EventInput);
|
||||
if (pluginConfig.openclaw) {
|
||||
await dispatchOpenClawEvent({
|
||||
config: pluginConfig.openclaw,
|
||||
rawEvent: "session.idle",
|
||||
context: {
|
||||
sessionId: sessionID,
|
||||
projectPath: pluginContext.directory,
|
||||
tmuxPaneId: managers.tmuxSessionManager.getTrackedPaneId?.(sessionID) ?? process.env.TMUX_PANE,
|
||||
},
|
||||
});
|
||||
const recovered = await recoverInterruptedToolResultsOnIdleEvent(syntheticIdle as EventInput);
|
||||
if (!recovered) {
|
||||
await dispatchToHooks(syntheticIdle as EventInput);
|
||||
if (pluginConfig.openclaw) {
|
||||
await dispatchOpenClawEvent({
|
||||
config: pluginConfig.openclaw,
|
||||
rawEvent: "session.idle",
|
||||
context: {
|
||||
sessionId: sessionID,
|
||||
projectPath: pluginContext.directory,
|
||||
tmuxPaneId: managers.tmuxSessionManager.getTrackedPaneId?.(sessionID) ?? process.env.TMUX_PANE,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@ import {
|
||||
|
||||
export const DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS = 250
|
||||
export const DEFAULT_PROMPT_DISPATCH_TIMEOUT_MS = 30_000
|
||||
export const DEFAULT_PROMPT_GATE_MESSAGES_FETCH_TIMEOUT_MS = 5_000
|
||||
|
||||
type PromptAsyncInput = {
|
||||
path?: { id?: string }
|
||||
@@ -16,9 +17,15 @@ type PromptAsyncInput = {
|
||||
[key: string]: unknown
|
||||
}
|
||||
|
||||
type PromptMessagesQuery = {
|
||||
directory: string
|
||||
limit?: number
|
||||
}
|
||||
|
||||
type PromptAsyncClient<TInput> = {
|
||||
session?: {
|
||||
status?: () => Promise<unknown>
|
||||
messages?: (input: { path: { id: string }; query: PromptMessagesQuery }) => Promise<unknown>
|
||||
promptAsync?: (input: TInput) => Promise<unknown>
|
||||
}
|
||||
}
|
||||
@@ -26,6 +33,7 @@ type PromptAsyncClient<TInput> = {
|
||||
type PromptClient<TInput> = {
|
||||
session?: {
|
||||
status?: () => Promise<unknown>
|
||||
messages?: (input: { path: { id: string }; query: PromptMessagesQuery }) => Promise<unknown>
|
||||
prompt?: (input: TInput) => Promise<unknown>
|
||||
}
|
||||
}
|
||||
@@ -40,6 +48,8 @@ type PromptAsyncReservation = {
|
||||
declare function setTimeout(callback: () => void, delay?: number): ReturnType<typeof globalThis.setTimeout>
|
||||
declare function clearTimeout(timeout: ReturnType<typeof globalThis.setTimeout>): void
|
||||
|
||||
let promptGateMessagesFetchTimeoutMsForTesting: number | undefined
|
||||
|
||||
export type PromptAsyncGateResult =
|
||||
| { status: "dispatched"; response: unknown }
|
||||
| { status: "active" }
|
||||
@@ -54,6 +64,14 @@ type PromptAsyncReservationReleaseOptions = {
|
||||
|
||||
const promptAsyncReservations = new Map<string, PromptAsyncReservation>()
|
||||
|
||||
export function _setPromptGateMessagesFetchTimeoutMsForTesting(value: number | undefined): void {
|
||||
promptGateMessagesFetchTimeoutMsForTesting = value
|
||||
}
|
||||
|
||||
function getPromptGateMessagesFetchTimeoutMs(): number {
|
||||
return promptGateMessagesFetchTimeoutMsForTesting ?? DEFAULT_PROMPT_GATE_MESSAGES_FETCH_TIMEOUT_MS
|
||||
}
|
||||
|
||||
function pruneExpiredReservations(now = Date.now()): void {
|
||||
for (const [sessionID, reservation] of promptAsyncReservations) {
|
||||
if (typeof reservation.expiresAt === "number" && reservation.expiresAt <= now) {
|
||||
@@ -119,9 +137,120 @@ async function withDispatchTimeout<T>(
|
||||
}
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null
|
||||
}
|
||||
|
||||
function getPromptQuery(input: unknown): PromptMessagesQuery {
|
||||
if (!isRecord(input)) {
|
||||
return { directory: "" }
|
||||
}
|
||||
const query = input.query
|
||||
if (!isRecord(query)) {
|
||||
return { directory: "" }
|
||||
}
|
||||
|
||||
const promptQuery: PromptMessagesQuery = { directory: "" }
|
||||
if (typeof query.directory === "string") {
|
||||
promptQuery.directory = query.directory
|
||||
}
|
||||
if (typeof query.limit === "number") {
|
||||
promptQuery.limit = query.limit
|
||||
}
|
||||
return promptQuery
|
||||
}
|
||||
|
||||
function getMessagesData(response: unknown): unknown[] {
|
||||
if (isRecord(response) && Array.isArray(response.data)) {
|
||||
return response.data
|
||||
}
|
||||
return Array.isArray(response) ? response : []
|
||||
}
|
||||
|
||||
function messageRole(message: unknown): string | undefined {
|
||||
if (!isRecord(message)) {
|
||||
return undefined
|
||||
}
|
||||
const info = message.info
|
||||
if (isRecord(info) && typeof info.role === "string") {
|
||||
return info.role
|
||||
}
|
||||
return typeof message.role === "string" ? message.role : undefined
|
||||
}
|
||||
|
||||
function partIsWaitingOnTool(part: unknown): boolean {
|
||||
if (!isRecord(part)) {
|
||||
return false
|
||||
}
|
||||
if (part.type !== "tool" && part.type !== "tool_use") {
|
||||
return false
|
||||
}
|
||||
|
||||
const state = part.state
|
||||
if (!isRecord(state)) {
|
||||
return false
|
||||
}
|
||||
return state.status === "pending" || state.status === "running"
|
||||
}
|
||||
|
||||
function latestAssistantTurnIsWaitingOnTools(messages: unknown[]): boolean {
|
||||
for (let index = messages.length - 1; index >= 0; index--) {
|
||||
const message = messages[index]
|
||||
const role = messageRole(message)
|
||||
if (role === "assistant") {
|
||||
if (!isRecord(message) || !Array.isArray(message.parts)) {
|
||||
return false
|
||||
}
|
||||
return message.parts.some(partIsWaitingOnTool)
|
||||
}
|
||||
if (role === "user") {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
async function sessionLatestAssistantIsWaitingOnTools<TInput>(args: {
|
||||
client: { session?: { messages?: (input: { path: { id: string }; query: PromptMessagesQuery }) => Promise<unknown> } }
|
||||
sessionID: string
|
||||
input: TInput
|
||||
sessionName: "promptAsync" | "prompt"
|
||||
source: string
|
||||
timeoutMs: number
|
||||
}): Promise<boolean> {
|
||||
const messages = args.client.session?.messages
|
||||
if (typeof messages !== "function") {
|
||||
return false
|
||||
}
|
||||
|
||||
try {
|
||||
const response = await withDispatchTimeout(
|
||||
messages({
|
||||
path: { id: args.sessionID },
|
||||
query: getPromptQuery(args.input),
|
||||
}),
|
||||
args.timeoutMs,
|
||||
`[prompt-async-gate] ${args.sessionName} session.messages`,
|
||||
)
|
||||
return latestAssistantTurnIsWaitingOnTools(getMessagesData(response))
|
||||
} catch (error) {
|
||||
log("[prompt-async-gate] latest assistant tool-state check failed", {
|
||||
sessionID: args.sessionID,
|
||||
source: args.source,
|
||||
error: String(error),
|
||||
})
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
async function dispatchAfterSessionIdle<TInput>(args: {
|
||||
sessionName: "promptAsync" | "prompt"
|
||||
client: { session?: { status?: () => Promise<unknown> } }
|
||||
client: {
|
||||
session?: {
|
||||
status?: () => Promise<unknown>
|
||||
messages?: (input: { path: { id: string }; query: PromptMessagesQuery }) => Promise<unknown>
|
||||
}
|
||||
}
|
||||
sessionID: string
|
||||
input: TInput
|
||||
source: string
|
||||
@@ -129,6 +258,7 @@ async function dispatchAfterSessionIdle<TInput>(args: {
|
||||
postDispatchHoldMs: number
|
||||
dispatchTimeoutMs: number
|
||||
checkStatus: boolean
|
||||
checkToolState: boolean
|
||||
dispatch: (input: TInput) => Promise<unknown>
|
||||
}): Promise<PromptAsyncGateResult> {
|
||||
const {
|
||||
@@ -141,6 +271,7 @@ async function dispatchAfterSessionIdle<TInput>(args: {
|
||||
postDispatchHoldMs,
|
||||
dispatchTimeoutMs,
|
||||
checkStatus,
|
||||
checkToolState,
|
||||
dispatch,
|
||||
} = args
|
||||
|
||||
@@ -186,6 +317,25 @@ async function dispatchAfterSessionIdle<TInput>(args: {
|
||||
return { status: "active" }
|
||||
}
|
||||
|
||||
if (
|
||||
checkToolState
|
||||
&& typeof client.session?.messages === "function"
|
||||
&& await sessionLatestAssistantIsWaitingOnTools({
|
||||
client,
|
||||
sessionID,
|
||||
input,
|
||||
sessionName,
|
||||
source,
|
||||
timeoutMs: Math.min(dispatchTimeoutMs, getPromptGateMessagesFetchTimeoutMs()),
|
||||
})
|
||||
) {
|
||||
log(`[prompt-async-gate] ${sessionName} skipped because latest assistant is waiting on tools`, {
|
||||
sessionID,
|
||||
source,
|
||||
})
|
||||
return { status: "active" }
|
||||
}
|
||||
|
||||
log(`[prompt-async-gate] ${sessionName} dispatching`, { sessionID, source })
|
||||
dispatchAttempted = true
|
||||
const response = await withDispatchTimeout(
|
||||
@@ -219,6 +369,7 @@ export async function promptAsyncAfterSessionIdle<TInput = PromptAsyncInput>(arg
|
||||
postDispatchHoldMs?: number
|
||||
dispatchTimeoutMs?: number
|
||||
checkStatus?: boolean
|
||||
checkToolState?: boolean
|
||||
}): Promise<PromptAsyncGateResult> {
|
||||
const {
|
||||
client,
|
||||
@@ -247,6 +398,7 @@ export async function promptAsyncAfterSessionIdle<TInput = PromptAsyncInput>(arg
|
||||
postDispatchHoldMs,
|
||||
dispatchTimeoutMs,
|
||||
checkStatus: args.checkStatus !== false,
|
||||
checkToolState: args.checkToolState !== false,
|
||||
dispatch: (dispatchInput) => dispatchPromptAsync(dispatchInput),
|
||||
})
|
||||
}
|
||||
@@ -260,6 +412,7 @@ export async function promptAfterSessionIdle<TInput = PromptAsyncInput>(args: {
|
||||
postDispatchHoldMs?: number
|
||||
dispatchTimeoutMs?: number
|
||||
checkStatus?: boolean
|
||||
checkToolState?: boolean
|
||||
}): Promise<PromptAsyncGateResult> {
|
||||
const {
|
||||
client,
|
||||
@@ -288,12 +441,14 @@ export async function promptAfterSessionIdle<TInput = PromptAsyncInput>(args: {
|
||||
postDispatchHoldMs,
|
||||
dispatchTimeoutMs,
|
||||
checkStatus: args.checkStatus !== false,
|
||||
checkToolState: args.checkToolState !== false,
|
||||
dispatch: (dispatchInput) => dispatchPrompt(dispatchInput),
|
||||
})
|
||||
}
|
||||
|
||||
export function releaseAllPromptAsyncReservationsForTesting(): void {
|
||||
promptAsyncReservations.clear()
|
||||
promptGateMessagesFetchTimeoutMsForTesting = undefined
|
||||
}
|
||||
|
||||
export function releasePromptAsyncReservation(
|
||||
|
||||
Reference in New Issue
Block a user