diff --git a/extensions/qa-lab/src/live-transports/matrix/substrate/config.test.ts b/extensions/qa-lab/src/live-transports/matrix/substrate/config.test.ts index 8b306b06e21d..b8ad40405d18 100644 --- a/extensions/qa-lab/src/live-transports/matrix/substrate/config.test.ts +++ b/extensions/qa-lab/src/live-transports/matrix/substrate/config.test.ts @@ -290,7 +290,10 @@ describe("matrix qa config", () => { }, groupMentionPatterns: ["\\S"], groupPolicy: "open", - streaming: true, + streaming: { + mode: "partial", + progress: { commandText: "raw" }, + }, }, sutAccessToken: "sut-token", sutAccountId: "sut", @@ -307,11 +310,12 @@ describe("matrix qa config", () => { chunkMode: "length", mode: "partial", preview: { toolProgress: true }, + progress: { commandText: "raw" }, }); expect(config.messages?.groupChat?.mentionPatterns).toEqual(["\\S"]); }); - it("resets tool progress when a scalar streaming override follows an opt-out", () => { + it("resets QA streaming detail overrides when a scalar override follows", () => { const optedOut = buildMatrixQaConfig({} as OpenClawConfig, { driverUserId: "@driver:matrix-qa.test", homeserver: "http://127.0.0.1:28008/", @@ -320,6 +324,7 @@ describe("matrix qa config", () => { streaming: { mode: "quiet", preview: { toolProgress: false }, + progress: { commandText: "raw" }, }, }, sutAccessToken: "sut-token", @@ -343,6 +348,7 @@ describe("matrix qa config", () => { chunkMode: "length", mode: "quiet", preview: { toolProgress: false }, + progress: { commandText: "raw" }, }); expect(reset.channels?.matrix?.accounts?.sut?.streaming).toEqual({ block: { enabled: false }, diff --git a/extensions/qa-lab/src/live-transports/matrix/substrate/config.ts b/extensions/qa-lab/src/live-transports/matrix/substrate/config.ts index 597dc9755f24..0bf8e07cd381 100644 --- a/extensions/qa-lab/src/live-transports/matrix/substrate/config.ts +++ b/extensions/qa-lab/src/live-transports/matrix/substrate/config.ts @@ -19,6 +19,9 @@ type MatrixQaStreamingConfig = { preview?: { toolProgress?: boolean; }; + progress?: { + commandText?: "raw" | "status"; + }; }; type MatrixQaAgentDefaultsOverrides = { blockStreamingChunk?: { @@ -128,6 +131,7 @@ type MatrixQaConfigSnapshot = { startupVerification?: "if-unverified" | "off"; streaming: MatrixQaStreamingMode; streamingPreviewToolProgress: boolean; + streamingProgressCommandText?: "raw" | "status"; textChunkLimit?: number; threadBindings: MatrixQaThreadBindingsConfigOverrides; threadReplies: MatrixQaThreadRepliesMode; @@ -445,6 +449,9 @@ function buildMatrixQaChannelAccountConfig(params: { chunkMode: params.snapshot.chunkMode ?? "length", mode: params.snapshot.streaming, preview: { toolProgress: params.snapshot.streamingPreviewToolProgress }, + ...(params.snapshot.streamingProgressCommandText !== undefined + ? { progress: { commandText: params.snapshot.streamingProgressCommandText } } + : {}), }, }; const startupVerificationConfig = @@ -497,6 +504,7 @@ function buildMatrixQaConfigSnapshot(params: { sutUserId: string; topology: MatrixQaProvisionedTopology; }): MatrixQaConfigSnapshot { + const streaming = params.overrides?.streaming; return { allowBots: params.overrides?.allowBots, autoJoin: params.overrides?.autoJoin ?? "off", @@ -516,10 +524,11 @@ function buildMatrixQaConfigSnapshot(params: { }), replyToMode: params.overrides?.replyToMode ?? "off", startupVerification: params.overrides?.startupVerification, - streaming: resolveMatrixQaStreamingMode(params.overrides?.streaming), - streamingPreviewToolProgress: resolveMatrixQaStreamingPreviewToolProgress( - params.overrides?.streaming, - ), + streaming: resolveMatrixQaStreamingMode(streaming), + streamingPreviewToolProgress: resolveMatrixQaStreamingPreviewToolProgress(streaming), + streamingProgressCommandText: isMatrixQaStreamingConfig(streaming) + ? streaming.progress?.commandText + : undefined, threadBindings: { ...params.overrides?.threadBindings }, textChunkLimit: params.overrides?.textChunkLimit, threadReplies: params.overrides?.threadReplies ?? "inbound", diff --git a/extensions/qa-lab/src/live-transports/shared/scenario-selection.test.ts b/extensions/qa-lab/src/live-transports/shared/scenario-selection.test.ts index 249f5e0542d6..a0029800deb7 100644 --- a/extensions/qa-lab/src/live-transports/shared/scenario-selection.test.ts +++ b/extensions/qa-lab/src/live-transports/shared/scenario-selection.test.ts @@ -138,6 +138,8 @@ describe("live transport QA scenario selection", () => { it.each([ { channelId: "matrix", scenarioId: "thread-follow-up" }, + { channelId: "buzz", scenarioId: "channel-canary" }, + { channelId: "telegram", scenarioId: "channel-canary" }, { channelId: "telegram", scenarioId: "channel-message-flows" }, ] as const)( "keeps $scenarioId eligible through both $channelId drivers", @@ -154,4 +156,17 @@ describe("live transport QA scenario selection", () => { expect(selectForDriver("crabline")).toEqual([scenarioId]); }, ); + + it("rejects the shared channel canary on unsupported Discord drivers", () => { + expect(() => + resolveCatalogLiveTransportQaScenarioIds({ + ...MOCK_LANE, + channelId: "discord", + channelDriver: "crabline", + scenarioIds: ["channel-canary"], + }), + ).toThrow( + "selected QA scenario(s) do not match the current QA lane: channel-canary (channel=qa-channel|telegram|buzz)", + ); + }); }); diff --git a/extensions/qa-lab/src/scenario-catalog-channels.test.ts b/extensions/qa-lab/src/scenario-catalog-channels.test.ts index 5136eb531729..f13a10845594 100644 --- a/extensions/qa-lab/src/scenario-catalog-channels.test.ts +++ b/extensions/qa-lab/src/scenario-catalog-channels.test.ts @@ -189,6 +189,35 @@ describe("qa scenario catalog channel contracts", () => { expect(scenario.gatewayConfigPatch).not.toHaveProperty("channels.telegram.groups"); }); + it("keeps the shared channel canary eligible for QA Channel, Telegram, and Buzz", () => { + const scenario = requireFlowScenario(readQaScenarioById("channel-canary")); + + expect(scenario.execution.channels).toEqual(["qa-channel", "telegram", "buzz"]); + }); + + it("keeps raw Matrix command text scoped to the mention-safety scenario", () => { + const scenario = requireFlowScenario( + readQaScenarioById("matrix-room-tool-progress-mention-safety"), + ); + const ordinaryProgressScenario = requireFlowScenario( + readQaScenarioById("matrix-room-tool-progress-preview"), + ); + + expect(scenario.execution.channel).toBe("matrix"); + expect(scenario.execution.config).toMatchObject({ + matrixConfigOverrides: { + streaming: { + mode: "partial", + progress: { commandText: "raw" }, + }, + toolProfile: "coding", + }, + }); + expect(ordinaryProgressScenario.execution.config).not.toHaveProperty( + "matrixConfigOverrides.streaming.progress.commandText", + ); + }); + it("keeps transcript-role delivery on the Crabline driver", () => { const scenario = readQaScenarioById("telegram-assistant-transcript-role-boundary"); const config = readQaScenarioExecutionConfig("telegram-assistant-transcript-role-boundary") as diff --git a/qa/scenarios/channels/channel-canary.yaml b/qa/scenarios/channels/channel-canary.yaml index 1f98ebb3c5b1..417e3f3dac23 100644 --- a/qa/scenarios/channels/channel-canary.yaml +++ b/qa/scenarios/channels/channel-canary.yaml @@ -21,6 +21,10 @@ scenario: - extensions/qa-lab/src/suite.ts execution: kind: flow + channels: + - qa-channel + - telegram + - buzz summary: Run the shared channel canary through QA Channel, Crabline, or a live adapter. transportPolicy: requireGroupMention: true diff --git a/qa/scenarios/channels/matrix-room-tool-progress-mention-safety.yaml b/qa/scenarios/channels/matrix-room-tool-progress-mention-safety.yaml index 219fe5db1d6b..9d6dccfc7baf 100644 --- a/qa/scenarios/channels/matrix-room-tool-progress-mention-safety.yaml +++ b/qa/scenarios/channels/matrix-room-tool-progress-mention-safety.yaml @@ -12,7 +12,10 @@ scenario: retryCount: 0 config: matrixConfigOverrides: - streaming: partial + streaming: + mode: partial + progress: + commandText: raw toolProfile: coding flow: diff --git a/src/infra/delivery-queue-sqlite-claim.test.ts b/src/infra/delivery-queue-sqlite-claim.test.ts new file mode 100644 index 000000000000..a0741c4e6f63 --- /dev/null +++ b/src/infra/delivery-queue-sqlite-claim.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it, vi } from "vitest"; +import { + claimDeliveryQueueEntryPlatformSend, + dispatchDeliveryQueueEntryPlatformSend, +} from "./delivery-queue-sqlite-claim.js"; +import { loadDeliveryQueueEntry, upsertDeliveryQueueEntry } from "./delivery-queue-sqlite.js"; +import { installDeliveryQueueTmpDirHooks } from "./outbound/delivery-queue.test-helpers.js"; + +describe("delivery queue SQLite dispatch ownership", () => { + const { tmpDir } = installDeliveryQueueTmpDirHooks(); + const queueName = "test-dispatch-owner"; + + it("atomically promotes dispatch ownership and rejects expired or replaced claims", () => { + vi.useFakeTimers(); + try { + vi.setSystemTime(new Date("2026-08-10T10:00:00.000Z")); + const stateDir = tmpDir(); + const id = "cron-direct-delivery:v1:dispatch-owner"; + upsertDeliveryQueueEntry({ + queueName, + entry: { + id, + enqueuedAt: Date.now(), + retryCount: 0, + completionRetention: { + idPrefix: "cron-direct-delivery:v1:", + maxAgeMs: 24 * 60 * 60_000, + maxEntries: 2, + }, + requiresProducerClaim: true, + }, + stateDir, + }); + + const expiredClaimId = claimDeliveryQueueEntryPlatformSend({ queueName, id, stateDir }); + if (!expiredClaimId) { + throw new Error("test invariant: the first producer claim must be available"); + } + vi.advanceTimersByTime(30_001); + expect( + dispatchDeliveryQueueEntryPlatformSend({ + queueName, + id, + claimId: expiredClaimId, + stateDir, + }), + ).toBe(false); + + const claimId = claimDeliveryQueueEntryPlatformSend({ queueName, id, stateDir }); + if (!claimId) { + throw new Error("test invariant: the replacement producer claim must be available"); + } + expect( + dispatchDeliveryQueueEntryPlatformSend({ + queueName, + id, + claimId: expiredClaimId, + stateDir, + }), + ).toBe(false); + expect( + dispatchDeliveryQueueEntryPlatformSend({ + queueName, + id, + claimId, + stateDir, + route: { replyToId: "thread-1" }, + }), + ).toBe(true); + expect(loadDeliveryQueueEntry(queueName, id, stateDir)).toMatchObject({ + recoveryState: "send_attempt_started", + platformSendAttemptId: claimId, + platformSendStartedAt: Date.now(), + effectiveReplyToId: "thread-1", + availableAt: Date.now() + 30_000, + }); + expect(loadDeliveryQueueEntry(queueName, id, stateDir)?.producerClaimId).toBeUndefined(); + + vi.advanceTimersByTime(30_001); + expect(dispatchDeliveryQueueEntryPlatformSend({ queueName, id, claimId, stateDir })).toBe( + false, + ); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/src/infra/delivery-queue-sqlite-claim.ts b/src/infra/delivery-queue-sqlite-claim.ts index c1676a17ba0e..3a5b565322cc 100644 --- a/src/infra/delivery-queue-sqlite-claim.ts +++ b/src/infra/delivery-queue-sqlite-claim.ts @@ -56,9 +56,9 @@ export function transitionOwnedDeliveryQueueEntry( ); } -function transitionUnsentDeliveryQueueEntry( +function transitionDeliveryQueueEntryPlatformSend( params: PlatformClaimParams, - operation: "claim" | "promote", + operation: "claim" | "promote" | "dispatch", transition: (entry: DeliveryQueueEntryState, now: number) => DeliveryQueueEntryState | undefined, ): boolean { // State-database opens reuse the canonical path-owned connection, so both @@ -70,13 +70,16 @@ function transitionUnsentDeliveryQueueEntry( database.db, () => { const current = loadDeliveryQueueEntry(params.queueName, params.id, params.stateDir); + if (!current) { + return false; + } if ( - !current || - (current.platformSendStartedAt !== undefined && - (operation !== "claim" || - current.platformSendStartedAt !== params.reconciledPlatformSendStartedAt || - current.platformSendAttemptId !== params.reconciledPlatformSendAttemptId || - typeof current.platformSendAttemptId !== "string")) + current.platformSendStartedAt !== undefined && + (operation === "promote" || + (operation === "claim" && + (current.platformSendStartedAt !== params.reconciledPlatformSendStartedAt || + current.platformSendAttemptId !== params.reconciledPlatformSendAttemptId || + typeof current.platformSendAttemptId !== "string"))) ) { return false; } @@ -102,7 +105,7 @@ export function claimDeliveryQueueEntryPlatformSend( params: PlatformClaimParams, ): string | undefined { const claimId = generateSecureUuid(); - return transitionUnsentDeliveryQueueEntry(params, "claim", (entry, now) => { + return transitionDeliveryQueueEntryPlatformSend(params, "claim", (entry, now) => { const reconciledNotSent = entry.recoveryState === "send_attempt_started" && typeof params.reconciledPlatformSendStartedAt === "number" && @@ -185,7 +188,7 @@ export function promoteDeliveryQueueEntryPlatformSend( route?: { replyToId?: string | null }; }, ): boolean { - return transitionUnsentDeliveryQueueEntry(params, "promote", (entry, now) => + return transitionDeliveryQueueEntryPlatformSend(params, "promote", (entry, now) => entry.recoveryState === "producer_claimed" && entry.producerClaimId === params.claimId && typeof entry.availableAt === "number" && @@ -207,3 +210,49 @@ export function promoteDeliveryQueueEntryPlatformSend( : undefined, ); } + +/** Atomically authorize dispatch, promoting a producer claim into the active attempt. */ +export function dispatchDeliveryQueueEntryPlatformSend( + params: PlatformClaimParams & { + claimId: string; + route?: { replyToId?: string | null }; + }, +): boolean { + return transitionDeliveryQueueEntryPlatformSend(params, "dispatch", (entry, now) => { + const producerOwned = + entry.recoveryState === "producer_claimed" && + entry.producerClaimId === params.claimId && + typeof entry.availableAt === "number" && + entry.availableAt > now; + const attemptOwned = + (entry.recoveryState === "send_attempt_started" || + entry.recoveryState === "unknown_after_send") && + entry.platformSendAttemptId === params.claimId && + (entry.requiresProducerClaim !== true || + (typeof entry.availableAt === "number" && entry.availableAt > now)); + if (!producerOwned && !attemptOwned) { + return undefined; + } + return { + ...entry, + // Exact reconciliation can skip pre-send promotion, so publish attempt identity + // atomically; later batch dispatches retain stronger unknown-after-send evidence. + availableAt: + entry.requiresProducerClaim === true + ? producerOwned + ? now + PLATFORM_SEND_OWNER_LEASE_MS + : entry.availableAt + : undefined, + producerClaimId: undefined, + platformSendAttemptId: params.claimId, + platformSendStartedAt: now, + ...(params.route && "replyToId" in params.route + ? { effectiveReplyToId: params.route.replyToId ?? null } + : {}), + recoveryState: + entry.recoveryState === "unknown_after_send" + ? "unknown_after_send" + : "send_attempt_started", + }; + }); +} diff --git a/src/infra/outbound/deliver-queue.exact-reconciliation.integration.test.ts b/src/infra/outbound/deliver-queue.exact-reconciliation.integration.test.ts new file mode 100644 index 000000000000..60314adcead8 --- /dev/null +++ b/src/infra/outbound/deliver-queue.exact-reconciliation.integration.test.ts @@ -0,0 +1,101 @@ +import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createMessageReceiptFromOutboundResults } from "../../channels/message/receipt.js"; +import type { ChannelMessageSendTextContext } from "../../channels/message/types.js"; +import type { OpenClawConfig } from "../../config/config.js"; +import { createEmptyPluginRegistry } from "../../plugins/registry.js"; +import { resetPluginRuntimeStateForTest, setActivePluginRegistry } from "../../plugins/runtime.js"; +import { createOutboundTestPlugin, createTestRegistry } from "../../test-utils/channel-plugins.js"; +import { getDeliveryQueueEntryStatus } from "../delivery-queue-sqlite.js"; +import { + boundedCronCompletionRetention, + drainMatrixReconnect, + matrixOutboundForQueueTest, +} from "./deliver.queue-integration.test-support.js"; +import { OUTBOUND_DELIVERY_QUEUE_NAME } from "./delivery-queue-media-staging.js"; +import type { DeliverFn } from "./delivery-queue.js"; +import { installDeliveryQueueTmpDirHooks } from "./delivery-queue.test-helpers.js"; + +let deliverOutboundPayloads: typeof import("./deliver.js").deliverOutboundPayloads; + +describe("exact Matrix delivery queue reconciliation", () => { + const fixtures = installDeliveryQueueTmpDirHooks(); + let tmpDir: string; + + beforeAll(async () => { + ({ deliverOutboundPayloads } = await import("./deliver.js")); + }); + + beforeEach(() => { + tmpDir = fixtures.tmpDir(); + }); + + afterEach(() => { + resetPluginRuntimeStateForTest(); + setActivePluginRegistry(createEmptyPluginRegistry()); + }); + + it.each(["required", "best_effort"] as const)( + "settles one exact Matrix %s send without restart replay", + async (queuePolicy) => { + process.env.OPENCLAW_STATE_DIR = tmpDir; + const deliveryIntentId = `cron-direct-delivery:v1:exact-${queuePolicy}-completion`; + const messageId = `exact-${queuePolicy}-message`; + const reconcileUnknownSend = vi.fn(); + const sendText = vi.fn(async (ctx: ChannelMessageSendTextContext) => { + expect(ctx.deliveryQueueId).toBe(deliveryIntentId); + await ctx.onPlatformSendDispatch?.(); + return { + messageId, + receipt: createMessageReceiptFromOutboundResults({ + results: [{ channel: "matrix", messageId }], + kind: "text", + }), + }; + }); + setActivePluginRegistry( + createTestRegistry([ + { + pluginId: "matrix", + source: "test", + plugin: { + ...createOutboundTestPlugin({ id: "matrix", outbound: matrixOutboundForQueueTest }), + message: { + id: "matrix", + durableFinal: { + capabilities: { text: true, reconcileUnknownSend: true }, + reconcileUnknownSendKinds: { text: true }, + reconcileUnknownSend, + }, + send: { text: sendText }, + }, + }, + }, + ]), + ); + const params = { + cfg: {} as OpenClawConfig, + channel: "matrix" as const, + to: "!room:example", + payloads: [{ text: "send exactly once with durable platform identity" }], + queuePolicy, + ...(queuePolicy === "best_effort" ? { bestEffort: true } : {}), + deliveryIntentId, + completionRetention: boundedCronCompletionRetention, + reusePendingDeliveryIntent: true, + requireUnknownSendReconciliation: true, + }; + + await expect(deliverOutboundPayloads(params)).resolves.toMatchObject([{ messageId }]); + expect(sendText).toHaveBeenCalledOnce(); + expect(reconcileUnknownSend).not.toHaveBeenCalled(); + expect( + getDeliveryQueueEntryStatus(OUTBOUND_DELIVERY_QUEUE_NAME, deliveryIntentId, tmpDir), + ).toBe("completed"); + + const recoveryDeliver = vi.fn(async () => []); + await drainMatrixReconnect({ deliver: recoveryDeliver, stateDir: tmpDir }); + expect(recoveryDeliver).not.toHaveBeenCalled(); + expect(sendText).toHaveBeenCalledOnce(); + }, + ); +}); diff --git a/src/infra/outbound/delivery-queue-platform-lease.ts b/src/infra/outbound/delivery-queue-platform-lease.ts index 380629357be5..9cde7dfe8978 100644 --- a/src/infra/outbound/delivery-queue-platform-lease.ts +++ b/src/infra/outbound/delivery-queue-platform-lease.ts @@ -1,9 +1,26 @@ import { claimDeliveryQueueEntryPlatformSend, + dispatchDeliveryQueueEntryPlatformSend, renewDeliveryQueueEntryPlatformSendLease, } from "../delivery-queue-sqlite-claim.js"; import { OUTBOUND_DELIVERY_QUEUE_NAME } from "./delivery-queue-media-staging.js"; +/** Atomically transfer a stable pending producer intent to one platform sender. */ +export async function claimDeliveryPlatformSendAttempt( + id: string, + stateDir?: string, + reconciledPlatformSendStartedAt?: number, + reconciledPlatformSendAttemptId?: string, +): Promise { + return claimDeliveryQueueEntryPlatformSend({ + queueName: OUTBOUND_DELIVERY_QUEUE_NAME, + id, + stateDir, + ...(reconciledPlatformSendStartedAt !== undefined ? { reconciledPlatformSendStartedAt } : {}), + ...(reconciledPlatformSendAttemptId !== undefined ? { reconciledPlatformSendAttemptId } : {}), + }); +} + /** Claim and atomically upgrade a live reusable producer to renewable ownership. */ export async function claimReusableDeliveryPlatformSendAttempt( id: string, @@ -30,3 +47,22 @@ export async function renewDeliveryPlatformSendLease( claimId, }); } + +/** Promote or refresh the exact live owner at recipient-visible dispatch. */ +export function markOwnedDeliveryPlatformSendDispatched( + id: string, + stateDir: string | undefined, + route: { replyToId?: string | null } | undefined, + claimId: string, +): void { + const dispatched = dispatchDeliveryQueueEntryPlatformSend({ + queueName: OUTBOUND_DELIVERY_QUEUE_NAME, + id, + stateDir, + route, + claimId, + }); + if (!dispatched) { + throw new Error(`Stable delivery platform claim was lost: ${id}`); + } +} diff --git a/src/infra/outbound/delivery-queue-storage.ts b/src/infra/outbound/delivery-queue-storage.ts index 5599d58011b7..b964c23b8df8 100644 --- a/src/infra/outbound/delivery-queue-storage.ts +++ b/src/infra/outbound/delivery-queue-storage.ts @@ -9,7 +9,6 @@ import type { import type { ReplyToMode } from "../../config/types.js"; import type { PluginHookReplyPayloadSendingContext } from "../../plugins/hook-types.js"; import { - claimDeliveryQueueEntryPlatformSend, promoteDeliveryQueueEntryPlatformSend, transitionOwnedDeliveryQueueEntry, } from "../delivery-queue-sqlite-claim.js"; @@ -43,6 +42,7 @@ import { OUTBOUND_DELIVERY_QUEUE_NAME, OUTBOUND_LEGACY_PREPARATION_QUEUE_NAME, } from "./delivery-queue-media-staging.js"; +import { markOwnedDeliveryPlatformSendDispatched } from "./delivery-queue-platform-lease.js"; import { StableDeliveryPreparationLostError, type StableDeliveryPreparation, @@ -474,21 +474,7 @@ export async function failDeliveryAfterPlatformSend( ); } -/** Atomically transfer a stable pending producer intent to one platform sender. */ -export async function claimDeliveryPlatformSendAttempt( - id: string, - stateDir?: string, - reconciledPlatformSendStartedAt?: number, - reconciledPlatformSendAttemptId?: string, -): Promise { - return claimDeliveryQueueEntryPlatformSend({ - queueName: OUTBOUND_DELIVERY_QUEUE_NAME, - id, - stateDir, - ...(reconciledPlatformSendStartedAt !== undefined ? { reconciledPlatformSendStartedAt } : {}), - ...(reconciledPlatformSendAttemptId !== undefined ? { reconciledPlatformSendAttemptId } : {}), - }); -} +export { claimDeliveryPlatformSendAttempt } from "./delivery-queue-platform-lease.js"; /** Reserve one durable delivery call before invoking the provider path. */ export async function reserveDeliveryAttempt( @@ -572,18 +558,23 @@ export async function markDeliveryPlatformSendDispatched( route?: { replyToId?: string | null }, expectedPlatformSendAttemptId?: string | null, ): Promise { + if (typeof expectedPlatformSendAttemptId === "string") { + markOwnedDeliveryPlatformSendDispatched(id, stateDir, route, expectedPlatformSendAttemptId); + return; + } updateQueuedDelivery( id, stateDir, (entry) => ({ ...entry, - // Dispatch still belongs to the promoted producer until provider I/O - // settles; clearing its lease lets another process replay an active send. - availableAt: expectedPlatformSendAttemptId ? entry.availableAt : undefined, + availableAt: undefined, producerClaimId: undefined, platformSendStartedAt: Date.now(), ...(route && "replyToId" in route ? { effectiveReplyToId: route.replyToId ?? null } : {}), - recoveryState: "send_attempt_started", + // A later batch send must not erase concrete evidence from an earlier result; + // recovery could otherwise replay the whole batch and duplicate that delivery. + recoveryState: + entry.recoveryState === "unknown_after_send" ? entry.recoveryState : "send_attempt_started", }), expectedPlatformSendAttemptId, ); diff --git a/src/infra/outbound/delivery-queue.storage.test.ts b/src/infra/outbound/delivery-queue.storage.test.ts index 6c830d88bbfd..95f938f50285 100644 --- a/src/infra/outbound/delivery-queue.storage.test.ts +++ b/src/infra/outbound/delivery-queue.storage.test.ts @@ -547,6 +547,25 @@ describe("delivery-queue storage", () => { expect(entry.recoveryState).toBe("send_attempt_started"); }); + it("keeps ambiguous post-send evidence across a later unclaimed batch dispatch", async () => { + const id = await enqueueTextDelivery( + { + channel: "forum", + to: "123", + payloads: [{ text: "test" }], + }, + tmpDir(), + ); + + await markDeliveryPlatformSendAttemptStarted(id, tmpDir()); + await markDeliveryPlatformOutcomeUnknown(id, tmpDir()); + await markDeliveryPlatformSendDispatched(id, tmpDir()); + + // Downgrading to send_attempt_started would let recovery replay the whole + // batch as not_sent and duplicate the payload that already reached the platform. + expect(readQueuedEntry(tmpDir(), id).recoveryState).toBe("unknown_after_send"); + }); + it("increments retryCount, records attempt time, and sets lastError", async () => { const id = await enqueueTextDelivery( {