diff --git a/extensions/qa-lab/src/bus-server.test.ts b/extensions/qa-lab/src/bus-server.test.ts index c688d2393078..0a8e83a7d788 100644 --- a/extensions/qa-lab/src/bus-server.test.ts +++ b/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 }); diff --git a/extensions/qa-lab/src/bus-server.ts b/extensions/qa-lab/src/bus-server.ts index be9d9c7f0d3c..70820b0c4ea1 100644 --- a/extensions/qa-lab/src/bus-server.ts +++ b/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": diff --git a/extensions/qa-lab/src/bus-state.ts b/extensions/qa-lab/src/bus-state.ts index 790845d381aa..70f915a83984 100644 --- a/extensions/qa-lab/src/bus-state.ts +++ b/extensions/qa-lab/src/bus-state.ts @@ -75,6 +75,7 @@ export function createQaBusState() { const threads = new Map(); const messages = new Map(); const events: QaBusEvent[] = []; + const acknowledgedPollCursors = new Map(); 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 }); }, diff --git a/extensions/qa-lab/src/scenario-flow-runner.test.ts b/extensions/qa-lab/src/scenario-flow-runner.test.ts index 3822bc1107fd..64e74f9809d0 100644 --- a/extensions/qa-lab/src/scenario-flow-runner.test.ts +++ b/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) => { + await mutate({}); + }, + }, + }, transport, state, scenario, @@ -104,6 +116,15 @@ async function runLoadedScenarioFlow( waitForTransportReady: async () => undefined, waitForQaChannelReady: async () => undefined, waitForNoOutbound: async () => undefined, + waitForCondition: async (check: () => T | Promise) => { + 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(); diff --git a/qa/scenarios/channels/qa-channel-reconnect-dedupe.yaml b/qa/scenarios/channels/qa-channel-reconnect-dedupe.yaml index ab786ebd7a80..119fd9f12788 100644 --- a/qa/scenarios/channels/qa-channel-reconnect-dedupe.yaml +++ b/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}`" diff --git a/qa/scenarios/memory/remember-across-conversations.yaml b/qa/scenarios/memory/remember-across-conversations.yaml index f5252ee7499b..079a6510abdc 100644 --- a/qa/scenarios/memory/remember-across-conversations.yaml +++ b/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: