From 519a135f02a4132d33a09bbfb38ff1b2475f6c7b Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 22:21:08 +0000 Subject: [PATCH 1/4] fix(chat): stop poison-turn checkpoint retry storms Redelivery against a still-running turn reused a Date.now() prompt and failed commitMessages' append-only deep-equal check. That mismatch was swallowed as TurnInputCommitLost and mapped to lost_lease recovery, requeuing forever. Reuse the exact durable checkpointed prompt on running-record replay, throw a typed AgentHistoryBoundaryError for permanent shape mismatches, and fail closed without lost_lease recovery wakes. Evidence: JUNIOR-62 / JUNIOR-7A storm on slack:C0B595QDZLL:1785871561.942119 Co-Authored-By: David Cramer --- .../junior/src/chat/agent-dispatch/work.ts | 8 +++++ packages/junior/src/chat/agent/index.ts | 34 +++++++++++++++---- packages/junior/src/chat/agent/prompt.ts | 31 ++++++++++++++--- packages/junior/src/chat/agent/resume.ts | 2 ++ .../src/chat/conversations/projection.ts | 3 +- packages/junior/src/chat/runtime/turn.ts | 27 +++++++++++++++ .../src/chat/services/turn-session-record.ts | 21 ++++++++++++ .../src/chat/task-execution/slack-work.ts | 11 ++++++ .../junior/src/chat/task-execution/worker.ts | 23 +++++++++++++ .../runtime/agent-run-provider-retry.test.ts | 11 ++++++ 10 files changed, 158 insertions(+), 13 deletions(-) diff --git a/packages/junior/src/chat/agent-dispatch/work.ts b/packages/junior/src/chat/agent-dispatch/work.ts index 8054edece..5a7d02229 100644 --- a/packages/junior/src/chat/agent-dispatch/work.ts +++ b/packages/junior/src/chat/agent-dispatch/work.ts @@ -11,6 +11,8 @@ import type { ConversationStore } from "@/chat/conversations/store"; import { getConversationStore } from "@/chat/db"; import { getConversationTurnBoundaryError, + isAgentHistoryBoundaryError, + isAgentHistoryBoundaryMessage, isCooperativeTurnYieldError, isTurnInputCommitLostError, TurnInputCommitLostError, @@ -523,6 +525,12 @@ export function createAgentDispatchConversationWorker( await markDispatchAwaitingResume(dispatch.id); return { status: "yielded" }; } + if ( + isAgentHistoryBoundaryError(error) || + isAgentHistoryBoundaryMessage(error) + ) { + throw error; + } if (isTurnInputCommitLostError(error)) { return { status: "lost_lease" }; } diff --git a/packages/junior/src/chat/agent/index.ts b/packages/junior/src/chat/agent/index.ts index ffee90649..398c05f9b 100644 --- a/packages/junior/src/chat/agent/index.ts +++ b/packages/junior/src/chat/agent/index.ts @@ -71,7 +71,11 @@ import { } from "@/chat/pi/transcript"; import { createTracedStreamFn } from "@/chat/pi/traced-stream"; import { shouldEmitDevAgentTrace } from "@/chat/runtime/dev-agent-trace"; -import { isTurnInputCommitLostError } from "@/chat/runtime/turn"; +import { + isAgentHistoryBoundaryError, + isAgentHistoryBoundaryMessage, + isTurnInputCommitLostError, +} from "@/chat/runtime/turn"; import type { AgentRunOutcome } from "@/chat/runtime/agent-run-outcome"; import { buildTurnResult } from "@/chat/services/turn-result"; import { decideReply } from "@/chat/services/assistant-reply"; @@ -802,6 +806,7 @@ async function executeAgentRunInPrivacyContext( inputMessages, inputMessagesAttribute, promptContentParts, + checkpointedPromptMessage, promptHistoryMessages, shouldPromptAgent, turnContexts, @@ -989,7 +994,11 @@ async function executeAgentRunInPrivacyContext( }); } } catch (error) { - if (isTurnInputCommitLostError(error)) { + if ( + isTurnInputCommitLostError(error) || + isAgentHistoryBoundaryError(error) || + isAgentHistoryBoundaryMessage(error) + ) { throw error; } logWarn("agent.turn.steering_messages_drain.failed", { @@ -1127,11 +1136,16 @@ async function executeAgentRunInPrivacyContext( "gen_ai.invoke_agent", spanContext, async () => { - const freshPromptMessage: PiMessage = { - role: "user", - content: promptContentParts, - timestamp: Date.now(), - } as PiMessage; + // Prefer the exact durable checkpointed prompt when replaying a + // still-running turn. A new Date.now() timestamp fails the + // append-only deep-equal prefix check in commitMessages. + const freshPromptMessage: PiMessage = + checkpointedPromptMessage ?? + ({ + role: "user", + content: promptContentParts, + timestamp: Date.now(), + } as PiMessage); if (shouldPromptAgent) { const promptPersisted = await runResume.requireDurableInputCheckpoint([ @@ -1510,6 +1524,12 @@ async function executeAgentRunInPrivacyContext( if (isTurnInputCommitLostError(error)) { throw error; } + if ( + isAgentHistoryBoundaryError(error) || + isAgentHistoryBoundaryMessage(error) + ) { + throw error; + } if (error instanceof AuthorizationFlowDisabledError) { throw error; } diff --git a/packages/junior/src/chat/agent/prompt.ts b/packages/junior/src/chat/agent/prompt.ts index 01a85ec2b..c27289097 100644 --- a/packages/junior/src/chat/agent/prompt.ts +++ b/packages/junior/src/chat/agent/prompt.ts @@ -68,6 +68,12 @@ export interface PromptAssembly { }>; inputMessagesAttribute: string | undefined; promptContentParts: UserContentPart[]; + /** + * Exact durable prompt message to re-checkpoint when replaying a still-running + * turn. Prefer this over synthesizing a new timestamped user message so + * commitMessages stays append-only / idempotent. + */ + checkpointedPromptMessage?: PiMessage; promptHistoryMessages: PiMessage[]; shouldPromptAgent: boolean; turnContexts: PluginTurnContext[]; @@ -377,11 +383,11 @@ function isUserContentPart(value: unknown): value is UserContentPart { // A failed input acknowledgement redelivers the same running checkpoint. // Reuse its exact prompt so plugin context cannot diverge from its durable event. -function checkpointedPromptContent(args: { +function checkpointedPromptMessage(args: { messages: PiMessage[] | undefined; turnStartMessageIndex: number | undefined; userContentParts: UserContentPart[]; -}): UserContentPart[] | undefined { +}): PiMessage | undefined { if ( !args.messages || args.turnStartMessageIndex === undefined || @@ -407,7 +413,7 @@ function checkpointedPromptContent(args: { if (!content.every(isUserContentPart)) { return undefined; } - return content; + return message as PiMessage; } /** Assemble prompt history, instructions, and telemetry input for one slice. */ @@ -461,14 +467,26 @@ export async function assemblePrompt(args: { requestContentParts, ) : args.existingSessionPiMessages!; - const replayedPromptContent = + // Redelivery against a still-running record must reuse the exact durable + // prompt message (including timestamp). A freshly synthesized Date.now() + // prompt fails commitMessages' deep-equal prefix check and storms retries. + const replayedPromptMessage = shouldPromptAgent && !args.resumedFromSessionRecord - ? checkpointedPromptContent({ + ? checkpointedPromptMessage({ messages: args.existingSessionPiMessages, turnStartMessageIndex: args.existingTurnStartMessageIndex, userContentParts: requestContentParts, }) : undefined; + const replayedPromptContent = (() => { + if (!replayedPromptMessage) { + return undefined; + } + const content = (replayedPromptMessage as { content?: unknown }).content; + return Array.isArray(content) && content.every(isUserContentPart) + ? content + : undefined; + })(); const needsBootstrapContextForPrompt = shouldPromptAgent && !replayedPromptContent && @@ -554,6 +572,9 @@ export async function assemblePrompt(args: { inputMessages, inputMessagesAttribute, promptContentParts, + ...(replayedPromptMessage + ? { checkpointedPromptMessage: replayedPromptMessage } + : {}), promptHistoryMessages, shouldPromptAgent, turnContexts: pluginUserPromptContributions.flatMap((contribution) => diff --git a/packages/junior/src/chat/agent/resume.ts b/packages/junior/src/chat/agent/resume.ts index 13091b821..017a29ada 100644 --- a/packages/junior/src/chat/agent/resume.ts +++ b/packages/junior/src/chat/agent/resume.ts @@ -176,6 +176,8 @@ export function createResumeState(args: ResumeStateArgs) { messages: PiMessage[], trailingMessageProvenance?: ConversationMessageProvenance[], ): Promise { + // Boundary mismatches rethrow from persistRunningSessionRecord so they + // never collapse into TurnInputCommitLost → lost_lease recovery. const persisted = await this.persistSafeBoundary( messages, trailingMessageProvenance, diff --git a/packages/junior/src/chat/conversations/projection.ts b/packages/junior/src/chat/conversations/projection.ts index e0c15190e..a4d8f0554 100644 --- a/packages/junior/src/chat/conversations/projection.ts +++ b/packages/junior/src/chat/conversations/projection.ts @@ -31,6 +31,7 @@ import { type PiConversationProjection, } from "@/chat/pi/conversation-events"; import { stripRuntimeTurnContext } from "@/chat/pi/transcript"; +import { AgentHistoryBoundaryError } from "@/chat/runtime/turn"; import { sanitizePostgresJson } from "@/db/postgres-json"; import type { ModelProfile } from "@/chat/model-profile"; import type { TurnReasoningLevel } from "@/chat/reasoning-level"; @@ -363,7 +364,7 @@ async function commitMessagesLocked( ...turnContextEvents, ]); } else { - throw new Error( + throw new AgentHistoryBoundaryError( `Agent history for ${args.conversationId} changed before its committed boundary`, ); } diff --git a/packages/junior/src/chat/runtime/turn.ts b/packages/junior/src/chat/runtime/turn.ts index 81dc3b399..0ff9c3086 100644 --- a/packages/junior/src/chat/runtime/turn.ts +++ b/packages/junior/src/chat/runtime/turn.ts @@ -73,6 +73,33 @@ export function isTurnInputCommitLostError( return error instanceof TurnInputCommitLostError; } +/** + * Durable agent history no longer matches the messages a turn is trying to + * checkpoint. This is a permanent shape mismatch for the current attempt, not + * a transient lease loss — callers must not requeue as lost_lease. + */ +export class AgentHistoryBoundaryError extends Error { + readonly code = "agent_history_boundary"; + + constructor(message: string) { + super(message); + this.name = "AgentHistoryBoundaryError"; + } +} + +/** Return whether an error is a permanent agent-history boundary mismatch. */ +export function isAgentHistoryBoundaryError( + error: unknown, +): error is AgentHistoryBoundaryError { + return error instanceof AgentHistoryBoundaryError; +} + +/** True when error text is the durable history-boundary mismatch message. */ +export function isAgentHistoryBoundaryMessage(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error); + return message.includes("changed before its committed boundary"); +} + /** Error indicating durable turn input should stay pending for a later worker. */ export class TurnInputDeferredError extends Error { readonly code = "turn_input_deferred"; diff --git a/packages/junior/src/chat/services/turn-session-record.ts b/packages/junior/src/chat/services/turn-session-record.ts index 7c8f82693..6faed0b19 100644 --- a/packages/junior/src/chat/services/turn-session-record.ts +++ b/packages/junior/src/chat/services/turn-session-record.ts @@ -15,6 +15,10 @@ import { isContinuablePiBoundary, trimTrailingAssistantMessages, } from "@/chat/pi/transcript"; +import { + isAgentHistoryBoundaryError, + isAgentHistoryBoundaryMessage, +} from "@/chat/runtime/turn"; import { addAgentTurnUsage, type AgentTurnUsage } from "@/chat/usage"; import { persistWithRetry } from "@/chat/services/persist-retry"; import { TurnSliceLimitExceededError } from "@/chat/services/turn-limit"; @@ -177,6 +181,23 @@ export async function persistRunningSessionRecord(args: { }); return true; } catch (recordError) { + // History-boundary mismatch is permanent for this attempt. Swallowing it as + // false promotes TurnInputCommitLost → lost_lease recovery, which requeues + // forever. Fail closed so the worker can count the attempt / dead-letter. + if ( + isAgentHistoryBoundaryError(recordError) || + isAgentHistoryBoundaryMessage(recordError) + ) { + logSessionRecordError( + recordError, + "agent.turn.running_session_record.boundary_mismatch", + args, + { + "app.ai.resume_slice_id": args.sliceId, + }, + ); + throw recordError; + } logSessionRecordError( recordError, "agent.turn.running_session_record.failed", diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index ac4ba50f4..82750367f 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -13,6 +13,8 @@ import type { SteeringCandidateMessage, } from "@/chat/runtime/slack-runtime"; import { + isAgentHistoryBoundaryError, + isAgentHistoryBoundaryMessage, isCooperativeTurnYieldError, isTurnInputDeferredError, isTurnInputCommitLostError, @@ -894,6 +896,15 @@ export function createSlackConversationWorker( if (isCooperativeTurnYieldError(error)) { return { status: "yielded" } satisfies ConversationWorkerResult; } + // History-boundary mismatch is permanent for this attempt. Do not map + // it to lost_lease recovery (infinite requeue). Let the worker catch + // count the attempt and dead-letter when retries are exhausted. + if ( + isAgentHistoryBoundaryError(error) || + isAgentHistoryBoundaryMessage(error) + ) { + throw error; + } if (isTurnInputCommitLostError(error)) { return { status: "lost_lease" } satisfies ConversationWorkerResult; } diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 79248a890..0351a03fd 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -4,6 +4,10 @@ import { getChatConfig } from "@/chat/config"; import { logException, logInfo, logWarn, withLogContext } from "@/chat/logging"; import type { ConversationStore } from "@/chat/conversations/store"; import { isProviderRetryError } from "@/chat/services/provider-error"; +import { + isAgentHistoryBoundaryError, + isAgentHistoryBoundaryMessage, +} from "@/chat/runtime/turn"; import { ConversationQueueMessageRejectedError, type ConversationQueueMessage, @@ -663,6 +667,14 @@ async function processConversationWorkInContext( // recovery nudge. Once durable recovery state is recorded and one nudge is // sent, the delivery is acknowledged; only when recording recovery state // itself fails is the error rethrown so plain redelivery retries it. + // + // History-boundary mismatches are permanent for the current attempt shape. + // Requeueing them (especially empty-attempt continue wakes) farms the same + // poison turn forever. Fail closed: count attempts when present, otherwise + // release without a recovery wake. + const permanentBoundaryMismatch = + isAgentHistoryBoundaryError(error) || + isAgentHistoryBoundaryMessage(error); let recoveryRecorded = false; try { const failure = @@ -683,6 +695,17 @@ async function processConversationWorkInContext( nowMs: errorNowMs, state: options.state, }); + } else if (permanentBoundaryMismatch) { + // Do not schedule recovery wakes for permanent shape mismatches. + // Attempts are already counted above when present; releasing without a + // wake stops the poison-turn farm (including empty-attempt continues). + await releaseConversationWork({ + conversationId, + leaseToken: lease.leaseToken, + conversationStore: options.conversationStore, + nowMs: errorNowMs, + state: options.state, + }); } else { const resumeRequested = await requestConversationContinuation({ conversationId, diff --git a/packages/junior/tests/component/runtime/agent-run-provider-retry.test.ts b/packages/junior/tests/component/runtime/agent-run-provider-retry.test.ts index 32ad917bb..d1b3319d6 100644 --- a/packages/junior/tests/component/runtime/agent-run-provider-retry.test.ts +++ b/packages/junior/tests/component/runtime/agent-run-provider-retry.test.ts @@ -1351,6 +1351,10 @@ describe("agent run continuation", () => { await executeAgentRun({ conversationId, turnId: sessionId, + // Production redelivery passes live projection history that already + // ends with the durable checkpoint. The agent must re-checkpoint that + // exact message (not a Date.now() rebuild) so commitMessages stays + // append-only. input: { messageText: "help me", piMessages: [checkpointedPrompt] }, routing: { destinationVisibility: "private", @@ -1358,6 +1362,9 @@ describe("agent run continuation", () => { source: TEST_SOURCE, actor: { platform: "slack", teamId: "T123", userId: "U123" }, }, + durability: { + onInputCommitted: async () => undefined, + }, }), ); @@ -1370,6 +1377,10 @@ describe("agent run continuation", () => { sessionRecord?.piMessages.filter((message) => message.role === "user") ?? []; expect(userMessages).toHaveLength(1); + expect(userMessages[0]).toMatchObject({ + role: "user", + timestamp: 5, + }); expect( JSON.stringify(sessionRecord?.piMessages).split("help me"), ).toHaveLength(2); From ac56663900d4e6207f04d001f15bcd44a48aba2d Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 22:36:39 +0000 Subject: [PATCH 2/4] fix(chat): resume already-checkpointed running turns A still-running turn with turnStartMessageIndex already owns its prompt. Treat that as resume so redelivery continues instead of rebuilding a Date.now() user message that fails append-only commit and farms lost_lease retries. Also stop swallowing permanent history-boundary mismatches as false. Co-Authored-By: David Cramer --- .../junior/src/chat/agent-dispatch/work.ts | 8 ---- packages/junior/src/chat/agent/index.ts | 38 +++++------------ packages/junior/src/chat/agent/prompt.ts | 41 +++++-------------- packages/junior/src/chat/agent/resume.ts | 2 - .../src/chat/conversations/projection.ts | 3 +- packages/junior/src/chat/runtime/turn.ts | 27 ------------ .../src/chat/services/turn-session-record.ts | 31 +++++++------- .../src/chat/task-execution/slack-work.ts | 11 ----- .../junior/src/chat/task-execution/worker.ts | 23 ----------- .../runtime/agent-run-provider-retry.test.ts | 6 +-- 10 files changed, 40 insertions(+), 150 deletions(-) diff --git a/packages/junior/src/chat/agent-dispatch/work.ts b/packages/junior/src/chat/agent-dispatch/work.ts index 5a7d02229..8054edece 100644 --- a/packages/junior/src/chat/agent-dispatch/work.ts +++ b/packages/junior/src/chat/agent-dispatch/work.ts @@ -11,8 +11,6 @@ import type { ConversationStore } from "@/chat/conversations/store"; import { getConversationStore } from "@/chat/db"; import { getConversationTurnBoundaryError, - isAgentHistoryBoundaryError, - isAgentHistoryBoundaryMessage, isCooperativeTurnYieldError, isTurnInputCommitLostError, TurnInputCommitLostError, @@ -525,12 +523,6 @@ export function createAgentDispatchConversationWorker( await markDispatchAwaitingResume(dispatch.id); return { status: "yielded" }; } - if ( - isAgentHistoryBoundaryError(error) || - isAgentHistoryBoundaryMessage(error) - ) { - throw error; - } if (isTurnInputCommitLostError(error)) { return { status: "lost_lease" }; } diff --git a/packages/junior/src/chat/agent/index.ts b/packages/junior/src/chat/agent/index.ts index 398c05f9b..8ebcd12d4 100644 --- a/packages/junior/src/chat/agent/index.ts +++ b/packages/junior/src/chat/agent/index.ts @@ -71,11 +71,7 @@ import { } from "@/chat/pi/transcript"; import { createTracedStreamFn } from "@/chat/pi/traced-stream"; import { shouldEmitDevAgentTrace } from "@/chat/runtime/dev-agent-trace"; -import { - isAgentHistoryBoundaryError, - isAgentHistoryBoundaryMessage, - isTurnInputCommitLostError, -} from "@/chat/runtime/turn"; +import { isTurnInputCommitLostError } from "@/chat/runtime/turn"; import type { AgentRunOutcome } from "@/chat/runtime/agent-run-outcome"; import { buildTurnResult } from "@/chat/services/turn-result"; import { decideReply } from "@/chat/services/assistant-reply"; @@ -806,7 +802,6 @@ async function executeAgentRunInPrivacyContext( inputMessages, inputMessagesAttribute, promptContentParts, - checkpointedPromptMessage, promptHistoryMessages, shouldPromptAgent, turnContexts, @@ -994,11 +989,7 @@ async function executeAgentRunInPrivacyContext( }); } } catch (error) { - if ( - isTurnInputCommitLostError(error) || - isAgentHistoryBoundaryError(error) || - isAgentHistoryBoundaryMessage(error) - ) { + if (isTurnInputCommitLostError(error)) { throw error; } logWarn("agent.turn.steering_messages_drain.failed", { @@ -1136,16 +1127,11 @@ async function executeAgentRunInPrivacyContext( "gen_ai.invoke_agent", spanContext, async () => { - // Prefer the exact durable checkpointed prompt when replaying a - // still-running turn. A new Date.now() timestamp fails the - // append-only deep-equal prefix check in commitMessages. - const freshPromptMessage: PiMessage = - checkpointedPromptMessage ?? - ({ - role: "user", - content: promptContentParts, - timestamp: Date.now(), - } as PiMessage); + const freshPromptMessage: PiMessage = { + role: "user", + content: promptContentParts, + timestamp: Date.now(), + } as PiMessage; if (shouldPromptAgent) { const promptPersisted = await runResume.requireDurableInputCheckpoint([ @@ -1155,6 +1141,10 @@ async function executeAgentRunInPrivacyContext( if (promptPersisted) { await runResume.commitInput(); } + } else if (durability.onInputCommitted) { + // Prompt already owned by the running record. Still ack mailbox + // ownership when redelivery is carrying a pending inbound. + await runResume.commitInput(); } /** Race one provider operation against the turn deadline and abort its owner. */ @@ -1524,12 +1514,6 @@ async function executeAgentRunInPrivacyContext( if (isTurnInputCommitLostError(error)) { throw error; } - if ( - isAgentHistoryBoundaryError(error) || - isAgentHistoryBoundaryMessage(error) - ) { - throw error; - } if (error instanceof AuthorizationFlowDisabledError) { throw error; } diff --git a/packages/junior/src/chat/agent/prompt.ts b/packages/junior/src/chat/agent/prompt.ts index c27289097..9fd611417 100644 --- a/packages/junior/src/chat/agent/prompt.ts +++ b/packages/junior/src/chat/agent/prompt.ts @@ -68,12 +68,6 @@ export interface PromptAssembly { }>; inputMessagesAttribute: string | undefined; promptContentParts: UserContentPart[]; - /** - * Exact durable prompt message to re-checkpoint when replaying a still-running - * turn. Prefer this over synthesizing a new timestamped user message so - * commitMessages stays append-only / idempotent. - */ - checkpointedPromptMessage?: PiMessage; promptHistoryMessages: PiMessage[]; shouldPromptAgent: boolean; turnContexts: PluginTurnContext[]; @@ -383,11 +377,11 @@ function isUserContentPart(value: unknown): value is UserContentPart { // A failed input acknowledgement redelivers the same running checkpoint. // Reuse its exact prompt so plugin context cannot diverge from its durable event. -function checkpointedPromptMessage(args: { +function checkpointedPromptContent(args: { messages: PiMessage[] | undefined; turnStartMessageIndex: number | undefined; userContentParts: UserContentPart[]; -}): PiMessage | undefined { +}): UserContentPart[] | undefined { if ( !args.messages || args.turnStartMessageIndex === undefined || @@ -413,7 +407,7 @@ function checkpointedPromptMessage(args: { if (!content.every(isUserContentPart)) { return undefined; } - return message as PiMessage; + return content; } /** Assemble prompt history, instructions, and telemetry input for one slice. */ @@ -441,11 +435,11 @@ export async function assemblePrompt(args: { userContentParts: UserContentPart[]; }): Promise { const source = args.routing.source; - const hasPromptCheckpoint = - args.resumedFromSessionRecord && - args.existingTurnStartMessageIndex !== undefined; - const shouldPromptAgent = - !args.resumedFromSessionRecord || !hasPromptCheckpoint; + // The turn-start cursor is the durable ownership signal for the prompt. + // Resume classification can lag (still-running redelivery), but once that + // cursor exists the prompt is already committed and must not be rebuilt. + const hasPromptCheckpoint = args.existingTurnStartMessageIndex !== undefined; + const shouldPromptAgent = !hasPromptCheckpoint; const requestContentParts: UserContentPart[] = [ ...(args.explicitSkill ? [ @@ -467,26 +461,14 @@ export async function assemblePrompt(args: { requestContentParts, ) : args.existingSessionPiMessages!; - // Redelivery against a still-running record must reuse the exact durable - // prompt message (including timestamp). A freshly synthesized Date.now() - // prompt fails commitMessages' deep-equal prefix check and storms retries. - const replayedPromptMessage = + const replayedPromptContent = shouldPromptAgent && !args.resumedFromSessionRecord - ? checkpointedPromptMessage({ + ? checkpointedPromptContent({ messages: args.existingSessionPiMessages, turnStartMessageIndex: args.existingTurnStartMessageIndex, userContentParts: requestContentParts, }) : undefined; - const replayedPromptContent = (() => { - if (!replayedPromptMessage) { - return undefined; - } - const content = (replayedPromptMessage as { content?: unknown }).content; - return Array.isArray(content) && content.every(isUserContentPart) - ? content - : undefined; - })(); const needsBootstrapContextForPrompt = shouldPromptAgent && !replayedPromptContent && @@ -572,9 +554,6 @@ export async function assemblePrompt(args: { inputMessages, inputMessagesAttribute, promptContentParts, - ...(replayedPromptMessage - ? { checkpointedPromptMessage: replayedPromptMessage } - : {}), promptHistoryMessages, shouldPromptAgent, turnContexts: pluginUserPromptContributions.flatMap((contribution) => diff --git a/packages/junior/src/chat/agent/resume.ts b/packages/junior/src/chat/agent/resume.ts index 017a29ada..13091b821 100644 --- a/packages/junior/src/chat/agent/resume.ts +++ b/packages/junior/src/chat/agent/resume.ts @@ -176,8 +176,6 @@ export function createResumeState(args: ResumeStateArgs) { messages: PiMessage[], trailingMessageProvenance?: ConversationMessageProvenance[], ): Promise { - // Boundary mismatches rethrow from persistRunningSessionRecord so they - // never collapse into TurnInputCommitLost → lost_lease recovery. const persisted = await this.persistSafeBoundary( messages, trailingMessageProvenance, diff --git a/packages/junior/src/chat/conversations/projection.ts b/packages/junior/src/chat/conversations/projection.ts index a4d8f0554..e0c15190e 100644 --- a/packages/junior/src/chat/conversations/projection.ts +++ b/packages/junior/src/chat/conversations/projection.ts @@ -31,7 +31,6 @@ import { type PiConversationProjection, } from "@/chat/pi/conversation-events"; import { stripRuntimeTurnContext } from "@/chat/pi/transcript"; -import { AgentHistoryBoundaryError } from "@/chat/runtime/turn"; import { sanitizePostgresJson } from "@/db/postgres-json"; import type { ModelProfile } from "@/chat/model-profile"; import type { TurnReasoningLevel } from "@/chat/reasoning-level"; @@ -364,7 +363,7 @@ async function commitMessagesLocked( ...turnContextEvents, ]); } else { - throw new AgentHistoryBoundaryError( + throw new Error( `Agent history for ${args.conversationId} changed before its committed boundary`, ); } diff --git a/packages/junior/src/chat/runtime/turn.ts b/packages/junior/src/chat/runtime/turn.ts index 0ff9c3086..81dc3b399 100644 --- a/packages/junior/src/chat/runtime/turn.ts +++ b/packages/junior/src/chat/runtime/turn.ts @@ -73,33 +73,6 @@ export function isTurnInputCommitLostError( return error instanceof TurnInputCommitLostError; } -/** - * Durable agent history no longer matches the messages a turn is trying to - * checkpoint. This is a permanent shape mismatch for the current attempt, not - * a transient lease loss — callers must not requeue as lost_lease. - */ -export class AgentHistoryBoundaryError extends Error { - readonly code = "agent_history_boundary"; - - constructor(message: string) { - super(message); - this.name = "AgentHistoryBoundaryError"; - } -} - -/** Return whether an error is a permanent agent-history boundary mismatch. */ -export function isAgentHistoryBoundaryError( - error: unknown, -): error is AgentHistoryBoundaryError { - return error instanceof AgentHistoryBoundaryError; -} - -/** True when error text is the durable history-boundary mismatch message. */ -export function isAgentHistoryBoundaryMessage(error: unknown): boolean { - const message = error instanceof Error ? error.message : String(error); - return message.includes("changed before its committed boundary"); -} - /** Error indicating durable turn input should stay pending for a later worker. */ export class TurnInputDeferredError extends Error { readonly code = "turn_input_deferred"; diff --git a/packages/junior/src/chat/services/turn-session-record.ts b/packages/junior/src/chat/services/turn-session-record.ts index 6faed0b19..d440817ed 100644 --- a/packages/junior/src/chat/services/turn-session-record.ts +++ b/packages/junior/src/chat/services/turn-session-record.ts @@ -15,10 +15,6 @@ import { isContinuablePiBoundary, trimTrailingAssistantMessages, } from "@/chat/pi/transcript"; -import { - isAgentHistoryBoundaryError, - isAgentHistoryBoundaryMessage, -} from "@/chat/runtime/turn"; import { addAgentTurnUsage, type AgentTurnUsage } from "@/chat/usage"; import { persistWithRetry } from "@/chat/services/persist-retry"; import { TurnSliceLimitExceededError } from "@/chat/services/turn-limit"; @@ -89,12 +85,19 @@ export async function loadTurnSessionRecord( ctx.conversationId, ctx.sessionId, ); - const hasAwaitingResumeRecord = Boolean( - existingSessionRecord && existingSessionRecord.state === "awaiting_resume", + // A still-running record with a committed turn-start cursor already owns its + // prompt. Treat that as resume so redelivery continues instead of rebuilding + // and re-checkpointing a non-identical user message. + const hasCommittedPrompt = + existingSessionRecord?.turnStartMessageIndex !== undefined; + const resumedFromSessionRecord = Boolean( + existingSessionRecord && + (existingSessionRecord.state === "awaiting_resume" || + (existingSessionRecord.state === "running" && hasCommittedPrompt)), ); return { - resumedFromSessionRecord: hasAwaitingResumeRecord, - currentSliceId: hasAwaitingResumeRecord + resumedFromSessionRecord, + currentSliceId: resumedFromSessionRecord ? existingSessionRecord!.sliceId : 1, existingSessionRecord, @@ -181,13 +184,11 @@ export async function persistRunningSessionRecord(args: { }); return true; } catch (recordError) { - // History-boundary mismatch is permanent for this attempt. Swallowing it as - // false promotes TurnInputCommitLost → lost_lease recovery, which requeues - // forever. Fail closed so the worker can count the attempt / dead-letter. - if ( - isAgentHistoryBoundaryError(recordError) || - isAgentHistoryBoundaryMessage(recordError) - ) { + // Permanent history-shape failures must not collapse into false → + // TurnInputCommitLost → lost_lease recovery, which requeues forever. + const message = + recordError instanceof Error ? recordError.message : String(recordError); + if (message.includes("changed before its committed boundary")) { logSessionRecordError( recordError, "agent.turn.running_session_record.boundary_mismatch", diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index 82750367f..ac4ba50f4 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -13,8 +13,6 @@ import type { SteeringCandidateMessage, } from "@/chat/runtime/slack-runtime"; import { - isAgentHistoryBoundaryError, - isAgentHistoryBoundaryMessage, isCooperativeTurnYieldError, isTurnInputDeferredError, isTurnInputCommitLostError, @@ -896,15 +894,6 @@ export function createSlackConversationWorker( if (isCooperativeTurnYieldError(error)) { return { status: "yielded" } satisfies ConversationWorkerResult; } - // History-boundary mismatch is permanent for this attempt. Do not map - // it to lost_lease recovery (infinite requeue). Let the worker catch - // count the attempt and dead-letter when retries are exhausted. - if ( - isAgentHistoryBoundaryError(error) || - isAgentHistoryBoundaryMessage(error) - ) { - throw error; - } if (isTurnInputCommitLostError(error)) { return { status: "lost_lease" } satisfies ConversationWorkerResult; } diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 0351a03fd..79248a890 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -4,10 +4,6 @@ import { getChatConfig } from "@/chat/config"; import { logException, logInfo, logWarn, withLogContext } from "@/chat/logging"; import type { ConversationStore } from "@/chat/conversations/store"; import { isProviderRetryError } from "@/chat/services/provider-error"; -import { - isAgentHistoryBoundaryError, - isAgentHistoryBoundaryMessage, -} from "@/chat/runtime/turn"; import { ConversationQueueMessageRejectedError, type ConversationQueueMessage, @@ -667,14 +663,6 @@ async function processConversationWorkInContext( // recovery nudge. Once durable recovery state is recorded and one nudge is // sent, the delivery is acknowledged; only when recording recovery state // itself fails is the error rethrown so plain redelivery retries it. - // - // History-boundary mismatches are permanent for the current attempt shape. - // Requeueing them (especially empty-attempt continue wakes) farms the same - // poison turn forever. Fail closed: count attempts when present, otherwise - // release without a recovery wake. - const permanentBoundaryMismatch = - isAgentHistoryBoundaryError(error) || - isAgentHistoryBoundaryMessage(error); let recoveryRecorded = false; try { const failure = @@ -695,17 +683,6 @@ async function processConversationWorkInContext( nowMs: errorNowMs, state: options.state, }); - } else if (permanentBoundaryMismatch) { - // Do not schedule recovery wakes for permanent shape mismatches. - // Attempts are already counted above when present; releasing without a - // wake stops the poison-turn farm (including empty-attempt continues). - await releaseConversationWork({ - conversationId, - leaseToken: lease.leaseToken, - conversationStore: options.conversationStore, - nowMs: errorNowMs, - state: options.state, - }); } else { const resumeRequested = await requestConversationContinuation({ conversationId, diff --git a/packages/junior/tests/component/runtime/agent-run-provider-retry.test.ts b/packages/junior/tests/component/runtime/agent-run-provider-retry.test.ts index d1b3319d6..0295827cb 100644 --- a/packages/junior/tests/component/runtime/agent-run-provider-retry.test.ts +++ b/packages/junior/tests/component/runtime/agent-run-provider-retry.test.ts @@ -1351,10 +1351,8 @@ describe("agent run continuation", () => { await executeAgentRun({ conversationId, turnId: sessionId, - // Production redelivery passes live projection history that already - // ends with the durable checkpoint. The agent must re-checkpoint that - // exact message (not a Date.now() rebuild) so commitMessages stays - // append-only. + // A running record with turnStartMessageIndex already owns the prompt. + // Redelivery must continue that history, not rebuild/recheckpoint it. input: { messageText: "help me", piMessages: [checkpointedPrompt] }, routing: { destinationVisibility: "private", From 416597050d0902fd389f711a630dd6df9d9be927 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 23:14:51 +0000 Subject: [PATCH 3/4] fix(chat): cap empty continue wakes and unbreak checkpoint CI Resume already-checkpointed running turns without fail-closed history throws that broke handoff follow-ups. Add a consecutive empty-wake ceiling so lost_lease/continue recovery cannot farm forever without mailbox progress. Co-Authored-By: David Cramer --- .../src/chat/services/turn-session-record.ts | 26 ++- .../junior/src/chat/task-execution/state.ts | 97 +++++++++++ .../junior/src/chat/task-execution/store.ts | 27 +++ .../junior/src/chat/task-execution/worker.ts | 159 +++++++++++++++--- .../plugins/plugin-prompt-hooks.test.ts | 24 ++- .../task-execution/conversation-work.test.ts | 70 ++++++++ 6 files changed, 354 insertions(+), 49 deletions(-) diff --git a/packages/junior/src/chat/services/turn-session-record.ts b/packages/junior/src/chat/services/turn-session-record.ts index d440817ed..a2425d9ad 100644 --- a/packages/junior/src/chat/services/turn-session-record.ts +++ b/packages/junior/src/chat/services/turn-session-record.ts @@ -184,24 +184,22 @@ export async function persistRunningSessionRecord(args: { }); return true; } catch (recordError) { - // Permanent history-shape failures must not collapse into false → - // TurnInputCommitLost → lost_lease recovery, which requeues forever. + // Boundary mismatch is permanent for this attempt's message shape. Log it + // distinctly, but still return false: the poison-turn farm is stopped by + // treating already-checkpointed running turns as resume (no rebuild / + // recheckpoint), not by failing closed here. Handoff/compaction follow-ups + // and no-checkpoint resume still rely on false when no mailbox ack is + // pending. const message = recordError instanceof Error ? recordError.message : String(recordError); - if (message.includes("changed before its committed boundary")) { - logSessionRecordError( - recordError, - "agent.turn.running_session_record.boundary_mismatch", - args, - { - "app.ai.resume_slice_id": args.sliceId, - }, - ); - throw recordError; - } + const boundaryMismatch = message.includes( + "changed before its committed boundary", + ); logSessionRecordError( recordError, - "agent.turn.running_session_record.failed", + boundaryMismatch + ? "agent.turn.running_session_record.boundary_mismatch" + : "agent.turn.running_session_record.failed", args, { "app.ai.resume_slice_id": args.sliceId, diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index aa8065268..4b5c02b3c 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -60,6 +60,8 @@ export const CONVERSATION_WORK_LEASE_TTL_MS = 90_000; export const CONVERSATION_WORK_CHECK_IN_INTERVAL_MS = 15_000; export const CONVERSATION_WORK_STALE_ENQUEUE_MS = 60_000; export const CONVERSATION_WORK_MAX_DELIVERY_ATTEMPTS = 5; +/** Empty continue/lost-lease wakes with no mailbox progress before fail-closed. */ +export const CONVERSATION_WORK_MAX_EMPTY_WAKES = 5; const inboundMessageSourceSchema = z.enum([ "api", @@ -127,6 +129,8 @@ export interface Lease { } export interface ConversationExecution { + /** Consecutive empty wakes (no mailbox attempt) that made no durable progress. */ + emptyWakeCount?: number; inboundMessageIds: string[]; lastCheckpointAtMs?: number; lastEnqueuedAtMs?: number; @@ -432,6 +436,7 @@ function normalizeExecution( pendingCount: pendingMessages.length, pendingMessages, lease, + emptyWakeCount: toOptionalNumber(value.emptyWakeCount), lastCheckpointAtMs: toOptionalNumber(value.lastCheckpointAtMs), lastEnqueuedAtMs: toOptionalNumber(value.lastEnqueuedAtMs), runId: toOptionalString(value.runId), @@ -1138,6 +1143,8 @@ export async function appendInboundMessage(args: { next, { ...current.execution, + // Fresh mailbox work resets the empty-wake poison budget. + emptyWakeCount: undefined, status, inboundMessageIds: [ ...current.execution.inboundMessageIds, @@ -1741,11 +1748,101 @@ export async function completeConversationWork(args: { }); } +/** + * Record one empty wake that made no mailbox progress (continue / lost-lease + * recovery with nothing to ack). Caps poison requeue farms that bypass the + * per-message delivery attempt counter. + */ +export async function recordEmptyWakeFailure(args: { + conversationId: string; + leaseToken: string; + nowMs?: number; + state?: StateAdapter; +}): Promise { + const nowMs = args.nowMs ?? now(); + return await withConversationMutation(args, async (state, lock) => { + const current = await readConversation(state, args.conversationId); + if (!current || current.execution.lease?.token !== args.leaseToken) { + return { + status: "lost_lease", + pendingCount: 0, + deadLetteredMessages: [], + }; + } + const emptyWakeCount = (current.execution.emptyWakeCount ?? 0) + 1; + const terminal = emptyWakeCount >= CONVERSATION_WORK_MAX_EMPTY_WAKES; + await writeConversation( + state, + lock, + withExecutionUpdate( + current, + { + ...current.execution, + emptyWakeCount, + ...(terminal + ? { + lease: undefined, + status: pendingMessages(current).length > 0 ? "pending" : "failed", + } + : {}), + }, + nowMs, + ), + ); + return { + status: "recorded", + pendingCount: current.execution.pendingMessages.length, + deadLetteredMessages: [], + ...(terminal ? { terminal: true as const } : {}), + }; + }); +} + +/** Clear empty-wake streak after durable progress (ack, complete, new work). */ +export async function clearEmptyWakeCount(args: { + conversationId: string; + leaseToken?: string; + nowMs?: number; + state?: StateAdapter; +}): Promise { + const nowMs = args.nowMs ?? now(); + return await withConversationMutation(args, async (state, lock) => { + const current = await readConversation(state, args.conversationId); + if (!current) { + return false; + } + if ( + args.leaseToken !== undefined && + current.execution.lease?.token !== args.leaseToken + ) { + return false; + } + if ((current.execution.emptyWakeCount ?? 0) === 0) { + return true; + } + await writeConversation( + state, + lock, + withExecutionUpdate( + current, + { + ...current.execution, + emptyWakeCount: undefined, + }, + nowMs, + ), + ); + return true; + }); +} + /** Failure outcome: `lost_lease` (another owner took over), `recorded` (attempt counted), or `skipped` (durable progress was made). */ export interface AttemptFailure { pendingCount: number; deadLetteredMessages: InboundMessage[]; status: "lost_lease" | "recorded" | "skipped"; + /** True when empty-wake failures hit the fail-closed ceiling. */ + terminal?: boolean; } /** diff --git a/packages/junior/src/chat/task-execution/store.ts b/packages/junior/src/chat/task-execution/store.ts index 05c256c61..a7c3c5ec0 100644 --- a/packages/junior/src/chat/task-execution/store.ts +++ b/packages/junior/src/chat/task-execution/store.ts @@ -9,6 +9,7 @@ export { CONVERSATION_WORK_CHECK_IN_INTERVAL_MS, CONVERSATION_WORK_LEASE_TTL_MS, CONVERSATION_WORK_MAX_DELIVERY_ATTEMPTS, + CONVERSATION_WORK_MAX_EMPTY_WAKES, CONVERSATION_WORK_STALE_ENQUEUE_MS, isFinalAttempt, isInvalidConversationRecordError, @@ -453,6 +454,32 @@ export async function completeConversationWork(args: { return result; } +/** Record one empty wake with no mailbox progress before fail-closed. */ +export async function recordEmptyWakeFailure(args: { + conversationId: string; + leaseToken: string; + conversationStore?: ConversationStore; + nowMs?: number; + state?: StateAdapter; +}) { + const result = await workState.recordEmptyWakeFailure(args); + if (result.status === "recorded") { + await recordExecutionMetadata(args); + } + return result; +} + +/** Clear empty-wake streak after durable progress. */ +export async function clearEmptyWakeCount(args: { + conversationId: string; + leaseToken?: string; + conversationStore?: ConversationStore; + nowMs?: number; + state?: StateAdapter; +}) { + return await workState.clearEmptyWakeCount(args); +} + /** Record one failed delivery attempt and dead-letter messages at their limit. */ export async function recordAttemptFailure(args: { conversationId: string; diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 79248a890..cfa5a87a0 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -14,6 +14,7 @@ import { beginConversationResume, checkInConversationWork, clearConsumedConversationWake, + clearEmptyWakeCount, completeConversationWork, CONVERSATION_WORK_CHECK_IN_INTERVAL_MS, countPendingConversationMessages, @@ -24,6 +25,7 @@ import { isFinalAttempt, isInvalidConversationRecordError, recordAttemptFailure, + recordEmptyWakeFailure, releaseConversationWork, requestConversationContinuation, startConversationWork, @@ -120,13 +122,37 @@ function nudgeIdempotencyKey( return `${reason}:${conversationId}:${nowMs}`; } +/** + * Requeue after lease loss. Empty wakes (no mailbox attempt) count toward + * CONVERSATION_WORK_MAX_EMPTY_WAKES so poison continue loops fail closed. + */ async function requestLostLeaseRecovery(args: { conversationId: string; destination: Destination; + /** True when this wake had no mailbox messages to attempt. */ + emptyWake: boolean; leaseToken: string; nowMs: number; options: ProcessConversationWorkOptions; -}): Promise { +}): Promise<"requeued" | "terminal" | "skipped"> { + if (args.emptyWake) { + const emptyFailure = await recordEmptyWakeFailure({ + conversationId: args.conversationId, + leaseToken: args.leaseToken, + nowMs: args.nowMs, + state: args.options.state, + }); + if (emptyFailure.status === "lost_lease") { + return "skipped"; + } + if (emptyFailure.terminal) { + logWarn("conversation.work.empty_wake.terminal", { + "app.conversation.empty_wake_count": "max", + }); + return "terminal"; + } + } + const resumeRequested = await requestConversationContinuation({ conversationId: args.conversationId, destination: args.destination, @@ -136,7 +162,7 @@ async function requestLostLeaseRecovery(args: { state: args.options.state, }); if (!resumeRequested) { - return; + return "skipped"; } const released = await releaseConversationWork({ conversationId: args.conversationId, @@ -146,7 +172,7 @@ async function requestLostLeaseRecovery(args: { state: args.options.state, }); if (!released) { - return; + return "skipped"; } await ensureConversationWake({ conversationId: args.conversationId, @@ -161,6 +187,7 @@ async function requestLostLeaseRecovery(args: { replaceExistingWake: true, state: args.options.state, }); + return "requeued"; } /** @@ -434,14 +461,17 @@ async function processConversationWorkInContext( leaseLost ) { markLeaseLost(); - await requestLostLeaseRecovery({ + const recovery = await requestLostLeaseRecovery({ conversationId, destination, + emptyWake: true, leaseToken: lease.leaseToken, nowMs: now(options), options, }); - return { status: "lost_lease" }; + return { + status: recovery === "terminal" ? "failed" : "lost_lease", + }; } const resumePending = leasedWork.execution.status === "awaiting_resume"; @@ -467,14 +497,17 @@ async function processConversationWorkInContext( }); if (!resumeStarted) { markLeaseLost(); - await requestLostLeaseRecovery({ + const recovery = await requestLostLeaseRecovery({ conversationId, destination, + emptyWake: true, leaseToken: lease.leaseToken, nowMs: now(options), options, }); - return { status: "lost_lease" }; + return { + status: recovery === "terminal" ? "failed" : "lost_lease", + }; } } @@ -517,25 +550,40 @@ async function processConversationWorkInContext( const result = await options.run(workerContext); hasRun = true; + const emptyWake = attemptMessageIds.length === 0; if (result.status === "lost_lease") { - await requestLostLeaseRecovery({ + const recovery = await requestLostLeaseRecovery({ conversationId, destination, + emptyWake, leaseToken: lease.leaseToken, nowMs: now(options), options, }); - return { status: "lost_lease" }; + return { + status: recovery === "terminal" ? "failed" : "lost_lease", + }; } if (leaseLost) { - await requestLostLeaseRecovery({ + const recovery = await requestLostLeaseRecovery({ conversationId, destination, + emptyWake, leaseToken: lease.leaseToken, nowMs: now(options), options, }); - return { status: "lost_lease" }; + return { + status: recovery === "terminal" ? "failed" : "lost_lease", + }; + } + if (!emptyWake) { + await clearEmptyWakeCount({ + conversationId, + leaseToken: lease.leaseToken, + nowMs: now(options), + state: options.state, + }); } if (result.status === "yielded") { const resumeRequested = await requestConversationContinuation({ @@ -622,6 +670,36 @@ async function processConversationWorkInContext( ) { break; } + // Continue/resume wakes with nothing left to ack must not loop forever. + if (attemptMessageIds.length === 0) { + const emptyFailure = await recordEmptyWakeFailure({ + conversationId, + leaseToken: lease.leaseToken, + nowMs: now(options), + state: options.state, + }); + if (emptyFailure.status === "lost_lease") { + return { status: "lost_lease" }; + } + if (emptyFailure.terminal) { + logWarn("conversation.work.empty_wake.terminal", { + "app.conversation.empty_wake_count": "max", + }); + return { status: "failed" }; + } + const resumeRequested = await requestConversationContinuation({ + conversationId, + destination, + leaseToken: lease.leaseToken, + conversationStore: options.conversationStore, + nowMs: now(options), + state: options.state, + }); + if (!resumeRequested) { + return { status: "lost_lease" }; + } + return await yieldWork(); + } } const completion = await completeConversationWork({ @@ -653,6 +731,12 @@ async function processConversationWorkInContext( : { status: "completed" }; } + // Lease may already be released by completeConversationWork. + await clearEmptyWakeCount({ + conversationId, + nowMs: now(options), + state: options.state, + }); logInfo("conversation.work.completed", { "app.worker.elapsed_ms": now(options) - startedAtMs, }); @@ -684,28 +768,47 @@ async function processConversationWorkInContext( state: options.state, }); } else { - const resumeRequested = await requestConversationContinuation({ - conversationId, - destination, - leaseToken: lease.leaseToken, - conversationStore: options.conversationStore, - nowMs: errorNowMs, - state: options.state, - }); - if (resumeRequested) { - await ensureConversationWake({ + let skipWake = false; + if (attemptMessageIds.length === 0) { + const emptyFailure = await recordEmptyWakeFailure({ + conversationId, + leaseToken: lease.leaseToken, + nowMs: errorNowMs, + state: options.state, + }); + if (emptyFailure.status === "lost_lease") { + skipWake = true; + } else if (emptyFailure.terminal) { + logWarn("conversation.work.empty_wake.terminal", { + "app.conversation.empty_wake_count": "max", + }); + skipWake = true; + } + } + if (!skipWake) { + const resumeRequested = await requestConversationContinuation({ conversationId, + destination, + leaseToken: lease.leaseToken, conversationStore: options.conversationStore, - idempotencyKey: nudgeIdempotencyKey( - "error", - conversationId, - errorNowMs, - ), nowMs: errorNowMs, - queue: options.queue, - replaceExistingWake: true, state: options.state, }); + if (resumeRequested) { + await ensureConversationWake({ + conversationId, + conversationStore: options.conversationStore, + idempotencyKey: nudgeIdempotencyKey( + "error", + conversationId, + errorNowMs, + ), + nowMs: errorNowMs, + queue: options.queue, + replaceExistingWake: true, + state: options.state, + }); + } } await releaseConversationWork({ conversationId, diff --git a/packages/junior/tests/component/plugins/plugin-prompt-hooks.test.ts b/packages/junior/tests/component/plugins/plugin-prompt-hooks.test.ts index 24fd24df2..263c8f101 100644 --- a/packages/junior/tests/component/plugins/plugin-prompt-hooks.test.ts +++ b/packages/junior/tests/component/plugins/plugin-prompt-hooks.test.ts @@ -117,7 +117,10 @@ import { z } from "zod"; import { executeAgentRun } from "@/chat/agent"; import { setPlugins } from "@/chat/plugins/agent-hooks"; import { disconnectStateAdapter } from "@/chat/state/adapter"; -import { upsertAgentTurnSessionRecord } from "@/chat/state/turn-session"; +import { + getAgentTurnSessionRecord, + upsertAgentTurnSessionRecord, +} from "@/chat/state/turn-session"; import { getConversationEventStore } from "@/chat/db"; import { TurnInputCommitLostError } from "@/chat/runtime/turn"; @@ -313,15 +316,12 @@ describe("plugin prompt hook composition", () => { }), ).rejects.toBeInstanceOf(TurnInputCommitLostError); + // Prompt is already owned by the still-running record. Redelivery continues + // that history instead of rebuilding / re-running user-prompt hooks. await executeAgentRun(request); expect(recallCount).toBe(1); - expect(JSON.stringify(captured.promptMessages[0])).toContain( - "Use pnpm snapshot 1.", - ); - expect(JSON.stringify(captured.promptMessages[0])).not.toContain( - "Use pnpm snapshot 2.", - ); + expect(captured.promptMessages).toEqual([]); const stored = await getConversationEventStore().loadByIdempotencyKey( LOCAL_DESTINATION.conversationId, `turn:${turnId}:context:memory:0`, @@ -332,6 +332,16 @@ describe("plugin prompt hook composition", () => { memories: [{ id: "memory-1", content: "Use pnpm snapshot 1." }], }, }); + const session = await getAgentTurnSessionRecord( + LOCAL_DESTINATION.conversationId, + turnId, + ); + expect(JSON.stringify(session?.piMessages)).toContain( + "Use pnpm snapshot 1.", + ); + expect(JSON.stringify(session?.piMessages)).not.toContain( + "Use pnpm snapshot 2.", + ); }); it("runs user prompt hooks for non-bootstrap follow-up prompts", async () => { diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 6f3538421..04837c2c3 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -13,6 +13,7 @@ import { completeConversationWork, CONVERSATION_WORK_LEASE_TTL_MS, CONVERSATION_WORK_MAX_DELIVERY_ATTEMPTS, + CONVERSATION_WORK_MAX_EMPTY_WAKES, countPendingConversationMessages, drainConversationMailbox, getConversationWorkState, @@ -1291,6 +1292,8 @@ describe("conversation work execution", () => { expect(state?.needsRun).toBe(true); expect(state ? countPendingConversationMessages(state) : 0).toBe(1); expect(state?.lastEnqueuedAtMs).toBe(2_000); + // Pending mailbox work is not an empty wake — do not burn the empty-wake budget. + expect(state?.execution.emptyWakeCount).toBeUndefined(); expect(queue.sentRecords()).toEqual([ { conversationId: CONVERSATION_ID, @@ -1299,6 +1302,73 @@ describe("conversation work execution", () => { ]); }); + it("fails closed after consecutive empty continue wakes without mailbox progress", async () => { + const queue = createConversationWorkQueueTestAdapter(); + let currentNowMs = 1_000; + // Seed an awaiting_resume conversation with no pending mailbox messages — + // the poison continue / lost_lease farm path. + await requestConversationWork({ + conversationId: CONVERSATION_ID, + destination: SLACK_DESTINATION, + nowMs: 1_000, + }); + const started = await startConversationWork({ + conversationId: CONVERSATION_ID, + nowMs: 1_000, + }); + expect(started.status).toBe("acquired"); + if (started.status !== "acquired") return; + await expect( + requestConversationContinuation({ + conversationId: CONVERSATION_ID, + destination: SLACK_DESTINATION, + leaseToken: started.leaseToken, + nowMs: 1_000, + }), + ).resolves.toBe(true); + await releaseConversationWork({ + conversationId: CONVERSATION_ID, + leaseToken: started.leaseToken, + nowMs: 1_000, + }); + + for (let attempt = 0; attempt < CONVERSATION_WORK_MAX_EMPTY_WAKES; attempt++) { + currentNowMs = 2_000 + attempt; + const result = await processConversationWork(conversationQueueMessage(), { + nowMs: () => currentNowMs, + queue, + run: async () => { + currentNowMs += 1; + return { status: "lost_lease" }; + }, + }); + if (attempt < CONVERSATION_WORK_MAX_EMPTY_WAKES - 1) { + expect(result).toEqual({ status: "lost_lease" }); + } else { + expect(result).toEqual({ status: "failed" }); + } + } + + const state = await getConversationWorkState({ + conversationId: CONVERSATION_ID, + }); + expect(state?.execution.emptyWakeCount).toBe( + CONVERSATION_WORK_MAX_EMPTY_WAKES, + ); + expect(state?.execution.status).toBe("failed"); + expect(state?.lease).toBeUndefined(); + // Final terminal wake must not schedule another recovery nudge. + expect( + queue + .sentRecords() + .filter((record) => + (record.idempotencyKey ?? "").startsWith( + `lost_lease:${CONVERSATION_ID}:`, + ), + ), + ).toHaveLength(CONVERSATION_WORK_MAX_EMPTY_WAKES - 1); + }); + it("drains pending messages and completes the leased conversation", async () => { const queue = createConversationWorkQueueTestAdapter(); await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); From a12033d465bdf2900d8b7a43ae9b75d017c86385 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 23:22:49 +0000 Subject: [PATCH 4/4] fix(chat): only count empty wakes on lost-lease/error paths Healthy resume+defer slices were treated as empty-wake failures and forced a yield. Count CONVERSATION_WORK_MAX_EMPTY_WAKES only when recovery has no mailbox progress, and clear the streak after successful runs. --- .../junior/src/chat/task-execution/state.ts | 3 +- .../junior/src/chat/task-execution/worker.ts | 45 +++---------------- .../task-execution/conversation-work.test.ts | 30 +++++++------ 3 files changed, 26 insertions(+), 52 deletions(-) diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 4b5c02b3c..f77b84dc1 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -1782,7 +1782,8 @@ export async function recordEmptyWakeFailure(args: { ...(terminal ? { lease: undefined, - status: pendingMessages(current).length > 0 ? "pending" : "failed", + status: + pendingMessages(current).length > 0 ? "pending" : "failed", } : {}), }, diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index cfa5a87a0..ba8f80d26 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -577,14 +577,13 @@ async function processConversationWorkInContext( status: recovery === "terminal" ? "failed" : "lost_lease", }; } - if (!emptyWake) { - await clearEmptyWakeCount({ - conversationId, - leaseToken: lease.leaseToken, - nowMs: now(options), - state: options.state, - }); - } + // Successful slices (including empty continue resumes) made progress. + await clearEmptyWakeCount({ + conversationId, + leaseToken: lease.leaseToken, + nowMs: now(options), + state: options.state, + }); if (result.status === "yielded") { const resumeRequested = await requestConversationContinuation({ conversationId, @@ -670,36 +669,6 @@ async function processConversationWorkInContext( ) { break; } - // Continue/resume wakes with nothing left to ack must not loop forever. - if (attemptMessageIds.length === 0) { - const emptyFailure = await recordEmptyWakeFailure({ - conversationId, - leaseToken: lease.leaseToken, - nowMs: now(options), - state: options.state, - }); - if (emptyFailure.status === "lost_lease") { - return { status: "lost_lease" }; - } - if (emptyFailure.terminal) { - logWarn("conversation.work.empty_wake.terminal", { - "app.conversation.empty_wake_count": "max", - }); - return { status: "failed" }; - } - const resumeRequested = await requestConversationContinuation({ - conversationId, - destination, - leaseToken: lease.leaseToken, - conversationStore: options.conversationStore, - nowMs: now(options), - state: options.state, - }); - if (!resumeRequested) { - return { status: "lost_lease" }; - } - return await yieldWork(); - } } const completion = await completeConversationWork({ diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 04837c2c3..1c1e290d2 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -1332,22 +1332,26 @@ describe("conversation work execution", () => { nowMs: 1_000, }); + const results: Array<{ status: string }> = []; for (let attempt = 0; attempt < CONVERSATION_WORK_MAX_EMPTY_WAKES; attempt++) { currentNowMs = 2_000 + attempt; - const result = await processConversationWork(conversationQueueMessage(), { - nowMs: () => currentNowMs, - queue, - run: async () => { - currentNowMs += 1; - return { status: "lost_lease" }; - }, - }); - if (attempt < CONVERSATION_WORK_MAX_EMPTY_WAKES - 1) { - expect(result).toEqual({ status: "lost_lease" }); - } else { - expect(result).toEqual({ status: "failed" }); - } + results.push( + await processConversationWork(conversationQueueMessage(), { + nowMs: () => currentNowMs, + queue, + run: async () => { + currentNowMs += 1; + return { status: "lost_lease" }; + }, + }), + ); } + expect(results.slice(0, -1)).toEqual( + Array.from({ length: CONVERSATION_WORK_MAX_EMPTY_WAKES - 1 }, () => ({ + status: "lost_lease", + })), + ); + expect(results.at(-1)).toEqual({ status: "failed" }); const state = await getConversationWorkState({ conversationId: CONVERSATION_ID,