From 7bcdb5823e6c15855c809ae32065370d8b729e07 Mon Sep 17 00:00:00 2001 From: joshavant <830519+joshavant@users.noreply.github.com> Date: Tue, 4 Aug 2026 18:33:48 -0500 Subject: [PATCH] test: prove repeated request recovery through gateway --- .../mock-openai/mock-openai-contracts.ts | 12 + .../mock-openai/mock-openai-events.test.ts | 11 + .../mock-openai/mock-openai-events.ts | 15 + .../mock-openai-responses-websocket.ts | 3 + .../src/providers/mock-openai/server.ts | 28 +- src/logging/diagnostic-run-activity.test.ts | 53 ++- src/logging/diagnostic-run-activity.ts | 19 +- src/logging/diagnostic-stability.test.ts | 25 ++ src/logging/diagnostic-stability.ts | 10 +- src/logging/diagnostic.test.ts | 65 ++++ src/logging/diagnostic.ts | 6 +- ...eway-repeated-request-recovery.e2e.test.ts | 302 ++++++++++++++++++ 12 files changed, 526 insertions(+), 23 deletions(-) create mode 100644 test/e2e/qa-lab/runtime/gateway-repeated-request-recovery.e2e.test.ts diff --git a/extensions/qa-lab/src/providers/mock-openai/mock-openai-contracts.ts b/extensions/qa-lab/src/providers/mock-openai/mock-openai-contracts.ts index ad202368d6d5..00676ebd8f48 100644 --- a/extensions/qa-lab/src/providers/mock-openai/mock-openai-contracts.ts +++ b/extensions/qa-lab/src/providers/mock-openai/mock-openai-contracts.ts @@ -39,6 +39,14 @@ export type QaMockProviderDispatchResult = { export type StreamEvent = | { type: "response.created"; response: { id: string } } + | { + type: "response.failed"; + response: { + id: string; + status: "failed"; + error?: { code: string; message: string }; + }; + } | { type: "response.output_item.added"; output_index?: number; @@ -220,6 +228,10 @@ export const QA_EMPTY_RESPONSE_RECOVERY_PROMPT_RE = /empty response continuation export const QA_EMPTY_RESPONSE_EXHAUSTION_PROMPT_RE = /empty response exhaustion qa check/i; export const QA_EMPTY_RESPONSE_SIDE_EFFECT_RECOVERY_PROMPT_RE = /empty response after write recovery qa check/i; +export const QA_REPEATED_REQUEST_RECOVERY_PROMPT_RE = /repeated request recovery gateway qa check/i; +export const QA_REPEATED_REQUEST_QUEUED_REPLY_PROMPT_RE = + /repeated request queued reply gateway qa check/i; +export const QA_REPEATED_REQUEST_QUEUED_REPLY_MARKER = "GATEWAY_REPEATED_REQUEST_QUEUED_OK"; export const QA_STREAMING_PROMPT_RE = /(?:partial|quiet) streaming qa check/i; export const QA_FINAL_ONLY_MARKER_STREAMING_PROMPT_RE = /final-only marker streaming qa check/i; export const QA_BLOCK_STREAMING_PROMPT_RE = /block streaming qa check/i; diff --git a/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.test.ts b/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.test.ts index 09907948fec4..bdceabe55de2 100644 --- a/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.test.ts +++ b/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.test.ts @@ -3,6 +3,7 @@ import type { StreamEvent } from "./mock-openai-contracts.js"; import { buildAssistantEvents, buildAssistantThenToolCallEvents, + buildFailedResponseEvents, buildReasoningAndAssistantEvents, buildReasoningOnlyEvents, } from "./mock-openai-events.js"; @@ -29,6 +30,16 @@ function readOutputItemSlots(events: StreamEvent[]) { } describe("mock OpenAI Responses output item slots", () => { + it("emits the provider no-details failure used by repeated-request recovery QA", () => { + expect(buildFailedResponseEvents()).toEqual([ + expect.objectContaining({ type: "response.created" }), + expect.objectContaining({ + type: "response.failed", + response: expect.not.objectContaining({ error: expect.anything() }), + }), + ]); + }); + it("indexes preview deltas and the final answer on the same assistant slot", () => { const events = buildAssistantEvents([ { diff --git a/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.ts b/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.ts index af92b2965de1..2cbd08467822 100644 --- a/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.ts +++ b/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.ts @@ -6,6 +6,21 @@ import { buildMockFunctionCall, buildToolCallEventsWithArgs, } from "./mock-openai-tooling.js"; + +export function buildFailedResponseEvents(): StreamEvent[] { + const responseId = `resp_qa_failed_${Date.now()}`; + return [ + { type: "response.created", response: { id: responseId } }, + { + type: "response.failed", + response: { + id: responseId, + status: "failed", + }, + }, + ]; +} + export function buildToolCallEvents(prompt: string): StreamEvent[] { const targetPath = readTargetFromPrompt(prompt); return buildToolCallEventsWithArgs("read", { path: targetPath }); diff --git a/extensions/qa-lab/src/providers/mock-openai/mock-openai-responses-websocket.ts b/extensions/qa-lab/src/providers/mock-openai/mock-openai-responses-websocket.ts index 4619079daa41..83aa54ef814a 100644 --- a/extensions/qa-lab/src/providers/mock-openai/mock-openai-responses-websocket.ts +++ b/extensions/qa-lab/src/providers/mock-openai/mock-openai-responses-websocket.ts @@ -196,6 +196,9 @@ export function attachQaMockResponsesWebSocketServer(params: { return; } const { events } = dispatched; + if (dispatched.responsePauseMs !== undefined) { + await sleep(dispatched.responsePauseMs); + } const completion = events.find((event) => event.type === "response.completed"); if (completion?.type === "response.completed") { if (!events.some((event) => event.type === "response.created")) { diff --git a/extensions/qa-lab/src/providers/mock-openai/server.ts b/extensions/qa-lab/src/providers/mock-openai/server.ts index 683aaee2f594..3489aecfdad3 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.ts @@ -38,6 +38,9 @@ import { QA_EMPTY_RESPONSE_RECOVERY_PROMPT_RE, QA_EMPTY_RESPONSE_EXHAUSTION_PROMPT_RE, QA_EMPTY_RESPONSE_SIDE_EFFECT_RECOVERY_PROMPT_RE, + QA_REPEATED_REQUEST_RECOVERY_PROMPT_RE, + QA_REPEATED_REQUEST_QUEUED_REPLY_PROMPT_RE, + QA_REPEATED_REQUEST_QUEUED_REPLY_MARKER, QA_STREAMING_PROMPT_RE, QA_FINAL_ONLY_MARKER_STREAMING_PROMPT_RE, QA_BLOCK_STREAMING_PROMPT_RE, @@ -139,6 +142,7 @@ import { buildAssistantEvents, buildReasoningOnlyEvents, buildReasoningAndAssistantEvents, + buildFailedResponseEvents, } from "./mock-openai-events.js"; import { extractLastUserText, @@ -280,6 +284,9 @@ const QA_STREAMING_TOOL_PROGRESS_CONTINUATION_RE = /^Continue with (?:the current Matrix QA scenario|the QA scenario plan and report worked, failed, and blocked items)\.$/i; const QA_CODE_MODE_TARGET_MARKER = "qa-code-mode-target:"; const QA_FAILED_TOOL_TERMINAL_RECOVERY_PROMPT_RE = /failed tool terminal recovery qa check/i; +// Keep each real provider request active long enough for retries to span the +// unchanged five-minute recovery bound while remaining below first-byte timeout. +const QA_REPEATED_REQUEST_RESPONSE_PAUSE_MS = 110_000; function isStreamingToolProgressContinuationText(text: string) { const trimmed = text.trim(); @@ -853,6 +860,14 @@ async function buildResponsesPayload( ) ? extractLatestToolOutput(input) : ""); + // The queued followup carries the stalled prompt in transcript history, so + // current-turn dispatch must win before the persistent recovery fixture. + if (QA_REPEATED_REQUEST_QUEUED_REPLY_PROMPT_RE.test(prompt)) { + return buildAssistantEvents(QA_REPEATED_REQUEST_QUEUED_REPLY_MARKER); + } + if (QA_REPEATED_REQUEST_RECOVERY_PROMPT_RE.test(allInputText)) { + return buildFailedResponseEvents(); + } const toolJson = parseToolOutputJson(scenarioToolOutput); if (codeModeControlJson?.status === "waiting" && hasToolDefinition(toolDeclarationBody, "wait")) { if ("cellId" in codeModeControlJson && typeof codeModeControlJson.cellId === "string") { @@ -2520,7 +2535,11 @@ export async function startQaMockOpenAiServer(params?: { : undefined); recordRequest({ ...requestSnapshotBase, - outcome: failure ? "error" : "success", + outcome: + failure || events.some((event) => event.type === "response.failed") ? "error" : "success", + ...(events.some((event) => event.type === "response.failed") + ? { errorCode: "response_failed_no_details" } + : {}), plannedToolCallId: plannedToolIdentity.callId, ...(request.route === "responses" && plannedToolIdentity.itemId ? { plannedToolItemId: plannedToolIdentity.itemId } @@ -2549,6 +2568,10 @@ export async function startQaMockOpenAiServer(params?: { ...(QA_FINAL_ONLY_MARKER_STREAMING_PROMPT_RE.test(allInputText) ? { previewPauseMs: finalOnlyMarkerPauseMs } : {}), + ...(QA_REPEATED_REQUEST_RECOVERY_PROMPT_RE.test(allInputText) && + !QA_REPEATED_REQUEST_QUEUED_REPLY_PROMPT_RE.test(prompt) + ? { responsePauseMs: QA_REPEATED_REQUEST_RESPONSE_PAUSE_MS } + : {}), }; }; const dispatchResponses = (request: Omit) => @@ -2696,6 +2719,9 @@ export async function startQaMockOpenAiServer(params?: { return; } const { events } = dispatched; + if (dispatched.responsePauseMs !== undefined) { + await sleep(dispatched.responsePauseMs); + } if (body.stream === false) { const completion = events.at(-1); if (!completion || completion.type !== "response.completed") { diff --git a/src/logging/diagnostic-run-activity.test.ts b/src/logging/diagnostic-run-activity.test.ts index 34d1a373fd80..bbe3d5ff1901 100644 --- a/src/logging/diagnostic-run-activity.test.ts +++ b/src/logging/diagnostic-run-activity.test.ts @@ -533,22 +533,25 @@ describe("repeated request liveness", () => { ).toBeUndefined(); }); - it("clears repeated request evidence on run completion and listener restart", async () => { + it("keeps repeated request evidence across same-logical-owner attempt rearming", async () => { const ref = { sessionId: "completed-session", sessionKey: "agent:main:completed" }; const runId = "completed-run"; startDiagnosticRunActivityTracking(); markDiagnosticEmbeddedRunStarted({ ...ref, runId }); - for (let attempt = 0; attempt < 2; attempt += 1) { - markDiagnosticModelStartedForTest({ - ...ref, - runId, - provider: "mock", - model: "request-model", - observationUnit: "request", - }); - } - expect(getDiagnosticSessionActivitySnapshot(ref).repeatedRequestNoProgressAgeMs).toBe(0); + markDiagnosticModelStartedForTest({ + ...ref, + runId, + provider: "mock", + model: "request-model", + observationUnit: "request", + }); + + markDiagnosticEmbeddedRunEnded(ref); + expect(getDiagnosticSessionActivitySnapshot(ref)).toMatchObject({ + activeWorkKind: undefined, + repeatedRequestNoProgressAgeMs: undefined, + }); emitTrustedDiagnosticEvent({ type: "run.completed", @@ -558,6 +561,34 @@ describe("repeated request liveness", () => { outcome: "completed", }); await waitForDiagnosticEventsDrained(); + markDiagnosticEmbeddedRunStarted({ ...ref, runId }); + markDiagnosticModelStartedForTest({ + ...ref, + runId, + provider: "mock", + model: "request-model", + observationUnit: "request", + }); + expect(getDiagnosticSessionActivitySnapshot(ref)).toMatchObject({ + hasActiveEmbeddedRun: true, + }); + expect( + getDiagnosticSessionActivitySnapshot(ref).repeatedRequestNoProgressAgeMs, + ).toBeGreaterThanOrEqual(0); + + markDiagnosticEmbeddedRunEnded(ref); + expect( + getDiagnosticSessionActivitySnapshot(ref).repeatedRequestNoProgressAgeMs, + ).toBeUndefined(); + + markDiagnosticEmbeddedRunStarted({ ...ref, runId: "successor-run" }); + markDiagnosticModelStartedForTest({ + ...ref, + runId: "successor-run", + provider: "mock", + model: "request-model", + observationUnit: "request", + }); expect( getDiagnosticSessionActivitySnapshot(ref).repeatedRequestNoProgressAgeMs, ).toBeUndefined(); diff --git a/src/logging/diagnostic-run-activity.ts b/src/logging/diagnostic-run-activity.ts index f192d962b53e..b08a47ba9569 100644 --- a/src/logging/diagnostic-run-activity.ts +++ b/src/logging/diagnostic-run-activity.ts @@ -359,9 +359,13 @@ function recordRunCompleted( if (!activity) { return; } - activityByRunId.delete(event.runId); activity.activeTools.clear(); activity.activeModelCalls.clear(); + if (activity.repeatedRequestOwnerRunId === event.runId) { + touchSessionActivity(activity, "run:attempt_completed"); // This run id re-arms after retries. + return; + } + activityByRunId.delete(event.runId); embeddedRunIndex.clear(activity); clearArgumentChurnActivity(activity, { runId: event.runId }); clearArgumentChurnPolicyWaits(activity, { runId: event.runId }); @@ -375,13 +379,11 @@ export function markDiagnosticEmbeddedRunStarted(params: { workKey?: string; }): void { const ownerRunId = params.runId?.trim() || params.sessionId.trim(); - const activity = resolveSessionActivity({ ...params, runId: ownerRunId, create: true }); - if (!activity) { - return; + const activity = resolveSessionActivity({ ...params, runId: ownerRunId, create: true })!; + // New owners must not inherit the prior owner's semantic-stall clock. + if (activity.repeatedRequestOwnerRunId !== ownerRunId) { + clearRepeatedRequestActivity(activity); } - // Registration is the ownership boundary. A replacement or re-armed run - // must never inherit the prior owner's semantic-stall clock. - clearRepeatedRequestActivity(activity); if (activity.argumentChurnStartedAt !== undefined) { clearArgumentChurnActivity(activity, { runId: ownerRunId }); } @@ -418,9 +420,8 @@ export function markDiagnosticEmbeddedRunEnded(params: { if (activity.activeEmbeddedRuns.size === 0) { clearArgumentChurnActivity(activity); clearArgumentChurnPolicyWaits(activity); - clearRepeatedRequestActivity(activity); } - touchSemanticSessionActivity(activity, "embedded_run:ended"); + touchSessionActivity(activity, "embedded_run:ended"); // Retained retry evidence is inert here. } function resolveEmbeddedRunWorkKey(params: { sessionId: string; workKey?: string }): string { diff --git a/src/logging/diagnostic-stability.test.ts b/src/logging/diagnostic-stability.test.ts index 2535532e9888..56a287e33546 100644 --- a/src/logging/diagnostic-stability.test.ts +++ b/src/logging/diagnostic-stability.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { emitDiagnosticEvent, + emitTrustedDiagnosticEvent, resetDiagnosticEventsForTest, waitForDiagnosticEventsDrained, } from "../infra/diagnostic-events.js"; @@ -857,6 +858,30 @@ describe("diagnostic stability recorder", () => { }); }); + it("records sanitized trusted model request instrumentation", async () => { + startDiagnosticStabilityRecorder(); + + emitTrustedDiagnosticEvent({ + type: "model.call.started", + runId: "private-run-id", + callId: "private-call-id", + sessionKey: "private-session-key", + provider: "openai", + model: "gpt-5.4", + observationUnit: "request", + }); + await waitForDiagnosticEventsDrained(); + + expect(getDiagnosticStabilitySnapshot({ type: "model.call.started" }).events).toEqual([ + expect.objectContaining({ + type: "model.call.started", + provider: "openai", + model: "gpt-5.4", + }), + ]); + expect(JSON.stringify(getDiagnosticStabilitySnapshot())).not.toContain("private-"); + }); + it("keeps async queue drop summaries after drained queued events for sinceSeq polling", async () => { startDiagnosticStabilityRecorder(); diff --git a/src/logging/diagnostic-stability.ts b/src/logging/diagnostic-stability.ts index 39da2719b919..b82aaaa18973 100644 --- a/src/logging/diagnostic-stability.ts +++ b/src/logging/diagnostic-stability.ts @@ -852,7 +852,15 @@ export function startDiagnosticStabilityRecorder(): void { if (event.type === "telemetry.exporter") { return; } - if (metadata.trusted || event.type === "log.record") { + // Model-call instrumentation is trusted core telemetry required by recovery. + // Other trusted events retain their dedicated owners outside this ring. + if ( + (metadata.trusted && + event.type !== "model.call.started" && + event.type !== "model.call.completed" && + event.type !== "model.call.error") || + event.type === "log.record" + ) { return; } appendRecord(sanitizeDiagnosticEvent(event)); diff --git a/src/logging/diagnostic.test.ts b/src/logging/diagnostic.test.ts index bcd5a96a99a8..0c26d04207e2 100644 --- a/src/logging/diagnostic.test.ts +++ b/src/logging/diagnostic.test.ts @@ -1075,6 +1075,12 @@ describe("stuck session diagnostics threshold", () => { for (let attempt = 2; attempt <= 6; attempt += 1) { vi.advanceTimersByTime(30_000); + logSessionStateChange({ + sessionId: "s1", + sessionKey: "main", + state: "processing", + reason: "run_started", + }); markDiagnosticModelStartedForTest({ sessionId: "s1", sessionKey: "main", @@ -1107,6 +1113,65 @@ describe("stuck session diagnostics threshold", () => { ); }); + it("does not recover repeated requests after semantic output resets the clock", () => { + const events: DiagnosticEventPayload[] = []; + const recoverStuckSession = vi.fn(); + const stuckSessionAbortMs = 90_000; + const unsubscribe = onDiagnosticEvent((event) => { + events.push(event); + }); + try { + startDiagnosticHeartbeat( + { diagnostics: { enabled: true } }, + { + recoverStuckSession, + testTimings: { stuckSessionWarnMs: 30_000, stuckSessionAbortMs }, + }, + ); + const ref = { sessionId: "s1", sessionKey: "main", runId: "run-1" }; + logSessionStateChange({ ...ref, state: "processing" }); + markDiagnosticEmbeddedRunStarted(ref); + markDiagnosticModelStartedForTest({ + ...ref, + provider: "mock", + model: "retrying-model", + observationUnit: "request", + }); + vi.advanceTimersByTime(30_000); + markDiagnosticModelStartedForTest({ + ...ref, + provider: "mock", + model: "retrying-model", + observationUnit: "request", + }); + markDiagnosticRunProgressForTest({ + ...ref, + reason: "assistant:progress", + progressKind: "semantic", + }); + + for (let elapsedMs = 0; elapsedMs < stuckSessionAbortMs; elapsedMs += 30_000) { + vi.advanceTimersByTime(30_000); + markDiagnosticRunProgressForTest({ + ...ref, + reason: "model_call:stream_progress", + progressKind: "liveness", + }); + } + } finally { + unsubscribe(); + } + + expect( + events.some( + (event) => + event.type === "session.stalled" && + event.reason === "repeated_model_requests_without_progress", + ), + ).toBe(false); + expect(recoverStuckSession).not.toHaveBeenCalled(); + }); + it("reports silent model calls as long-running before the abort threshold", async () => { const events: DiagnosticEventPayload[] = []; const recoverStuckSession = vi.fn(); diff --git a/src/logging/diagnostic.ts b/src/logging/diagnostic.ts index 12b97a908368..b4273483ae93 100644 --- a/src/logging/diagnostic.ts +++ b/src/logging/diagnostic.ts @@ -1325,13 +1325,17 @@ export function startDiagnosticHeartbeat( activity, staleMs: stuckSessionWarnMs, }); + const repeatedRequestAttention = + state.state === "processing" && + (activity.repeatedRequestNoProgressAgeMs ?? 0) > stuckSessionWarnMs; if ( (state.state === "processing" && ageMs > stuckSessionWarnMs) || + repeatedRequestAttention || idleQueuedRecoverableStall ) { const attentionAgeMs = idleQueuedRecoverableStall ? (activity.lastProgressAgeMs ?? ageMs) - : ageMs; + : Math.max(ageMs, activity.repeatedRequestNoProgressAgeMs ?? 0); const classification = logSessionAttention({ sessionId: state.sessionId, sessionKey: state.sessionKey, diff --git a/test/e2e/qa-lab/runtime/gateway-repeated-request-recovery.e2e.test.ts b/test/e2e/qa-lab/runtime/gateway-repeated-request-recovery.e2e.test.ts new file mode 100644 index 000000000000..afe450eac0ac --- /dev/null +++ b/test/e2e/qa-lab/runtime/gateway-repeated-request-recovery.e2e.test.ts @@ -0,0 +1,302 @@ +import { randomUUID } from "node:crypto"; +import { setTimeout as sleep } from "node:timers/promises"; +import { afterEach, describe, expect, it } from "vitest"; +import { startQaLiveLaneGateway } from "../../../../extensions/qa-lab/runtime-api.js"; + +type StabilityEvent = { + seq?: unknown; + type?: unknown; + action?: unknown; + reason?: unknown; + outcome?: unknown; + ageMs?: unknown; +}; + +type StabilitySnapshot = { + lastSeq?: unknown; + events?: StabilityEvent[]; +}; + +type GatewayChatRun = { + runId?: unknown; + status?: unknown; + stopReason?: unknown; +}; + +type GatewayChatMessage = { + role?: unknown; + content?: unknown; +}; + +type GatewayChatHistory = { + messages?: GatewayChatMessage[]; +}; + +const RECOVERY_PROMPT = + "Repeated request recovery Gateway QA check. Keep attempting without producing a reply."; +const QUEUED_PROMPT = + "Repeated request queued reply Gateway QA check. Reply with the fixture marker."; +const QUEUED_REPLY_MARKER = "GATEWAY_REPEATED_REQUEST_QUEUED_OK"; +const RECOVERY_REASON = "repeated_model_requests_without_progress"; +const PRODUCTION_RECOVERY_BOUND_MS = 360_000; +const HISTORY_RETRY_TIMEOUT_MS = 10_000; +const HISTORY_RETRY_INTERVAL_MS = 250; + +let harness: Awaited> | undefined; + +afterEach(async () => { + await harness?.stop().catch(() => undefined); + harness = undefined; +}); + +async function readStability( + gateway: Awaited>["gateway"], + sinceSeq?: number, +): Promise { + return (await gateway.call( + "diagnostics.stability", + { limit: 1000, ...(sinceSeq === undefined ? {} : { sinceSeq }) }, + { timeoutMs: 10_000 }, + )) as StabilitySnapshot; +} + +async function waitForStability( + gateway: Awaited>["gateway"], + sinceSeq: number, + predicate: (events: StabilityEvent[]) => boolean, + timeoutMs: number, +): Promise { + const startedAt = Date.now(); + let latest: StabilityEvent[] = []; + while (Date.now() - startedAt < timeoutMs) { + latest = (await readStability(gateway, sinceSeq)).events ?? []; + if (predicate(latest)) { + return latest; + } + await sleep(1_000); + } + throw new Error( + `timed out after ${timeoutMs}ms waiting for stability events: ${JSON.stringify(latest)}`, + ); +} + +function messageText(message: GatewayChatMessage): string { + if (typeof message.content === "string") { + return message.content.trim(); + } + if (!Array.isArray(message.content)) { + return ""; + } + return message.content + .flatMap((part) => + part && typeof part === "object" && typeof (part as { text?: unknown }).text === "string" + ? [(part as { text: string }).text] + : [], + ) + .join("\n") + .trim(); +} + +function historyContainsQueuedReply(history: GatewayChatHistory): boolean { + const messages = history.messages ?? []; + const queuedIndex = messages.findLastIndex( + (message) => message.role === "user" && messageText(message).includes(QUEUED_PROMPT), + ); + return ( + queuedIndex >= 0 && + messages + .slice(queuedIndex + 1) + .some( + (message) => message.role === "assistant" && messageText(message) === QUEUED_REPLY_MARKER, + ) + ); +} + +function resolveRetryableHistoryDelayMs(error: unknown): number | null { + let current = error; + for (let depth = 0; depth < 4; depth += 1) { + if (typeof current !== "object" || current === null || Array.isArray(current)) { + break; + } + const shaped = current as { + cause?: unknown; + code?: unknown; + details?: unknown; + gatewayCode?: unknown; + retryable?: unknown; + retryAfterMs?: unknown; + }; + const code = shaped.gatewayCode ?? shaped.code; + if (code === "UNAVAILABLE" && shaped.retryable === true) { + const detailMethod = + typeof shaped.details === "object" && shaped.details !== null + ? (shaped.details as { method?: unknown }).method + : undefined; + if (typeof detailMethod !== "string" || detailMethod === "chat.history") { + return typeof shaped.retryAfterMs === "number" && Number.isFinite(shaped.retryAfterMs) + ? Math.max(100, Math.min(Math.floor(shaped.retryAfterMs), 5_000)) + : HISTORY_RETRY_INTERVAL_MS; + } + } + current = shaped.cause; + } + return null; +} + +async function waitForQueuedReply( + gateway: Awaited>["gateway"], + sessionKey: string, +): Promise { + const startedAt = Date.now(); + let latestHistory: GatewayChatHistory = {}; + let lastRetryableError: unknown; + while (Date.now() - startedAt < HISTORY_RETRY_TIMEOUT_MS) { + let delayMs = HISTORY_RETRY_INTERVAL_MS; + try { + const history = (await gateway.call( + "chat.history", + { sessionKey, limit: 20 }, + { timeoutMs: 10_000 }, + )) as GatewayChatHistory; + latestHistory = history; + lastRetryableError = undefined; + if (historyContainsQueuedReply(history)) { + return history; + } + } catch (error) { + const retryDelayMs = resolveRetryableHistoryDelayMs(error); + if (retryDelayMs === null) { + throw error; + } + lastRetryableError = error; + delayMs = retryDelayMs; + } + const remainingMs = HISTORY_RETRY_TIMEOUT_MS - (Date.now() - startedAt); + if (remainingMs <= 0) { + break; + } + await sleep(Math.min(delayMs, remainingMs)); + } + const observed = (latestHistory.messages ?? []).map((message) => ({ + role: message.role, + text: messageText(message), + })); + const message = `timed out waiting for queued reply in chat.history after ${HISTORY_RETRY_TIMEOUT_MS}ms: ${JSON.stringify(observed)}`; + throw lastRetryableError === undefined + ? new Error(message) + : new Error(message, { cause: lastRetryableError }); +} + +describe("Gateway repeated-request recovery", () => { + it( + "aborts the real stalled owner once and releases one queued followup", + { timeout: 510_000 }, + async () => { + harness = await startQaLiveLaneGateway({ + repoRoot: process.cwd(), + providerMode: "mock-openai", + primaryModel: "mock-openai/gpt-5.6-luna", + alternateModel: "mock-openai/gpt-5.6-luna-alt", + transport: { + requiredPluginIds: [], + createGatewayConfig: () => ({ + messages: { queue: { mode: "followup" } }, + }), + }, + transportBaseUrl: "http://127.0.0.1", + controlUiEnabled: false, + mutateConfig: (config) => ({ ...config, diagnostics: { enabled: true } }), + }); + const { gateway } = harness; + + const baseline = await readStability(gateway); + const baselineSeq = typeof baseline.lastSeq === "number" ? baseline.lastSeq : 0; + const sessionKey = `agent:qa:repeated-request-recovery-${randomUUID()}`; + const active = (await gateway.call( + "chat.send", + { + sessionKey, + message: RECOVERY_PROMPT, + deliver: false, + idempotencyKey: randomUUID(), + }, + { timeoutMs: 30_000 }, + )) as GatewayChatRun; + expect(active).toMatchObject({ status: "started" }); + expect(typeof active.runId).toBe("string"); + + await waitForStability( + gateway, + baselineSeq, + (events) => events.filter((event) => event.type === "model.call.started").length >= 2, + 150_000, + ); + + const queued = (await gateway.call( + "chat.send", + { + sessionKey, + message: QUEUED_PROMPT, + queueMode: "followup", + deliver: false, + idempotencyKey: randomUUID(), + }, + { timeoutMs: 30_000 }, + )) as GatewayChatRun; + expect(queued).toMatchObject({ status: "started" }); + expect(typeof queued.runId).toBe("string"); + + const events = await waitForStability( + gateway, + baselineSeq, + (records) => records.some((event) => event.type === "session.recovery.completed"), + 350_000, + ); + const stalled = events.filter( + (event) => event.type === "session.stalled" && event.reason === RECOVERY_REASON, + ); + const requested = events.filter( + (event) => event.type === "session.recovery.requested" && event.reason === RECOVERY_REASON, + ); + const completed = events.filter((event) => event.type === "session.recovery.completed"); + + expect(stalled).toHaveLength(1); + expect(stalled[0]?.ageMs).toEqual(expect.any(Number)); + expect(stalled[0]?.ageMs as number).toBeGreaterThanOrEqual(PRODUCTION_RECOVERY_BOUND_MS); + expect(requested).toEqual([ + expect.objectContaining({ action: "abort", reason: RECOVERY_REASON }), + ]); + expect(completed).toEqual([ + expect.objectContaining({ action: "abort_embedded_run", outcome: "aborted" }), + ]); + expect( + events.filter((event) => event.type === "model.call.started").length, + ).toBeGreaterThanOrEqual(3); + + const activeTerminal = (await gateway.call( + "agent.wait", + { runId: active.runId, timeoutMs: 30_000 }, + { timeoutMs: 35_000 }, + )) as GatewayChatRun; + expect(activeTerminal.status).not.toBe("ok"); + + const queuedTerminal = (await gateway.call( + "agent.wait", + { runId: queued.runId, timeoutMs: 30_000 }, + { timeoutMs: 35_000 }, + )) as GatewayChatRun; + expect(queuedTerminal.status).toBe("ok"); + + const history = await waitForQueuedReply(gateway, sessionKey); + expect(historyContainsQueuedReply(history)).toBe(true); + + const finalEvents = (await readStability(gateway, baselineSeq)).events ?? []; + expect( + finalEvents.filter((event) => event.type === "session.recovery.requested"), + ).toHaveLength(1); + expect( + finalEvents.filter((event) => event.type === "session.recovery.completed"), + ).toHaveLength(1); + }, + ); +});