fix(recovery): avoid duplicate continuation prompts
This commit is contained in:
@@ -196,7 +196,7 @@ describe("createCompactionContextInjector recovery", () => {
|
||||
expect(recoveryPart?.metadata).toEqual({ compaction_continue: true })
|
||||
})
|
||||
|
||||
it("retries recovery when the recovered prompt config still mismatches expected model or tools", async () => {
|
||||
it("does not immediately retry recovery when the recovered prompt config still mismatches expected model or tools", async () => {
|
||||
//#given
|
||||
const promptAsyncRecorder = createPromptAsyncRecorder()
|
||||
const mismatchResponse = [
|
||||
@@ -247,7 +247,61 @@ describe("createCompactionContextInjector recovery", () => {
|
||||
})
|
||||
|
||||
//#then
|
||||
expect(promptAsyncRecorder.calls.length).toBe(2)
|
||||
expect(promptAsyncRecorder.calls.length).toBe(1)
|
||||
})
|
||||
|
||||
it("#given post-dispatch config read is stale #when a second compaction event arrives immediately #then recovery prompt is not duplicated", async () => {
|
||||
//#given
|
||||
const promptAsyncRecorder = createPromptAsyncRecorder()
|
||||
const checkpointedPromptConfig = [
|
||||
{
|
||||
info: {
|
||||
role: "user",
|
||||
agent: "atlas",
|
||||
model: { providerID: "openai", modelID: "gpt-5" },
|
||||
tools: { bash: true },
|
||||
},
|
||||
},
|
||||
]
|
||||
const incompletePromptConfig = [
|
||||
{
|
||||
info: {
|
||||
role: "user",
|
||||
agent: "atlas",
|
||||
model: { providerID: "openai", modelID: "gpt-5" },
|
||||
},
|
||||
},
|
||||
]
|
||||
const ctx = createMockContext(
|
||||
[
|
||||
checkpointedPromptConfig,
|
||||
incompletePromptConfig,
|
||||
incompletePromptConfig,
|
||||
incompletePromptConfig,
|
||||
incompletePromptConfig,
|
||||
incompletePromptConfig,
|
||||
],
|
||||
promptAsyncRecorder.promptAsync,
|
||||
)
|
||||
const injector = createCompactionContextInjector({ ctx })
|
||||
|
||||
//#when
|
||||
await injector.capture("ses_stale_recovery_read")
|
||||
await injector.event({
|
||||
event: {
|
||||
type: "session.compacted",
|
||||
properties: { sessionID: "ses_stale_recovery_read" },
|
||||
},
|
||||
})
|
||||
await injector.event({
|
||||
event: {
|
||||
type: "session.compacted",
|
||||
properties: { sessionID: "ses_stale_recovery_read" },
|
||||
},
|
||||
})
|
||||
|
||||
//#then
|
||||
expect(promptAsyncRecorder.calls.length).toBe(1)
|
||||
})
|
||||
|
||||
it("does not treat reasoning-only assistant messages as a no-text tail", async () => {
|
||||
|
||||
@@ -21,7 +21,7 @@ import {
|
||||
import { AGENT_RECOVERY_PROMPT, NO_TEXT_TAIL_THRESHOLD, RECOVERY_COOLDOWN_MS, RECENT_COMPACTION_WINDOW_MS } from "./constants"
|
||||
import type { CompactionContextClient } from "./types"
|
||||
import type { TailMonitorState } from "./tail-monitor"
|
||||
import { dispatchInternalPrompt, releasePromptAsyncReservation } from "../shared/prompt-async-gate"
|
||||
import { dispatchInternalPrompt } from "../shared/prompt-async-gate"
|
||||
|
||||
export function createRecoveryLogic(
|
||||
ctx: CompactionContextClient | undefined,
|
||||
@@ -107,6 +107,7 @@ export function createRecoveryLogic(
|
||||
})
|
||||
return false
|
||||
}
|
||||
tailState.lastRecoveryAt = now
|
||||
|
||||
const recoveredPromptConfig = await resolveLatestSessionPromptConfig(ctx, sessionID)
|
||||
if (!isPromptConfigRecovered(recoveredPromptConfig, expectedPromptConfig)) {
|
||||
@@ -118,9 +119,6 @@ export function createRecoveryLogic(
|
||||
hasTools: !!tools,
|
||||
recoveredPromptConfig,
|
||||
})
|
||||
releasePromptAsyncReservation(sessionID, "compaction-context-injector:incomplete-recovery", {
|
||||
reservedBy: "compaction-context-injector",
|
||||
})
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -132,7 +130,6 @@ export function createRecoveryLogic(
|
||||
setSessionTools(sessionID, tools)
|
||||
}
|
||||
|
||||
tailState.lastRecoveryAt = now
|
||||
tailState.consecutiveNoTextMessages = 0
|
||||
|
||||
log(`[compaction-context-injector] Re-injected checkpointed agent config`, {
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
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"
|
||||
import { dispatchInternalPrompt, releaseAllPromptAsyncReservationsForTesting } from "../../shared/prompt-async-gate"
|
||||
|
||||
type RecoverableInfo = Parameters<ReturnType<typeof createSessionRecoveryHook>["handleSessionRecovery"]>[0]
|
||||
|
||||
@@ -103,6 +103,86 @@ describe("session-recovery hook persistent dedupe", () => {
|
||||
expect(result).toBe(false)
|
||||
expect(counts.abort).toBe(1)
|
||||
})
|
||||
|
||||
test("#given recovery is blocked by a peer prompt reservation #when the same error is observed after the reservation clears #then recovery retries once", async () => {
|
||||
// given
|
||||
const sessionID = "ses_recovery_gate_block"
|
||||
const promptAsyncCalls: PromptAsyncCall[] = []
|
||||
let releasePeerPrompt: (() => void) | undefined
|
||||
const peerPrompt = new Promise<void>((resolve) => {
|
||||
releasePeerPrompt = resolve
|
||||
})
|
||||
const peerReservation = dispatchInternalPrompt({
|
||||
mode: "async",
|
||||
client: {
|
||||
session: {
|
||||
promptAsync: async () => {
|
||||
await peerPrompt
|
||||
},
|
||||
},
|
||||
},
|
||||
sessionID,
|
||||
input: { path: { id: sessionID }, body: { parts: [{ type: "text", text: "peer" }] } },
|
||||
source: "test:peer-recovery-blocker",
|
||||
settleMs: 0,
|
||||
})
|
||||
await Promise.resolve()
|
||||
|
||||
const info: RecoverableInfo = {
|
||||
id: "msg_tool_missing",
|
||||
role: "assistant",
|
||||
sessionID,
|
||||
error: { message: "messages.2 has tool_use without a matching tool_result" },
|
||||
}
|
||||
const ctx = {
|
||||
client: {
|
||||
session: {
|
||||
abort: async () => ({}),
|
||||
messages: async () => ({
|
||||
data: [
|
||||
{
|
||||
info: {
|
||||
id: info.id,
|
||||
role: "assistant",
|
||||
error: info.error,
|
||||
},
|
||||
parts: [
|
||||
{
|
||||
type: "tool_use",
|
||||
id: "toolu_recovery_gate",
|
||||
name: "bash",
|
||||
input: {},
|
||||
state: { status: "running" },
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
}),
|
||||
promptAsync: async (call: PromptAsyncCall) => {
|
||||
promptAsyncCalls.push(call)
|
||||
return {}
|
||||
},
|
||||
},
|
||||
tui: {
|
||||
showToast: async () => ({}),
|
||||
},
|
||||
},
|
||||
directory: "/tmp/session-recovery-gate-test",
|
||||
}
|
||||
const hook = createSessionRecoveryHook(ctx as never)
|
||||
|
||||
// when
|
||||
const firstResult = await hook.handleSessionRecovery(info)
|
||||
releasePeerPrompt?.()
|
||||
await peerReservation
|
||||
releaseAllPromptAsyncReservationsForTesting()
|
||||
const secondResult = await hook.handleSessionRecovery(info)
|
||||
|
||||
// then
|
||||
expect(firstResult).toBe(false)
|
||||
expect(secondResult).toBe(true)
|
||||
expect(promptAsyncCalls).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe("session-recovery hook interrupted idle recovery", () => {
|
||||
|
||||
@@ -191,6 +191,7 @@ export function createSessionRecoveryHook(ctx: PluginInput, options?: SessionRec
|
||||
if (!assistantMsgID) return false
|
||||
if (processingErrors.has(assistantMsgID)) return false
|
||||
processingErrors.add(assistantMsgID)
|
||||
let shouldKeepProcessingError = false
|
||||
|
||||
try {
|
||||
if (onAbortCallback) {
|
||||
@@ -268,21 +269,21 @@ export function createSessionRecoveryHook(ctx: PluginInput, options?: SessionRec
|
||||
await resumeSession(ctx.client, resumeConfig)
|
||||
}
|
||||
} else if (errorType === "assistant_prefill_unsupported") {
|
||||
shouldKeepProcessingError = true
|
||||
success = false
|
||||
}
|
||||
|
||||
if (success) {
|
||||
shouldKeepProcessingError = true
|
||||
}
|
||||
return success
|
||||
} catch (err) {
|
||||
log("[session-recovery] Recovery failed:", err)
|
||||
return false
|
||||
} finally {
|
||||
// Keep assistantMsgID in processingErrors permanently so that a
|
||||
// stale duplicate session.error for the SAME assistant message
|
||||
// does not retrigger recovery (and a second resumeSession
|
||||
// promptAsync injection) after the first attempt resolves.
|
||||
// Successful recovery starts a new assistant message on the next
|
||||
// turn with a different id, so this dedupe never blocks future
|
||||
// legitimate errors.
|
||||
if (!shouldKeepProcessingError) {
|
||||
processingErrors.delete(assistantMsgID)
|
||||
}
|
||||
if (sessionID && onRecoveryCompleteCallback) {
|
||||
onRecoveryCompleteCallback(sessionID)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user