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

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" });
+  });
 });

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 })}`"

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 })}`"