From dc93ea6d699a3ab30915fd3282fd12aa0ac8aeb2 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 4 Aug 2026 22:29:15 -0700 Subject: [PATCH] fix(qa): follow durable finals after streamed previews (#119378) * test(qa): isolate memory scenario evidence * test(qa): follow durable progress completion * test(qa): run doctor migration noninteractively * test(qa): split active memory follow-up * test(qa): wait for preview retirement * test(qa): require ordered preview retirement * test(qa): record ordered durable reply evidence * test(qa): preserve durable delivery budgets * test(qa): isolate durable lifecycle by account * test(qa): scope durable reply assertions by account * test(qa): preserve durable conversation identity --- .../src/scenario-catalog-causality.test.ts | 127 +++++++++++++++++- .../memory/memory-tools-channel-context.yaml | 42 ++++-- .../runtime/agent-progress-evidence.yaml | 44 ++++-- 3 files changed, 185 insertions(+), 28 deletions(-) diff --git a/extensions/qa-lab/src/scenario-catalog-causality.test.ts b/extensions/qa-lab/src/scenario-catalog-causality.test.ts index 04f9f8431b52..3df445cf9b04 100644 --- a/extensions/qa-lab/src/scenario-catalog-causality.test.ts +++ b/extensions/qa-lab/src/scenario-catalog-causality.test.ts @@ -1,6 +1,8 @@ import { describe, expect, it } from "vitest"; +import { createQaBusState } from "./bus-state.js"; import { readQaScenarioById, readQaScenarioExecutionConfig } from "./scenario-catalog.js"; import { readFlowAssertExpression, requireFlowScenario } from "./scenario-catalog.test-utils.js"; +import { runLoadedScenarioFlow } from "./scenario-flow-runner.test-support.js"; describe("qa scenario catalog causality", () => { it("loads live gateway sentinel scenarios for harness self-health", () => { @@ -144,24 +146,30 @@ describe("qa scenario catalog causality", () => { "thread-memory-isolation", "poll", "finalRequest.toolOutputCallId === searchResultRequest.plannedToolCallId", + null, ], [ "memory-tools-channel-context", "poll", "finalRequest.toolOutputCallId === searchResultRequest.plannedToolCallId", + "durableChannelLifecycle", ], [ "agent-tool-consumption", "immediate", "getResultRequest.toolOutputCallId === searchResultRequest.plannedToolCallId", + null, ], ] as const)( "asserts the complete memory tool chain before %s delivery", - (scenarioId, requestCollectionMode, finalLinkNeedle) => { + (scenarioId, requestCollectionMode, finalLinkNeedle, durableWaitSaveAs) => { const scenario = requireFlowScenario(readQaScenarioById(scenarioId)); const actions = scenario.execution.flow?.steps[0]?.actions ?? []; - const outboundIndex = actions.findIndex( - (action) => (action as { call?: string }).call === "waitForOutboundMessage", + const outboundIndex = actions.findIndex((action) => + durableWaitSaveAs + ? (action as { call?: string }).call === "waitForCondition" && + (action as { saveAs?: string }).saveAs === durableWaitSaveAs + : (action as { call?: string }).call === "waitForOutboundMessage", ); const requestCollectionIndex = actions.findIndex((action) => requestCollectionMode === "poll" @@ -193,6 +201,15 @@ describe("qa scenario catalog causality", () => { expect(finalRequestAssertIndex, scenarioId).toBeGreaterThan(searchResultAssertIndex); expect(outboundIndex, scenarioId).toBeGreaterThan(finalRequestAssertIndex); + if (durableWaitSaveAs) { + const durableWait = actions[outboundIndex] as + | { args?: Array<{ lambda?: { expr?: string } }> } + | undefined; + const durableExpr = durableWait?.args?.[0]?.lambda?.expr ?? ""; + expect(durableExpr, scenarioId).toContain("event.cursor < finalSent.cursor"); + expect(durableExpr, scenarioId).toContain("event.cursor < previewRetired.cursor"); + } + if (requestCollectionMode === "poll") { const requestPoll = actions[requestCollectionIndex] as | { args?: Array<{ lambda?: { expr?: string } }> } @@ -212,4 +229,108 @@ describe("qa scenario catalog causality", () => { } }, ); + + it.each([ + ["memory-tools-channel-context", "durableChannelLifecycle", 30000], + ["agent-progress-evidence", "durableCompletionLifecycle", 60000], + ] as const)("keeps the policy-aware durable delivery budget for %s", (scenarioId, saveAs, ms) => { + const scenario = requireFlowScenario(readQaScenarioById(scenarioId)); + const actions = scenario.execution.flow?.steps[0]?.actions ?? []; + const durableWait = actions.find( + (action) => + (action as { call?: string }).call === "waitForCondition" && + (action as { saveAs?: string }).saveAs === saveAs, + ); + + expect(durableWait, scenarioId).toMatchObject({ + args: [expect.any(Object), { expr: `liveTurnTimeoutMs(env, ${ms})` }], + }); + }); + + it.each([ + { + scenarioId: "memory-tools-channel-context", + saveAs: "durableChannelLifecycle", + cursorName: "busCursorBeforeInbound", + conversationKey: "channelId", + markerKey: "expectedNeedle", + targetPrefix: "channel", + }, + { + scenarioId: "agent-progress-evidence", + saveAs: "durableCompletionLifecycle", + cursorName: "busCursorBefore", + conversationKey: "conversationId", + markerKey: "completionText", + targetPrefix: "dm", + }, + ] as const)("isolates $scenarioId durable lifecycle evidence by account", async (fixture) => { + const scenario = requireFlowScenario(readQaScenarioById(fixture.scenarioId)); + const actions = scenario.execution.flow?.steps[0]?.actions ?? []; + const durableWaitIndex = actions.findIndex( + (action) => + (action as { call?: string }).call === "waitForCondition" && + (action as { saveAs?: string }).saveAs === fixture.saveAs, + ); + const cardinalityAssertIndex = actions.findIndex((action) => + readFlowAssertExpression(action).includes( + fixture.scenarioId === "memory-tools-channel-context" + ? "visibleChannelOutbounds.length === 1" + : "completionMessages.length === 1", + ), + ); + expect(durableWaitIndex, fixture.scenarioId).toBeGreaterThanOrEqual(0); + expect(cardinalityAssertIndex, fixture.scenarioId).toBeGreaterThan(durableWaitIndex); + if (durableWaitIndex < 0 || cardinalityAssertIndex <= durableWaitIndex) { + throw new Error(`missing durable lifecycle assertion path for ${fixture.scenarioId}`); + } + const postWaitAssertionPath = actions.slice(durableWaitIndex, cardinalityAssertIndex + 1); + + const config = scenario.execution.config ?? {}; + const conversationId = String(config[fixture.conversationKey]); + const marker = String(config[fixture.markerKey]); + const target = `${fixture.targetPrefix}:${conversationId}`; + const state = createQaBusState(); + for (const accountId of ["foreign", "qa-channel"]) { + const preview = state.addOutboundMessage({ accountId, to: target, text: marker }); + state.deleteMessage({ accountId, messageId: preview.id }); + state.addOutboundMessage({ accountId, to: target, text: marker }); + } + const foreignKind = fixture.targetPrefix === "dm" ? "channel" : "dm"; + const foreignKindTarget = `${foreignKind}:${conversationId}`; + const foreignKindPreview = state.addOutboundMessage({ + accountId: "qa-channel", + to: foreignKindTarget, + text: marker, + }); + state.deleteMessage({ accountId: "qa-channel", messageId: foreignKindPreview.id }); + state.addOutboundMessage({ + accountId: "qa-channel", + to: foreignKindTarget, + text: marker, + }); + + await expect( + runLoadedScenarioFlow(fixture.scenarioId, { + state, + flow: { + steps: [ + { + name: "keeps foreign account lifecycle evidence isolated", + actions: [ + { set: "outboundStartIndex", value: { expr: "0" } }, + { set: fixture.cursorName, value: { expr: "0" } }, + ...postWaitAssertionPath, + { + assert: { + expr: `${fixture.saveAs}.message.accountId === transport.accountId`, + }, + }, + ], + }, + ], + }, + }), + ).resolves.toMatchObject({ status: "pass" }); + }); }); diff --git a/qa/scenarios/memory/memory-tools-channel-context.yaml b/qa/scenarios/memory/memory-tools-channel-context.yaml index c160a5af9e45..64dc3b5da0b3 100644 --- a/qa/scenarios/memory/memory-tools-channel-context.yaml +++ b/qa/scenarios/memory/memory-tools-channel-context.yaml @@ -79,6 +79,9 @@ flow: - set: outboundStartIndex value: expr: "state.getSnapshot().messages.length" + - set: busCursorBeforeInbound + value: + expr: "state.getSnapshot().cursor" - sendInbound: conversation: id: @@ -123,27 +126,38 @@ flow: expr: "finalRequest.cursor > searchResultRequest.cursor && finalRequest.toolOutputCallId === searchResultRequest.plannedToolCallId && finalRequest.toolOutputStructuredError !== true && String(finalRequest.toolOutput ?? '').includes(config.expectedMemoryPath) && String(finalRequest.toolOutput ?? '').includes(config.expectedNeedle) && !finalRequest.plannedToolName" message: expr: "`final request did not consume the matching successful memory_get result: ${JSON.stringify(finalRequest)}`" - - call: waitForOutboundMessage - saveAs: outbound + - call: waitForCondition + saveAs: durableChannelLifecycle args: - - ref: state - lambda: - params: [candidate] - expr: "candidate.direction === 'outbound' && candidate.conversation.id === config.channelId && candidate.conversation.kind === 'channel' && candidate.text.includes(config.expectedNeedle)" + expr: |- + (() => { + const snapshot = state.getSnapshot(); + const visible = snapshot.messages.slice(outboundStartIndex).filter((message) => message.accountId === transport.accountId && message.direction === "outbound" && !message.deleted && message.conversation.id === config.channelId && message.conversation.kind === "channel" && message.text.includes(config.expectedNeedle)); + if (visible.length !== 1) return undefined; + const candidate = visible[0]; + const events = snapshot.events.filter((event) => event.cursor > busCursorBeforeInbound && "message" in event && event.message.accountId === transport.accountId && event.message.direction === "outbound" && event.message.conversation.id === config.channelId && event.message.conversation.kind === "channel" && event.message.text.includes(config.expectedNeedle)); + const finalSent = events.find((event) => event.kind === "outbound-message" && event.message.id === candidate.id); + if (!finalSent) return undefined; + const previewRetired = events.find((event) => event.kind === "message-deleted" && event.message.id !== candidate.id && event.cursor < finalSent.cursor); + if (!previewRetired) return undefined; + const previewSent = events.find((event) => event.kind === "outbound-message" && event.message.id === previewRetired.message.id && event.cursor < previewRetired.cursor); + return previewSent ? { message: candidate, previewId: previewRetired.message.id, finalId: candidate.id, previewSentCursor: previewSent.cursor, previewDeletedCursor: previewRetired.cursor, finalSentCursor: finalSent.cursor } : undefined; + })() - expr: liveTurnTimeoutMs(env, 30000) - - sinceIndex: - ref: outboundStartIndex - - call: sleep - args: [4000] + - set: durableChannelOutbound + value: + expr: durableChannelLifecycle.message - set: visibleChannelOutbounds value: - expr: "state.getSnapshot().messages.slice(outboundStartIndex).filter((message) => message.direction === 'outbound' && !message.deleted && message.conversation.id === config.channelId && message.conversation.kind === 'channel')" + expr: "state.getSnapshot().messages.slice(outboundStartIndex).filter((message) => message.accountId === transport.accountId && message.direction === 'outbound' && !message.deleted && message.conversation.id === config.channelId && message.conversation.kind === 'channel')" - assert: - expr: "visibleChannelOutbounds.length === 1 && visibleChannelOutbounds[0].id === outbound.id" + # Tool-backed finals replace their streamed preview; prove sent -> deleted -> final ordering first. + expr: "visibleChannelOutbounds.length === 1 && visibleChannelOutbounds[0].id === durableChannelOutbound.id" message: expr: "`expected exactly one visible QA-room reply after the ordered memory result: ${JSON.stringify(visibleChannelOutbounds)}`" - assert: - expr: "(outbound.text.match(new RegExp(config.expectedNeedle, 'g')) ?? []).length === 1" + expr: "(durableChannelOutbound.text.match(new RegExp(config.expectedNeedle, 'g')) ?? []).length === 1" message: - expr: "`final QA-room reply must contain ${config.expectedNeedle} exactly once: ${outbound.text}`" - detailsExpr: "`${outbound.text}; path=${config.expectedMemoryPath}; matched=${searchResultRequest.toolOutputCallId === searchPlanRequest.plannedToolCallId}; room=${config.channelId}`" + expr: "`final QA-room reply must contain ${config.expectedNeedle} exactly once: ${durableChannelOutbound.text}`" + detailsExpr: "`${durableChannelOutbound.text}; path=${config.expectedMemoryPath}; matched=${searchResultRequest.toolOutputCallId === searchPlanRequest.plannedToolCallId}; room=${config.channelId}; lifecycle=${JSON.stringify({ previewId: durableChannelLifecycle.previewId, previewSentCursor: durableChannelLifecycle.previewSentCursor, previewDeletedCursor: durableChannelLifecycle.previewDeletedCursor, finalId: durableChannelLifecycle.finalId, finalSentCursor: durableChannelLifecycle.finalSentCursor })}`" diff --git a/qa/scenarios/runtime/agent-progress-evidence.yaml b/qa/scenarios/runtime/agent-progress-evidence.yaml index bdaba2aea150..807a34e207b9 100644 --- a/qa/scenarios/runtime/agent-progress-evidence.yaml +++ b/qa/scenarios/runtime/agent-progress-evidence.yaml @@ -132,7 +132,6 @@ flow: ref: config.completionText timeoutMs: expr: liveTurnTimeoutMs(env, 60000) - saveAs: outbound - call: fs.readFile saveAs: artifact args: @@ -184,39 +183,62 @@ flow: expr: "writeResultRequest.toolOutputCallId === writeRequest.plannedToolCallId && writeResultRequest.toolOutputStructuredError !== true && /successfully (?:wrote|created|updated|replaced)/i.test(String(writeResultRequest.toolOutput ?? '')) && !writeResultRequest.plannedToolName" message: expr: "`request 4 did not consume the successful write result before terminal generation: ${JSON.stringify(writeResultRequest)}`" + - call: waitForCondition + saveAs: durableCompletionLifecycle + args: + - lambda: + expr: |- + (() => { + const snapshot = state.getSnapshot(); + const visible = snapshot.messages.filter((message) => message.accountId === transport.accountId && message.direction === "outbound" && !message.deleted && message.conversation.id === config.conversationId && message.conversation.kind === "direct" && message.text.includes(config.completionText)); + if (visible.length !== 1) return undefined; + const candidate = visible[0]; + const events = snapshot.events.filter((event) => event.cursor > busCursorBefore && "message" in event && event.message.accountId === transport.accountId && event.message.direction === "outbound" && event.message.conversation.id === config.conversationId && event.message.conversation.kind === "direct" && event.message.text.includes(config.completionText)); + const finalSent = events.find((event) => event.kind === "outbound-message" && event.message.id === candidate.id); + if (!finalSent) return undefined; + const previewRetired = events.find((event) => event.kind === "message-deleted" && event.message.id !== candidate.id && event.cursor < finalSent.cursor); + if (!previewRetired) return undefined; + const previewSent = events.find((event) => event.kind === "outbound-message" && event.message.id === previewRetired.message.id && event.cursor < previewRetired.cursor); + return previewSent ? { message: candidate, previewId: previewRetired.message.id, finalId: candidate.id, previewSentCursor: previewSent.cursor, previewDeletedCursor: previewRetired.cursor, finalSentCursor: finalSent.cursor } : undefined; + })() + - expr: liveTurnTimeoutMs(env, 60000) + - set: durableCompletion + value: + expr: durableCompletionLifecycle.message - set: allBusEvents value: - expr: "state.getSnapshot().events.filter((event) => event.cursor > busCursorBefore && 'message' in event && event.message.direction === 'outbound' && event.message.conversation.id === config.conversationId)" + expr: "state.getSnapshot().events.filter((event) => event.cursor > busCursorBefore && 'message' in event && event.message.accountId === transport.accountId && event.message.direction === 'outbound' && event.message.conversation.id === config.conversationId && event.message.conversation.kind === 'direct')" - set: busEvents value: expr: "allBusEvents.filter((event) => event.kind === 'outbound-message' || event.kind === 'message-edited')" - set: completionEvents value: expr: "busEvents.filter((event) => event.message.text.includes(config.completionText))" - - set: finalCompletionEvent - value: - expr: "completionEvents.findLast((event) => event.message.id === outbound.id)" - set: completionMessages value: - expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound' && message.conversation.id === config.conversationId && message.text.includes(config.completionText) && !message.deleted)" + expr: "state.getSnapshot().messages.filter((message) => message.accountId === transport.accountId && message.direction === 'outbound' && message.conversation.id === config.conversationId && message.conversation.kind === 'direct' && message.text.includes(config.completionText) && !message.deleted)" - assert: - expr: "completionMessages.length === 1 && completionMessages[0].id === outbound.id" + # Tool-backed finals replace their streamed preview; prove sent -> deleted -> final ordering first. + expr: "completionMessages.length === 1 && completionMessages[0].id === durableCompletion.id" message: expr: "`expected exactly one durable completion reply: ${JSON.stringify(completionMessages)}`" + - set: finalCompletionEvent + value: + expr: "completionEvents.findLast((event) => event.message.id === durableCompletion.id)" - assert: expr: "completionEvents.length >= 1 && finalCompletionEvent && completionEvents.every((event) => artifactStat.mtimeMs <= (event.message.editedAt ?? event.message.timestamp))" message: expr: "`completion appeared before the write-backed terminal request: artifactMtime=${artifactStat.mtimeMs} events=${JSON.stringify(completionEvents)}`" - assert: - expr: "completionEvents.filter((event) => event.message.id !== outbound.id).every((event) => allBusEvents.some((candidate) => candidate.kind === 'message-deleted' && candidate.cursor > event.cursor && candidate.message.id === event.message.id))" + expr: "completionEvents.filter((event) => event.message.id !== durableCompletion.id).every((event) => allBusEvents.some((candidate) => candidate.kind === 'message-deleted' && candidate.cursor > event.cursor && candidate.message.id === event.message.id))" message: expr: "`a transient completion preview remained durable: ${JSON.stringify(allBusEvents)}`" - assert: - expr: "outbound.replyToId === inbound.id && outbound.text.includes(config.artifactFile) && outbound.text.includes(config.completionMarker)" + expr: "durableCompletion.replyToId === inbound.id && durableCompletion.text.includes(config.artifactFile) && durableCompletion.text.includes(config.completionMarker)" message: - expr: "`durable completion did not cite the artifact and marker: ${JSON.stringify(outbound)}`" + expr: "`durable completion did not cite the artifact and marker: ${JSON.stringify(durableCompletion)}`" - assert: expr: "artifactStat.mtimeMs <= (finalCompletionEvent.message.editedAt ?? finalCompletionEvent.message.timestamp)" message: expr: "`artifact postdated the durable completion reply: artifactMtime=${artifactStat.mtimeMs} replyTimestamp=${finalCompletionEvent.message.editedAt ?? finalCompletionEvent.message.timestamp}`" - detailsExpr: "`read:${firstReadRequest.plannedToolCallId} -> read:${secondReadRequest.plannedToolCallId} -> write:${writeRequest.plannedToolCallId} -> result:${writeResultRequest.toolOutputCallId}; artifactBeforeReply=${artifactStat.mtimeMs <= (finalCompletionEvent.message.editedAt ?? finalCompletionEvent.message.timestamp)}; durableCompletions=${completionMessages.length}`" + detailsExpr: "`read:${firstReadRequest.plannedToolCallId} -> read:${secondReadRequest.plannedToolCallId} -> write:${writeRequest.plannedToolCallId} -> result:${writeResultRequest.toolOutputCallId}; artifactBeforeReply=${artifactStat.mtimeMs <= (finalCompletionEvent.message.editedAt ?? finalCompletionEvent.message.timestamp)}; durableCompletions=${completionMessages.length}; lifecycle=${JSON.stringify({ previewId: durableCompletionLifecycle.previewId, previewSentCursor: durableCompletionLifecycle.previewSentCursor, previewDeletedCursor: durableCompletionLifecycle.previewDeletedCursor, finalId: durableCompletionLifecycle.finalId, finalSentCursor: durableCompletionLifecycle.finalSentCursor })}`"