fix(qa): resume channel cursor after gateway restart (#111096)

extensions/qa-lab/src/bus-server.test.ts

@@ -137,6 +137,69 @@ describe("qa-bus server", () => {
     });
   });
 
+  it("resumes an account after its last acknowledged cursor when the client restarts", async () => {
+    const state = createQaBusState();
+    const bus = await startQaBusServer({ state });
+    stops.push(bus["stop"]);
+
+    const consumed = state.addInboundMessage({
+      accountId: "acct-a",
+      conversation: { id: "first", kind: "direct" },
+      senderId: "acct-a-user",
+      text: "consumed before restart",
+    });
+    const firstPoll = await pollQaBus({
+      baseUrl: bus.baseUrl,
+      accountId: "acct-a",
+      cursor: 0,
+      timeoutMs: 0,
+    });
+    expect(firstPoll.events.map((event) => event.cursor)).toEqual([1]);
+
+    await pollQaBus({
+      baseUrl: bus.baseUrl,
+      accountId: "acct-a",
+      cursor: firstPoll.cursor,
+      timeoutMs: 0,
+    });
+    expect(state.getAcknowledgedPollCursor("acct-a")).toBe(firstPoll.cursor);
+    const queuedDuringRestart = state.addInboundMessage({
+      accountId: "acct-a",
+      conversation: { id: "second", kind: "direct" },
+      senderId: "acct-a-user",
+      text: "queued during restart",
+    });
+
+    const restartedPoll = await pollQaBus({
+      baseUrl: bus.baseUrl,
+      accountId: "acct-a",
+      cursor: 0,
+      timeoutMs: 0,
+    });
+    const restartedMessageIds = restartedPoll.events.flatMap((event) =>
+      "message" in event ? [event.message.id] : [],
+    );
+    expect(restartedMessageIds).toEqual([queuedDuringRestart.id]);
+    expect(restartedMessageIds).not.toContain(consumed.id);
+
+    state.reset();
+    const queuedAfterReset = state.addInboundMessage({
+      accountId: "acct-a",
+      conversation: { id: "third", kind: "direct" },
+      senderId: "acct-a-user",
+      text: "queued after bus reset",
+    });
+    const resetPoll = await pollQaBus({
+      baseUrl: bus.baseUrl,
+      accountId: "acct-a",
+      cursor: 0,
+      timeoutMs: 0,
+    });
+    expect(
+      resetPoll.events.flatMap((event) => ("message" in event ? [event.message.id] : [])),
+    ).toEqual([queuedAfterReset.id]);
+  });
+
   it("rejects malformed poll numeric fields before long-polling", async () => {
     const state = createQaBusState();
     const bus = await startQaBusServer({ state });

extensions/qa-lab/src/bus-server.ts

@@ -233,12 +233,16 @@ export async function handleQaBusRequest(params: {
         return true;
       case "/v1/poll": {
         const input = normalizeQaBusPollInput(body);
+        const pollInput = {
+          ...input,
+          cursor: params.state.resolvePollCursor(input),
+        };
         const timeoutMs = input.timeoutMs ?? 0;
         const accountId = normalizeAccountId(input.accountId);
-        const initial = params.state.poll(input);
+        const initial = params.state.poll(pollInput);
         const effectiveStartCursor = resolveQaBusPollStartCursor({
           currentCursor: initial.cursor,
-          requestedCursor: input.cursor,
+          requestedCursor: pollInput.cursor,
         });
         if (initial.events.length > 0 || timeoutMs === 0) {
           writeJson(params.res, 200, initial);
@@ -253,7 +257,7 @@ export async function handleQaBusRequest(params: {
         } catch {
           // timeout ok for long-poll
         }
-        writeJson(params.res, 200, params.state.poll(input));
+        writeJson(params.res, 200, params.state.poll(pollInput));
         return true;
       }
       case "/v1/wait":

extensions/qa-lab/src/bus-state.ts

@@ -75,6 +75,7 @@ export function createQaBusState() {
   const threads = new Map<string, QaBusThread>();
   const messages = new Map<string, QaBusMessage>();
   const events: QaBusEvent[] = [];
+  const acknowledgedPollCursors = new Map<string, number>();
   let cursor = 0;
   const waiters = createQaBusWaiterStore(() =>
     buildQaBusSnapshot({
@@ -161,7 +162,7 @@ export function createQaBusState() {
       messages.clear();
       events.length = 0;
       // Keep the cursor monotonic across resets so long-poll clients do not
-      // miss fresh events after the bus is cleared mid-session.
+      // miss fresh events and retained restart acknowledgements remain valid.
       waiters.reset();
     },
     getSnapshot() {
@@ -289,6 +290,20 @@ export function createQaBusState() {
     searchMessages(input: QaBusSearchMessagesInput) {
       return searchQaBusMessages({ messages, input });
     },
+    resolvePollCursor(input: QaBusPollInput = {}) {
+      const accountId = normalizeAccountId(input.accountId);
+      const requestedCursor = input.cursor ?? 0;
+      const acknowledgedCursor = acknowledgedPollCursors.get(accountId) ?? 0;
+      if (requestedCursor > acknowledgedCursor && requestedCursor <= cursor) {
+        acknowledgedPollCursors.set(accountId, requestedCursor);
+      }
+      // A restarted channel consumer begins at zero. Resume its account cursor
+      // so retained events are not replayed, while still returning unacked work.
+      return requestedCursor === 0 ? acknowledgedCursor : requestedCursor;
+    },
+    getAcknowledgedPollCursor(accountId?: string) {
+      return acknowledgedPollCursors.get(normalizeAccountId(accountId)) ?? 0;
+    },
     poll(input: QaBusPollInput = {}) {
       return pollQaBusEvents({ events, cursor, input });
     },

extensions/qa-lab/src/scenario-flow-runner.test.ts

@@ -44,6 +44,7 @@ async function runLoadedScenarioFlow(
   const state = params.state ?? createQaBusState();
   let waitCount = 0;
   const transport = {
+    accountId: "qa-channel",
     state,
     reset: async () => {
       state.reset();
@@ -80,6 +81,10 @@ async function runLoadedScenarioFlow(
             (!input.textIncludes || candidate.text.includes(input.textIncludes)),
         );
       if (match) {
+        state.resolvePollCursor({
+          accountId: "qa-channel",
+          cursor: state.getSnapshot().cursor,
+        });
         return match;
       }
       throw new Error(`timed out after ${input.timeoutMs}ms waiting for outbound marker`);
@@ -93,7 +98,14 @@ async function runLoadedScenarioFlow(
         }),
   };
   const api = {
-    env: { providerMode: "mock-openai" },
+    env: {
+      providerMode: "mock-openai",
+      gateway: {
+        restartAfterStateMutation: async (mutate: (context: unknown) => Promise<void>) => {
+          await mutate({});
+        },
+      },
+    },
     transport,
     state,
     scenario,
@@ -104,6 +116,15 @@ async function runLoadedScenarioFlow(
     waitForTransportReady: async () => undefined,
     waitForQaChannelReady: async () => undefined,
     waitForNoOutbound: async () => undefined,
+    waitForCondition: async <T>(check: () => T | Promise<T | undefined>) => {
+      for (let attempt = 0; attempt < 10; attempt += 1) {
+        const value = await check();
+        if (value !== undefined) {
+          return value;
+        }
+      }
+      throw new Error("test condition was not met");
+    },
     sleep: async () => undefined,
     reset: async () => {
       state.reset();

qa/scenarios/channels/qa-channel-reconnect-dedupe.yaml

@@ -9,10 +9,10 @@ scenario:
     secondary:
       - channels.dedup
       - runtime.delivery
-  objective: Verify qa-channel readiness polling keeps prior delivery stable and does not replay the last outbound message.
+  objective: Verify qa-channel resumes after a Gateway restart without replaying consumed inbound events or completed delivery.
   successCriteria:
     - Agent replies once before a reconnect-style readiness cycle.
-    - qa-channel reports ready again without replaying prior outbound delivery.
+    - qa-channel reports ready again without replaying prior inbound or outbound delivery.
     - Follow-up delivery produces one new reply without duplicating the first reply.
   docsRefs:
     - docs/channels/qa-channel.md
@@ -23,7 +23,7 @@ scenario:
     - extensions/qa-lab/src/suite-runtime-gateway.ts
   execution:
     kind: flow
-    summary: Verify qa-channel readiness recovery does not duplicate old outbound delivery.
+    summary: Verify qa-channel Gateway restart recovery does not replay consumed messages.
     channel: qa-channel
     config:
       firstPrompt: "@openclaw Reconnect dedupe setup marker. Reply exactly: RECONNECT-FIRST-OK"
@@ -44,78 +44,91 @@ flow:
             - ref: env
             - 60000
         - call: reset
-        - set: sessionKey
-          value:
-            expr: "`agent:qa:channel-reconnect:${randomUUID().slice(0, 8)}`"
-        - call: runAgentPrompt
-          args:
-            - ref: env
-            - sessionKey:
-                ref: sessionKey
-              to: channel:qa-room
-              message:
-                expr: config.firstPrompt
-              timeoutMs:
-                expr: liveTurnTimeoutMs(env, 45000)
-        - call: waitForOutboundMessage
-          saveAs: firstOutbound
-          args:
-            - ref: state
-            - lambda:
-                params: [candidate]
-                expr: "candidate.conversation.id === 'qa-room' && candidate.direction === 'outbound' && String(candidate.text ?? '').includes(config.firstMarker)"
-            - expr: liveTurnTimeoutMs(env, 60000)
-        - set: beforeRestartCursor
+        - sendInbound:
+            conversation:
+              id: qa-room
+              kind: group
+            senderId: qa-driver
+            senderName: QA Driver
+            text:
+              expr: config.firstPrompt
+        - waitForOutbound:
+            textIncludes:
+              ref: config.firstMarker
+            timeoutMs:
+              expr: liveTurnTimeoutMs(env, 60000)
+        - set: beforeRestartMessageIndex
           value:
             expr: state.getSnapshot().messages.length
-        - call: sleep
+        - set: beforeRestartOutboundCount
+          value:
+            expr: "state.getSnapshot().messages.filter((candidate) => candidate.direction === 'outbound').length"
+        - set: beforeRestartCursor
+          value:
+            expr: state.getSnapshot().cursor
+        - call: waitForCondition
           args:
-            - 1000
+            - lambda:
+                expr: "state.getAcknowledgedPollCursor(transport.accountId) >= beforeRestartCursor ? true : undefined"
+            - 10000
+            - 25
+        - assert:
+            expr: "typeof env.gateway.restartAfterStateMutation === 'function'"
+            message: qa gateway child does not expose restartAfterStateMutation
+        - call: env.gateway.restartAfterStateMutation
+          args:
+            - lambda:
+                async: true
+                params: [ctx]
+                expr: Promise.resolve()
+        - call: waitForGatewayHealthy
+          args:
+            - ref: env
+            - 60000
         - call: waitForQaChannelReady
           args:
             - ref: env
             - 60000
+        - waitForNoOutbound:
+            quietMs: 3000
+            sinceIndex:
+              ref: beforeRestartOutboundCount
         - set: firstMatchesBeforeFollowup
           value:
             expr: "state.getSnapshot().messages.filter((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === 'qa-room')"
         - assert:
             expr: "firstMatchesBeforeFollowup.length === 1 && String(firstMatchesBeforeFollowup[0]?.text ?? '').includes(config.firstMarker)"
             message:
-              expr: "`readiness cycle should preserve exactly one marked first reply, saw ${firstMatchesBeforeFollowup.length}; transcript=${formatTransportTranscript(state, { conversationId: 'qa-room' })}`"
-        - call: runAgentPrompt
-          args:
-            - ref: env
-            - sessionKey:
-                ref: sessionKey
-              to: channel:qa-room
-              message:
-                expr: config.secondPrompt
-              timeoutMs:
-                expr: liveTurnTimeoutMs(env, 45000)
-        - call: waitForOutboundMessage
-          saveAs: secondOutbound
-          args:
-            - ref: state
-            - lambda:
-                params: [candidate]
-                expr: "candidate.conversation.id === 'qa-room' && candidate.direction === 'outbound' && String(candidate.text ?? '').includes(config.secondMarker)"
-            - expr: liveTurnTimeoutMs(env, 60000)
-            - sinceIndex:
-                ref: beforeRestartCursor
+              expr: "`restart should preserve exactly one marked first reply, saw ${firstMatchesBeforeFollowup.length}; transcript=${formatTransportTranscript(state, { conversationId: 'qa-room' })}`"
+        - sendInbound:
+            conversation:
+              id: qa-room
+              kind: group
+            senderId: qa-driver
+            senderName: QA Driver
+            text:
+              expr: config.secondPrompt
+        - waitForOutbound:
+            textIncludes:
+              ref: config.secondMarker
+            timeoutMs:
+              expr: liveTurnTimeoutMs(env, 60000)
+            sinceIndex:
+              ref: beforeRestartOutboundCount
         - set: snapshot
           value:
             expr: state.getSnapshot()
         - set: firstMatches
           value:
-            expr: "snapshot.messages.slice(0, beforeRestartCursor).filter((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === 'qa-room' && String(candidate.text ?? '').includes(config.firstMarker))"
+            expr: "snapshot.messages.slice(0, beforeRestartMessageIndex).filter((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === 'qa-room' && String(candidate.text ?? '').includes(config.firstMarker))"
         - set: secondMatches
           value:
-            expr: "snapshot.messages.slice(beforeRestartCursor).filter((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === 'qa-room' && String(candidate.text ?? '').includes(config.secondMarker))"
+            expr: "snapshot.messages.slice(beforeRestartMessageIndex).filter((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === 'qa-room' && String(candidate.text ?? '').includes(config.secondMarker))"
         - set: postRestartOutbounds
           value:
-            expr: "snapshot.messages.slice(beforeRestartCursor).filter((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === 'qa-room')"
+            expr: "snapshot.messages.slice(beforeRestartMessageIndex).filter((candidate) => candidate.direction === 'outbound' && candidate.conversation.id === 'qa-room')"
         - assert:
             expr: "firstMatches.length === 1 && secondMatches.length === 1 && postRestartOutbounds.length === 1 && !postRestartOutbounds.some((candidate) => String(candidate.text ?? '').includes(config.firstMarker))"
             message:
               expr: "`expected one marked pre-restart reply and exactly one marked post-restart reply without replaying the first marker; first=${firstMatches.length} second=${secondMatches.length} post=${postRestartOutbounds.length}; transcript=${formatTransportTranscript(state, { conversationId: 'qa-room' })}`"
-      detailsExpr: "`before=${firstOutbound.text}\\nafter=${secondOutbound.text}`"
+      detailsExpr: "`before=${config.firstMarker}\\nafter=${config.secondMarker}`"

qa/scenarios/memory/remember-across-conversations.yaml

@@ -248,7 +248,7 @@ flow:
                   expr: "env.mock ? (await fetchJson(`${env.mock.baseUrl}/debug/requests`)).length : 0"
               - set: targetStartIndex
                 value:
-                  expr: state.getSnapshot().messages.length
+                  expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound').length"
               - sendInbound:
                   conversation:
                     id: