From 5ab422947616a1e6b48b38e7766ee0d194d9e84a Mon Sep 17 00:00:00 2001 From: Haakam Aujla Date: Thu, 23 Jul 2026 09:21:45 +0000 Subject: [PATCH 1/3] Harden durable ingress edge cases --- README.md | 4 + src/channel/src/catch-up.test.ts | 38 +++++++++- src/channel/src/catch-up.ts | 71 +++++++++--------- src/channel/src/durable-receive.test.ts | 98 +++++++++++++++++++++++-- src/channel/src/durable-receive.ts | 56 ++++++++++---- src/channel/src/inbound.test.ts | 18 +++-- src/channel/src/inbound.ts | 2 +- src/channel/src/ingress.ts | 33 ++++++--- src/channel/src/media.ts | 2 + 9 files changed, 247 insertions(+), 75 deletions(-) diff --git a/README.md b/README.md index 4b542c7..28c8274 100644 --- a/README.md +++ b/README.md @@ -104,6 +104,10 @@ Security defaults worth knowing: - `dmPolicy: "open"` requires `allowFrom` to include `"*"`. - Every reply re-hydrates the triggering message and re-authorizes its `From`, so an untrusted `Reply-To` cannot redirect delivery. +The configured `inboxId` also identifies the durable receive queue. Keep its casing stable across +upgrades: changing only letter case can create a new queue identity, so messages covered only by +older completion tombstones may be dispatched once more during the migration. + ## Tools | Tool | Purpose | diff --git a/src/channel/src/catch-up.test.ts b/src/channel/src/catch-up.test.ts index 73ef9a3..c6d4a18 100644 --- a/src/channel/src/catch-up.test.ts +++ b/src/channel/src/catch-up.test.ts @@ -196,7 +196,7 @@ describe("AgentMail durable REST catch-up", () => { ); }); - it("does not advance the cursor beyond the sweep's fixed upper bound", async () => { + it("does not advance the cursor from an invalid timestamp's local fallback", async () => { const store = memoryStore(); const malformed = message({ id: "bad_timestamp", timestamp: 1_100 }) as AgentMail.MessageItem; (malformed as { timestamp: unknown }).timestamp = new Date(Number.NaN); @@ -223,7 +223,41 @@ describe("AgentMail durable REST catch-up", () => { expect(list).toHaveBeenLastCalledWith( "inbox_1", expect.objectContaining({ - after: new Date(2_000_000 - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS), + after: new Date(1_000_000 - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS), + }), + expect.any(Object), + ); + }); + + it("does not regress a stored high-water mark when the clock moves backward", async () => { + const store = memoryStore(); + const list = vi + .fn() + .mockResolvedValueOnce({ + count: 1, + messages: [message({ id: "message_1", timestamp: 1_900_000 })], + }) + .mockResolvedValue({ count: 0, messages: [] }); + let nowMs = 1_000_000; + const session = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list } } } as never, + store: store as never, + now: () => nowMs, + }); + const receive = vi.fn(async () => undefined); + + nowMs = 2_000_000; + await session.run({ receive, abortSignal: new AbortController().signal }); + nowMs = 1_500_000; + await session.run({ receive, abortSignal: new AbortController().signal }); + nowMs = 2_000_000; + await session.run({ receive, abortSignal: new AbortController().signal }); + + expect(list).toHaveBeenLastCalledWith( + "inbox_1", + expect.objectContaining({ + after: new Date(1_900_000 - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS), }), expect.any(Object), ); diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts index 4a78fc1..f0b3946 100644 --- a/src/channel/src/catch-up.ts +++ b/src/channel/src/catch-up.ts @@ -187,38 +187,28 @@ async function persistCursor(params: { upperBoundAtMs: number; established: boolean; }): Promise { + const merge = (currentValue: unknown): AgentMailCatchUpCursor => { + const current = normalizeCursor(currentValue); + const baselineAtMs = current?.baselineAtMs ?? params.baselineAtMs; + return { + version: CURSOR_VERSION, + baselineAtMs, + // A backward-moving clock may lower this run's upper bound, but must never lower a + // high-water mark already committed by an earlier run. Clamp only the new contribution. + highWaterAtMs: Math.max( + baselineAtMs, + current?.highWaterAtMs ?? 0, + Math.min(params.upperBoundAtMs, params.highWaterAtMs), + ), + established: current?.established === true || params.established, + }; + }; const update = params.store.update; if (update) { - await update(params.key, (currentValue) => { - const current = normalizeCursor(currentValue); - return { - version: CURSOR_VERSION, - baselineAtMs: current?.baselineAtMs ?? params.baselineAtMs, - highWaterAtMs: Math.max( - current?.baselineAtMs ?? params.baselineAtMs, - Math.min( - params.upperBoundAtMs, - Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs), - ), - ), - established: current?.established === true || params.established, - }; - }); + await update(params.key, merge); return; } - const current = normalizeCursor(await params.store.lookup(params.key)); - await params.store.register(params.key, { - version: CURSOR_VERSION, - baselineAtMs: current?.baselineAtMs ?? params.baselineAtMs, - highWaterAtMs: Math.max( - current?.baselineAtMs ?? params.baselineAtMs, - Math.min( - params.upperBoundAtMs, - Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs), - ), - ), - established: current?.established === true || params.established, - }); + await params.store.register(params.key, merge(await params.store.lookup(params.key))); } export async function createAgentMailCatchUpSession(params: { @@ -304,16 +294,18 @@ export async function createAgentMailCatchUpSession(params: { } // Invalid provider time must not turn a missed live event into a permanent drop. Admit // with local observation time; durable message-id dedupe keeps overlap scans safe. - const messageTimestampMs = - resolveReceivedAgentMailMessageTimestampMs(message, params.account.inboxId) ?? - scanUpperBoundAtMs; + const providerTimestampMs = resolveReceivedAgentMailMessageTimestampMs( + message, + params.account.inboxId, + ); + const admissionTimestampMs = providerTimestampMs ?? scanUpperBoundAtMs; try { await receive({ accountId: params.account.accountId, inboxId: params.account.inboxId, messageId: message.messageId, transport: "rest", - receivedAt: messageTimestampMs, + receivedAt: admissionTimestampMs, arrivedAt: now(), }); } catch (error) { @@ -329,11 +321,16 @@ export async function createAgentMailCatchUpSession(params: { throw error; } admitted += 1; - highWaterAtMs = Math.max( - highWaterAtMs, - Math.min(messageTimestampMs, scanUpperBoundAtMs), - ); - pageAdvanced = true; + // Local observation time makes an invalid provider timestamp admissible, but it is not + // evidence that every older provider message has been indexed. Only real provider time + // may advance the REST scan cursor. + if (providerTimestampMs !== null) { + highWaterAtMs = Math.max( + highWaterAtMs, + Math.min(providerTimestampMs, scanUpperBoundAtMs), + ); + pageAdvanced = true; + } } if (pageAdvanced) { // Persist once per page. If admission fails mid-page, the cursor stays behind the page diff --git a/src/channel/src/durable-receive.test.ts b/src/channel/src/durable-receive.test.ts index 2452ff7..db21b70 100644 --- a/src/channel/src/durable-receive.test.ts +++ b/src/channel/src/durable-receive.test.ts @@ -61,6 +61,52 @@ describe("AgentMail durable ingress", () => { expect(deletePending).toHaveBeenCalledWith(id); }); + it("coordinates capacity across journal facades for the same queue", async () => { + const pendingRows = new Map(); + const baseJournal = () => + ({ + accept: vi.fn(async (id: string, payload: AgentMailIngressRecord) => { + const existing = pendingRows.get(id); + if (existing) { + return { + kind: "pending", + duplicate: true, + record: { ...existing, attempts: 0 }, + }; + } + const accepted = { id, payload }; + pendingRows.set(id, accepted); + return { kind: "accepted", duplicate: false, record: accepted }; + }), + pending: vi.fn(async () => + [...pendingRows.values()].map((item) => ({ ...item, attempts: 0 })), + ), + complete: vi.fn(), + release: vi.fn(), + deletePending: vi.fn(async (id: string) => pendingRows.delete(id)), + }) as never; + const coordinationKey = `capacity-test-${crypto.randomUUID()}`; + const first = withAgentMailIngressCapacity(baseJournal(), 3, coordinationKey); + const second = withAgentMailIngressCapacity(baseJournal(), 3, coordinationKey); + + await first.accept("message_1", record); + await second.accept("message_2", record); + const admissions = await Promise.allSettled([ + first.accept("message_3", record), + second.accept("message_4", record), + ]); + + expect(admissions.filter((result) => result.status === "fulfilled")).toHaveLength(1); + expect( + admissions.filter( + (result) => + result.status === "rejected" && + result.reason instanceof AgentMailIngressCapacityError, + ), + ).toHaveLength(1); + expect(pendingRows).toHaveLength(3); + }); + it("keeps an overflow row retryable when rollback deletion fails", async () => { const id = createAgentMailDurableInboundId(record); const fail = vi.fn(async () => true); @@ -98,7 +144,12 @@ describe("AgentMail durable ingress", () => { await expect( processAgentMailIngress({ journal: journal as never, record, dispatch }), ).resolves.toBe("accepted"); - await vi.waitFor(() => expect(complete).toHaveBeenCalledWith(id)); + await vi.waitFor(() => + expect(complete).toHaveBeenCalledWith( + id, + expect.objectContaining({ completedAt: expect.any(Number) }), + ), + ); expect(dispatch).toHaveBeenCalledOnce(); expect(fail).not.toHaveBeenCalled(); }); @@ -204,6 +255,26 @@ describe("AgentMail durable ingress", () => { expect(order).toEqual(["accept", "dispatch", "complete"]); }); + it("retains future-dated completion tombstones through the provider scan horizon", async () => { + const futureReceivedAt = Date.now() + 24 * 60 * 60 * 1000; + const complete = vi.fn(async () => undefined); + await processAgentMailIngress({ + journal: { + accept: async () => ({ kind: "accepted", duplicate: false, record: {} }), + complete, + release: vi.fn(), + } as never, + record: { ...record, receivedAt: futureReceivedAt }, + dispatch: vi.fn(async () => undefined), + }); + + await vi.waitFor(() => + expect(complete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record), { + completedAt: futureReceivedAt, + }), + ); + }); + it("returns after durable admission without waiting for agent dispatch", async () => { let finishDispatch!: () => void; const dispatch = vi.fn( @@ -768,7 +839,13 @@ describe("AgentMail durable ingress", () => { record, expect.objectContaining({ onTurnAdopted: expect.any(Function) }), ); - expect(complete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record)); + expect(complete).toHaveBeenCalledWith( + createAgentMailDurableInboundId(record), + expect.objectContaining({ completedAt: expect.any(Number) }), + ); + // Let the completed dispatch remove itself from the process-wide active-dispatch registry + // before exercising a fresh pending row with the same durable id. + await new Promise((resolve) => setTimeout(resolve, 0)); dispatch.mockClear(); complete.mockClear(); @@ -786,7 +863,10 @@ describe("AgentMail durable ingress", () => { dispatch, }); await vi.waitFor(() => expect(dispatch).toHaveBeenCalledOnce()); - expect(complete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record)); + expect(complete).toHaveBeenCalledWith( + createAgentMailDurableInboundId(record), + expect.objectContaining({ completedAt: expect.any(Number) }), + ); }); it("keeps WhatsApp-aligned retention values", () => { @@ -812,7 +892,12 @@ describe("AgentMail durable ingress", () => { } as never, dispatch, }); - await vi.waitFor(() => expect(complete).toHaveBeenCalledWith("durable_1")); + await vi.waitFor(() => + expect(complete).toHaveBeenCalledWith( + "durable_1", + expect.objectContaining({ completedAt: expect.any(Number) }), + ), + ); expect(dispatch).toHaveBeenCalledWith( record, expect.objectContaining({ onTurnAdopted: expect.any(Function) }), @@ -937,7 +1022,10 @@ describe("AgentMail durable ingress", () => { firstAbort.abort(); rejectFirst(new Error("old account stopped")); await vi.waitFor(() => expect(replacementDispatch).toHaveBeenCalledOnce()); - expect(replacementComplete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record)); + expect(replacementComplete).toHaveBeenCalledWith( + createAgentMailDurableInboundId(record), + expect.objectContaining({ completedAt: expect.any(Number) }), + ); replacementAbort.abort(); }); }); diff --git a/src/channel/src/durable-receive.ts b/src/channel/src/durable-receive.ts index b5539c9..d149a75 100644 --- a/src/channel/src/durable-receive.ts +++ b/src/channel/src/durable-receive.ts @@ -38,6 +38,16 @@ type AgentMailJournal = ReturnType< ) => Promise; }; +type AgentMailCapacityAdmissionState = { + admissionChain: Promise; + pendingEstimate: number | null; +}; + +// Account reloads can briefly leave old and replacement journal facades open over the same +// persisted queue. Keep their count-and-accept sequence on one chain, keyed by that queue's stable +// storage identity, rather than coordinating only calls made through one facade. +const capacityAdmissionStates = new Map(); + /** * Wraps the store-backed journal with an admission cap. The published SDK's queue journal only * evicts on capacity; durable email must instead reject NEW mail while keeping already-accepted @@ -46,33 +56,48 @@ type AgentMailJournal = ReturnType< export function withAgentMailIngressCapacity( journal: AgentMailJournal, maxPendingEntries: number, + coordinationKey?: string, ): AgentMailJournal { // Serialize check-and-enqueue. The published queue journal has no atomic admission cap, so two // concurrent transports (live WebSocket + REST catch-up) could otherwise both observe free space - // and push past the bound. Chaining admissions makes the count-then-accept step atomic per - // journal; the chain never rejects so one failed admission cannot poison later ones. - let admissionChain: Promise = Promise.resolve(); + // and push past the bound. Chaining admissions makes the count-then-accept step atomic across + // every facade for the queue; the chain never rejects so one failed admission cannot poison + // later ones. + const state = coordinationKey + ? (capacityAdmissionStates.get(coordinationKey) ?? { + admissionChain: Promise.resolve(), + pendingEstimate: null, + }) + : { + admissionChain: Promise.resolve(), + pendingEstimate: null, + }; + if (coordinationKey && !capacityAdmissionStates.has(coordinationKey)) { + capacityAdmissionStates.set(coordinationKey, state); + } // Upper-bound estimate of the pending count. It only increases on new admissions (never on // completion/retention pruning), so the O(pending) scan is skipped on the common below-cap path // and only runs to re-sync from the source of truth when the estimate first reaches the cap. - let pendingEstimate: number | null = null; return { ...journal, accept: (id, payload, options) => { - const admission = admissionChain.then(async () => { + const admission = state.admissionChain.then(async () => { const accepted = await journal.accept(id, payload, options); // Let the queue perform its atomic id lookup first. Completed tombstones and pending // duplicates consume no new capacity and must remain harmless even while the queue is full. if (accepted.kind !== "accepted") { return accepted; } - if (pendingEstimate === null || pendingEstimate >= maxPendingEntries) { + if ( + state.pendingEstimate === null || + state.pendingEstimate >= maxPendingEntries + ) { const pending = await journal.pending(); - pendingEstimate = pending.length; - if (pendingEstimate > maxPendingEntries) { + state.pendingEstimate = pending.length; + if (state.pendingEstimate > maxPendingEntries) { try { if (await journal.deletePending(id)) { - pendingEstimate -= 1; + state.pendingEstimate -= 1; } } catch { // Never turn a capacity rejection into a terminal tombstone. If rollback deletion is @@ -82,11 +107,11 @@ export function withAgentMailIngressCapacity( throw new AgentMailIngressCapacityError(); } } else { - pendingEstimate += 1; + state.pendingEstimate += 1; } return accepted; }); - admissionChain = admission.then( + state.admissionChain = admission.then( () => undefined, () => undefined, ); @@ -100,10 +125,12 @@ export function createAgentMailDurableInboundReceiveJournal(params: { inboxId: string; }): AgentMailJournal { const runtime = getAgentMailRuntime(); + const stateDir = runtime.state.resolveStateDir(); + const queueAccountId = sha256Hex(`${params.accountId}\n${params.inboxId}`).slice(0, 24); const queue = runtime.state.openChannelIngressQueue( { - accountId: sha256Hex(`${params.accountId}\n${params.inboxId}`).slice(0, 24), - stateDir: runtime.state.resolveStateDir(), + accountId: queueAccountId, + stateDir, }, ); const retention = { @@ -125,7 +152,7 @@ export function createAgentMailDurableInboundReceiveJournal(params: { const extendedJournal: AgentMailJournal = { accept: async (id, payload, options) => { await prune(); - const result = await queue.enqueue(id.trim(), payload, options); + const result = await queue.enqueue(id, payload, options); await prune(id); if (result.kind === "accepted") { return { kind: "accepted", duplicate: false, record: result.record }; @@ -171,5 +198,6 @@ export function createAgentMailDurableInboundReceiveJournal(params: { return withAgentMailIngressCapacity( extendedJournal, AGENTMAIL_DURABLE_PENDING_MAX_ENTRIES, + `${stateDir}\nagentmail\n${queueAccountId}`, ); } diff --git a/src/channel/src/inbound.test.ts b/src/channel/src/inbound.test.ts index 690a038..92d4301 100644 --- a/src/channel/src/inbound.test.ts +++ b/src/channel/src/inbound.test.ts @@ -337,13 +337,16 @@ describe("AgentMail REST-authoritative inbound", () => { expect(onTurnAbandoned).toHaveBeenCalledOnce(); }); - it("cleans up deferred-turn attachments when account shutdown aborts the queued turn", async () => { + it("does not delete core-owned deferred attachments when account shutdown aborts", async () => { rm.mockClear(); loadAgentMailInboundAttachments.mockResolvedValueOnce({ paths: ["/tmp/deferred-abort.bin"], types: ["application/octet-stream"], }); const controller = new AbortController(); + let lifecycle: + | { onDeferred: () => void; onAbandoned: () => Promise } + | undefined; await dispatchAgentMailInboundEvent({ cfg: {}, account, @@ -355,9 +358,10 @@ describe("AgentMail REST-authoritative inbound", () => { run: async ({ turnAdoptionLifecycle, }: { - turnAdoptionLifecycle: { onDeferred: () => void }; + turnAdoptionLifecycle: typeof lifecycle; }) => { - turnAdoptionLifecycle.onDeferred(); + lifecycle = turnAdoptionLifecycle; + turnAdoptionLifecycle?.onDeferred(); }, }, session: { resolveStorePath: () => "/tmp/s.json", recordInboundSession: vi.fn() }, @@ -369,9 +373,11 @@ describe("AgentMail REST-authoritative inbound", () => { expect(rm).not.toHaveBeenCalled(); controller.abort(); - await vi.waitFor(() => - expect(rm).toHaveBeenCalledWith("/tmp/deferred-abort.bin", { force: true }), - ); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(rm).not.toHaveBeenCalled(); + + await lifecycle?.onAbandoned(); + expect(rm).toHaveBeenCalledWith("/tmp/deferred-abort.bin", { force: true }); }); it("cleans up attachments when inbound handling resolves without adoption", async () => { diff --git a/src/channel/src/inbound.ts b/src/channel/src/inbound.ts index 839432e..8fc46c6 100644 --- a/src/channel/src/inbound.ts +++ b/src/channel/src/inbound.ts @@ -270,7 +270,7 @@ export async function dispatchAgentMailInboundEvent(params: { let detachAbortCleanup = () => {}; if (params.abortSignal) { const onAbort = () => { - if (!turnAdoptionObserved) { + if (!turnAdoptionObserved && !turnDeferred) { void cleanupInboundMedia(); } }; diff --git a/src/channel/src/ingress.ts b/src/channel/src/ingress.ts index 178ddcc..e2bfb93 100644 --- a/src/channel/src/ingress.ts +++ b/src/channel/src/ingress.ts @@ -8,12 +8,7 @@ import type { AgentMailIngressRecord } from "./types.js"; // Re-exported for transports (websocket) that catch capacity backpressure by class. export { AgentMailIngressCapacityError }; -type AgentMailJournal = ReturnType & { - fail?: ( - id: string, - options: { reason: string; message?: string; failedAt?: number }, - ) => Promise; -}; +type AgentMailJournal = ReturnType; type DispatchParams = { journal: AgentMailJournal; @@ -78,6 +73,24 @@ const activeDispatches = new Map(); const retryDelayMs = createBackoff(30 * 60_000); +async function completeAgentMailIngress(params: { + journal: AgentMailJournal; + id: string; + record: AgentMailIngressRecord; +}): Promise { + // REST catch-up scans by provider timestamp. A live message stamped in the future must retain its + // tombstone until seven days after that provider time, or it can re-enter the scan window only + // after a locally-timestamped completion marker has already expired. + const completedAt = Date.now(); + const providerReceivedAt = + Number.isFinite(params.record.receivedAt) && params.record.receivedAt >= 0 + ? params.record.receivedAt + : completedAt; + await params.journal.complete(params.id, { + completedAt: Math.max(completedAt, providerReceivedAt), + }); +} + // True when a dispatch failure is a provider-projection race: either a 404 (message not yet // REST-visible) or a not-yet-projected `received` label. Both resolve on their own within seconds. function isHydrationRetryable(error: unknown): boolean { @@ -203,7 +216,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro adoptionTask = (async () => { while (!params.abortSignal?.aborted) { try { - await params.journal.complete(params.id); + await completeAgentMailIngress(params); params.dispatchCompleted = true; turnAdopted = true; settleDeferred("adopted"); @@ -308,7 +321,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro message: lastError, }); } else { - await params.journal.complete(params.id); + await completeAgentMailIngress(params); } } catch { // Best effort: TTL pruning still reclaims the row if the terminal marker cannot @@ -346,7 +359,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro message: errorText(dispatchError), }); } else { - await params.journal.complete(params.id); + await completeAgentMailIngress(params); } } catch { // Best effort: TTL pruning still reclaims the row if the terminal marker cannot persist. @@ -394,7 +407,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro params.dispatchCompleted = true; } try { - await params.journal.complete(params.id); + await completeAgentMailIngress(params); return true; } catch { completionAttempts += 1; diff --git a/src/channel/src/media.ts b/src/channel/src/media.ts index df86ae5..888e1b2 100644 --- a/src/channel/src/media.ts +++ b/src/channel/src/media.ts @@ -55,6 +55,8 @@ export async function loadAgentMailInboundAttachments(params: { const accepted = params.attachments.filter(isAcceptedAttachment); let declaredBytes = 0; for (const attachment of accepted) { + // Hydrated AgentMail.Attachment objects require `size` in the provider SDK contract. Keep the + // runtime check as a fail-closed guard for malformed or version-skewed provider payloads. const declaredSize: unknown = attachment.size; if ( typeof declaredSize !== "number" || From 7b66e0a7d2c1cfbfe36c4c4684a61a6a3d6ee24c Mon Sep 17 00:00:00 2001 From: Haakam Aujla Date: Thu, 23 Jul 2026 09:50:12 +0000 Subject: [PATCH 2/3] Fix restart recovery and timestamp bounds --- src/channel/src/catch-up.test.ts | 20 +++++++ src/channel/src/catch-up.ts | 16 +++++- src/channel/src/durable-receive.test.ts | 37 +++++++------ src/channel/src/durable-receive.ts | 3 +- src/channel/src/inbound.test.ts | 9 +++- src/channel/src/inbound.ts | 6 ++- src/channel/src/ingress.ts | 7 ++- src/channel/src/received-message.ts | 18 +++++++ src/channel/src/reply-metadata.ts | 44 ++++++++++++++++ src/channel/src/send.test.ts | 43 +++++++++++++++ src/channel/src/send.ts | 12 +++-- src/channel/src/webhook.test.ts | 26 ++++++++++ src/channel/src/webhook.ts | 15 +++++- src/channel/src/websocket.test.ts | 69 +++++++++++++++++++++++++ src/channel/src/websocket.ts | 12 ++++- 15 files changed, 311 insertions(+), 26 deletions(-) create mode 100644 src/channel/src/reply-metadata.ts diff --git a/src/channel/src/catch-up.test.ts b/src/channel/src/catch-up.test.ts index c6d4a18..0b29ee1 100644 --- a/src/channel/src/catch-up.test.ts +++ b/src/channel/src/catch-up.test.ts @@ -298,6 +298,26 @@ describe("AgentMail durable REST catch-up", () => { ); }); + it("does not admit a provider timestamp beyond the bounded future-skew window", async () => { + const store = memoryStore(); + const nowMs = 1_000_000; + const list = vi.fn(async () => ({ + count: 1, + messages: [message({ id: "far_future", timestamp: nowMs + 365 * 24 * 60 * 60_000 })], + })); + const session = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list } } } as never, + store: store as never, + now: () => nowMs, + }); + const receive = vi.fn(async () => undefined); + + await session.run({ receive, abortSignal: new AbortController().signal }); + + expect(receive).not.toHaveBeenCalled(); + }); + it("lists from the monitoring baseline on a deep sweep", async () => { const store = memoryStore(); const list = vi.fn(async () => ({ diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts index f0b3946..a03006a 100644 --- a/src/channel/src/catch-up.ts +++ b/src/channel/src/catch-up.ts @@ -9,6 +9,7 @@ import { } from "./durable-receive.js"; import { AGENTMAIL_RECEIVED_LABEL, + isAgentMailProviderTimestampWithinFutureSkew, isReceivedAgentMailMessage, resolveReceivedAgentMailMessageTimestampMs, } from "./received-message.js"; @@ -294,10 +295,23 @@ export async function createAgentMailCatchUpSession(params: { } // Invalid provider time must not turn a missed live event into a permanent drop. Admit // with local observation time; durable message-id dedupe keeps overlap scans safe. - const providerTimestampMs = resolveReceivedAgentMailMessageTimestampMs( + const rawProviderTimestampMs = resolveReceivedAgentMailMessageTimestampMs( message, params.account.inboxId, ); + if ( + rawProviderTimestampMs !== null && + !isAgentMailProviderTimestampWithinFutureSkew( + rawProviderTimestampMs, + scanUpperBoundAtMs, + ) + ) { + params.log?.warn?.( + `AgentMail catch-up ignored message ${message.messageId} timestamped too far in the future`, + ); + continue; + } + const providerTimestampMs = rawProviderTimestampMs; const admissionTimestampMs = providerTimestampMs ?? scanUpperBoundAtMs; try { await receive({ diff --git a/src/channel/src/durable-receive.test.ts b/src/channel/src/durable-receive.test.ts index db21b70..a537fb3 100644 --- a/src/channel/src/durable-receive.test.ts +++ b/src/channel/src/durable-receive.test.ts @@ -6,6 +6,7 @@ import { createAgentMailDurableInboundId, withAgentMailIngressCapacity, } from "./durable-receive.js"; +import { AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS } from "./received-message.js"; import { AgentMailIngressCapacityError, processAgentMailIngress, @@ -256,23 +257,29 @@ describe("AgentMail durable ingress", () => { }); it("retains future-dated completion tombstones through the provider scan horizon", async () => { - const futureReceivedAt = Date.now() + 24 * 60 * 60 * 1000; + const now = 1_000_000; + const futureReceivedAt = now + 365 * 24 * 60 * 60 * 1000; + const dateNow = vi.spyOn(Date, "now").mockReturnValue(now); const complete = vi.fn(async () => undefined); - await processAgentMailIngress({ - journal: { - accept: async () => ({ kind: "accepted", duplicate: false, record: {} }), - complete, - release: vi.fn(), - } as never, - record: { ...record, receivedAt: futureReceivedAt }, - dispatch: vi.fn(async () => undefined), - }); + try { + await processAgentMailIngress({ + journal: { + accept: async () => ({ kind: "accepted", duplicate: false, record: {} }), + complete, + release: vi.fn(), + } as never, + record: { ...record, receivedAt: futureReceivedAt }, + dispatch: vi.fn(async () => undefined), + }); - await vi.waitFor(() => - expect(complete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record), { - completedAt: futureReceivedAt, - }), - ); + await vi.waitFor(() => + expect(complete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record), { + completedAt: now + AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS, + }), + ); + } finally { + dateNow.mockRestore(); + } }); it("returns after durable admission without waiting for agent dispatch", async () => { diff --git a/src/channel/src/durable-receive.ts b/src/channel/src/durable-receive.ts index d149a75..cbd494b 100644 --- a/src/channel/src/durable-receive.ts +++ b/src/channel/src/durable-receive.ts @@ -62,7 +62,8 @@ export function withAgentMailIngressCapacity( // concurrent transports (live WebSocket + REST catch-up) could otherwise both observe free space // and push past the bound. Chaining admissions makes the count-then-accept step atomic across // every facade for the queue; the chain never rejects so one failed admission cannot poison - // later ones. + // later ones. This intentionally introduces brief head-of-line coupling for facades sharing one + // queue; each critical section contains only the queue admission and occasional capacity scan. const state = coordinationKey ? (capacityAdmissionStates.get(coordinationKey) ?? { admissionChain: Promise.resolve(), diff --git a/src/channel/src/inbound.test.ts b/src/channel/src/inbound.test.ts index 92d4301..00469d3 100644 --- a/src/channel/src/inbound.test.ts +++ b/src/channel/src/inbound.test.ts @@ -256,8 +256,15 @@ describe("AgentMail REST-authoritative inbound", () => { replyToId: "message_1", replyToTag: false, replyToCurrent: true, + channelData: { existing: true, agentmail: { existingAgentMail: true } }, }), - ).toEqual({ text: "reply" }); + ).toEqual({ + text: "reply", + channelData: { + existing: true, + agentmail: { existingAgentMail: true, triggerArrivedAt: 1 }, + }, + }); expect(delivery.durable()).toMatchObject({ to: "message:message_1", replyToId: "message_1", diff --git a/src/channel/src/inbound.ts b/src/channel/src/inbound.ts index 8fc46c6..ce1741a 100644 --- a/src/channel/src/inbound.ts +++ b/src/channel/src/inbound.ts @@ -13,6 +13,7 @@ import { agentMailInboxIdsEqual, resolveAgentMailTimestampMs, } from "./received-message.js"; +import { withAgentMailTriggerArrival } from "./reply-metadata.js"; import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; const CHANNEL_ID = "agentmail"; @@ -394,7 +395,10 @@ export async function dispatchAgentMailInboundEvent(params: { // implicit id owned by the durable delivery contract below, so remove those derived // payload directives immediately before delivery. preparePayload: (payload) => { - const prepared = { ...payload }; + const prepared = withAgentMailTriggerArrival( + payload, + params.record.arrivedAt ?? params.record.receivedAt, + ); delete prepared.replyToId; delete prepared.replyToTag; delete prepared.replyToCurrent; diff --git a/src/channel/src/ingress.ts b/src/channel/src/ingress.ts index e2bfb93..5edef45 100644 --- a/src/channel/src/ingress.ts +++ b/src/channel/src/ingress.ts @@ -2,6 +2,7 @@ import { type AgentMailLog, errorText } from "./log.js"; import type { createAgentMailDurableInboundReceiveJournal } from "./durable-receive.js"; import { AgentMailIngressCapacityError, createAgentMailDurableInboundId } from "./durable-receive.js"; import { HYDRATION_NOT_FOUND_RETRY_WINDOW_MS } from "./inbound.js"; +import { capAgentMailProviderTimestampForRetention } from "./received-message.js"; import { createBackoff, waitForRetry } from "./retry.js"; import type { AgentMailIngressRecord } from "./types.js"; @@ -82,10 +83,14 @@ async function completeAgentMailIngress(params: { // tombstone until seven days after that provider time, or it can re-enter the scan window only // after a locally-timestamped completion marker has already expired. const completedAt = Date.now(); - const providerReceivedAt = + const rawProviderReceivedAt = Number.isFinite(params.record.receivedAt) && params.record.receivedAt >= 0 ? params.record.receivedAt : completedAt; + const providerReceivedAt = capAgentMailProviderTimestampForRetention( + rawProviderReceivedAt, + completedAt, + ); await params.journal.complete(params.id, { completedAt: Math.max(completedAt, providerReceivedAt), }); diff --git a/src/channel/src/received-message.ts b/src/channel/src/received-message.ts index c69314f..88167b2 100644 --- a/src/channel/src/received-message.ts +++ b/src/channel/src/received-message.ts @@ -1,6 +1,10 @@ import type { AgentMail } from "agentmail"; export const AGENTMAIL_RECEIVED_LABEL = "received"; +// Email timestamps are provider-authored and may have ordinary clock skew, but allowing arbitrary +// future values into durable retention can pin tombstones for years. One day is deliberately more +// permissive than normal clock drift while keeping storage retention bounded. +export const AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS = 24 * 60 * 60 * 1000; export function normalizeAgentMailInboxId(value: string): string { return value.trim().toLocaleLowerCase("en-US"); @@ -20,6 +24,20 @@ export function resolveAgentMailTimestampMs(value: unknown): number | null { return Number.isFinite(timestampMs) && timestampMs >= 0 ? timestampMs : null; } +export function isAgentMailProviderTimestampWithinFutureSkew( + timestampMs: number, + observedAtMs: number, +): boolean { + return timestampMs <= observedAtMs + AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS; +} + +export function capAgentMailProviderTimestampForRetention( + timestampMs: number, + observedAtMs: number, +): number { + return Math.min(timestampMs, observedAtMs + AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS); +} + export function isReceivedAgentMailMessage( message: AgentMail.MessageItem | AgentMail.Message, inboxId: string, diff --git a/src/channel/src/reply-metadata.ts b/src/channel/src/reply-metadata.ts new file mode 100644 index 0000000..a9ceccb --- /dev/null +++ b/src/channel/src/reply-metadata.ts @@ -0,0 +1,44 @@ +import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; + +const AGENTMAIL_CHANNEL_DATA_KEY = "agentmail"; +const TRIGGER_ARRIVED_AT_KEY = "triggerArrivedAt"; + +function validTimestamp(value: unknown): value is number { + return typeof value === "number" && Number.isFinite(value) && value >= 0; +} + +/** Persist local inbound timing with the durable reply so restart reconciliation can classify 404s. */ +export function withAgentMailTriggerArrival( + payload: ReplyPayload, + triggerArrivedAt: number, +): ReplyPayload { + const channelData = payload.channelData ?? {}; + const currentAgentMailData = channelData[AGENTMAIL_CHANNEL_DATA_KEY]; + const agentMailData = + currentAgentMailData && typeof currentAgentMailData === "object" + ? currentAgentMailData + : {}; + return { + ...payload, + channelData: { + ...channelData, + [AGENTMAIL_CHANNEL_DATA_KEY]: { + ...agentMailData, + [TRIGGER_ARRIVED_AT_KEY]: triggerArrivedAt, + }, + }, + }; +} + +/** Read local inbound timing from a persisted reply, falling back for pre-migration queue rows. */ +export function resolveAgentMailTriggerArrival( + payload: ReplyPayload, + fallbackAt: number, +): number { + const agentMailData = payload.channelData?.[AGENTMAIL_CHANNEL_DATA_KEY]; + if (!agentMailData || typeof agentMailData !== "object") { + return fallbackAt; + } + const triggerArrivedAt = (agentMailData as Record)[TRIGGER_ARRIVED_AT_KEY]; + return validTimestamp(triggerArrivedAt) ? triggerArrivedAt : fallbackAt; +} diff --git a/src/channel/src/send.test.ts b/src/channel/src/send.test.ts index 5deb2b7..888f3f1 100644 --- a/src/channel/src/send.test.ts +++ b/src/channel/src/send.test.ts @@ -550,6 +550,49 @@ describe("AgentMail reply-only outbound", () => { expect(reply).not.toHaveBeenCalled(); }); + it("uses persisted inbound arrival time to classify a 404 after restart", async () => { + reply.mockClear(); + const unavailableClient = { + inboxes: { + messages: { + get: vi.fn(async () => { + throw new AgentMailError({ message: "not projected", statusCode: 404 }); + }), + reply, + }, + }, + } as never; + const now = 10 * 60_000; + const result = await reconcileAgentMailUnknownSend( + { + cfg: { + channels: { + agentmail: { apiKey: "key", inboxId: "inbox_1", allowFrom: ["sender@example.com"] }, + }, + }, + queueId: "queue_1", + channel: "agentmail", + to: "message:msg_1", + accountId: "default", + // The recovered queue timestamp can be stale after restart; the durable inbound metadata + // remains the authoritative projection-window reference. + enqueuedAt: 0, + retryCount: 2, + effectiveReplyToId: "msg_1", + payloads: [ + { + text: "Hello", + channelData: { agentmail: { triggerArrivedAt: now - 1_000 } }, + }, + ], + } as never, + { client: unavailableClient, now: () => now }, + ); + + expect(result).toMatchObject({ status: "unresolved", retryable: true }); + expect(reply).not.toHaveBeenCalled(); + }); + it("treats a triggering-message 404 as terminal after the projection window", async () => { const now = 10 * 60_000; const client = { diff --git a/src/channel/src/send.ts b/src/channel/src/send.ts index ad9d426..9018e6b 100644 --- a/src/channel/src/send.ts +++ b/src/channel/src/send.ts @@ -13,6 +13,7 @@ import { sha256Hex } from "./digest.js"; import { isAgentMailSenderAllowed, parseSingleFromMailbox } from "./mailbox.js"; import { AgentMailMediaPolicyError, loadAgentMailOutboundAttachments } from "./media.js"; import { agentMailInboxIdsEqual } from "./received-message.js"; +import { resolveAgentMailTriggerArrival } from "./reply-metadata.js"; import { HYDRATION_NOT_FOUND_RETRY_WINDOW_MS } from "./inbound.js"; // A media-size policy violation is genuinely terminal: the same payload will always exceed the @@ -272,9 +273,10 @@ export async function reconcileAgentMailUnknownSend( retryable: false, }; } - const recoveryReferenceAt = ctx.platformSendStartedAt ?? ctx.enqueuedAt; - const recoveryAgeMs = Math.max(0, (options.now?.() ?? Date.now()) - recoveryReferenceAt); - if (recoveryAgeMs >= AGENTMAIL_UNKNOWN_SEND_MAX_AGE_MS) { + const nowMs = options.now?.() ?? Date.now(); + const sendRecoveryReferenceAt = ctx.platformSendStartedAt ?? ctx.enqueuedAt; + const sendRecoveryAgeMs = Math.max(0, nowMs - sendRecoveryReferenceAt); + if (sendRecoveryAgeMs >= AGENTMAIL_UNKNOWN_SEND_MAX_AGE_MS) { return { status: "unresolved", error: "AgentMail recovery is too close to the provider idempotency-key expiry", @@ -290,6 +292,8 @@ export async function reconcileAgentMailUnknownSend( (value): value is string => typeof value === "string" && value.trim().length > 0, ); const text = rendered?.text ?? payload.text ?? ""; + const triggerArrivedAt = resolveAgentMailTriggerArrival(payload, ctx.enqueuedAt); + const triggerAgeMs = Math.max(0, nowMs - triggerArrivedAt); const recoveredPayload = { ...payload, text }; delete recoveredPayload.mediaUrl; delete recoveredPayload.mediaUrls; @@ -322,7 +326,7 @@ export async function reconcileAgentMailUnknownSend( if (isTerminalMediaPolicyFailure(error)) { return { status: "unresolved", error: error.message, retryable: false }; } - if (isDeletedTriggerFailure(error, recoveryAgeMs)) { + if (isDeletedTriggerFailure(error, triggerAgeMs)) { // A 404 that persists beyond the provider projection window is treated as deletion. return { status: "unresolved", error: errorText(error), retryable: false }; } diff --git a/src/channel/src/webhook.test.ts b/src/channel/src/webhook.test.ts index 85a6087..8b1c571 100644 --- a/src/channel/src/webhook.test.ts +++ b/src/channel/src/webhook.test.ts @@ -81,6 +81,32 @@ describe("AgentMail webhook", () => { ); }); + it("rejects far-future signed timestamps before durable admission", async () => { + const now = Date.now(); + const dateNow = vi.spyOn(Date, "now").mockReturnValue(now); + const receive = vi.fn(async () => undefined); + const body = JSON.stringify({ + type: "event", + event_type: "message.received", + message: { + inbox_id: "inbox_1", + message_id: "message_future", + timestamp: new Date(now + 365 * 24 * 60 * 60_000).toISOString(), + }, + }); + try { + const res = response(); + await createAgentMailWebhookHandler({ account: account(), verifier, receive })( + request(body, signed(body)), + res, + ); + expect(res.statusCode).toBe(200); + expect(receive).not.toHaveBeenCalled(); + } finally { + dateNow.mockRestore(); + } + }); + it("acknowledges but ignores a signed event with empty identifiers", async () => { const receive = vi.fn(async () => undefined); const body = JSON.stringify({ diff --git a/src/channel/src/webhook.ts b/src/channel/src/webhook.ts index 03d82cc..b096011 100644 --- a/src/channel/src/webhook.ts +++ b/src/channel/src/webhook.ts @@ -5,7 +5,11 @@ import { } from "openclaw/plugin-sdk/webhook-ingress"; import { Webhook } from "svix"; import { type AgentMailLog, errorText } from "./log.js"; -import { agentMailInboxIdsEqual, resolveAgentMailTimestampMs } from "./received-message.js"; +import { + agentMailInboxIdsEqual, + isAgentMailProviderTimestampWithinFutureSkew, + resolveAgentMailTimestampMs, +} from "./received-message.js"; import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; const MAX_WEBHOOK_BODY_BYTES = 1024 * 1024; @@ -117,6 +121,15 @@ export function createAgentMailWebhookHandler(params: { } try { const nowMs = Date.now(); + if ( + event.receivedAt !== undefined && + !isAgentMailProviderTimestampWithinFutureSkew(event.receivedAt, nowMs) + ) { + // Acknowledging a signed but policy-invalid event prevents provider retry amplification. + // REST catch-up will admit it only if its provider time later enters the bounded window. + params.log?.warn?.("AgentMail webhook ignored an event timestamped too far in the future"); + return respond(res, 200); + } await params.receive({ accountId: params.account.accountId, inboxId: params.account.inboxId, diff --git a/src/channel/src/websocket.test.ts b/src/channel/src/websocket.test.ts index f541179..8fb9558 100644 --- a/src/channel/src/websocket.test.ts +++ b/src/channel/src/websocket.test.ts @@ -112,6 +112,75 @@ describe("AgentMail WebSocket ingress", () => { expect(waitForOpen).not.toHaveBeenCalled(); }); + it("resets reconnect backoff when connect returns an already-open socket", async () => { + handlers.clear(); + connect.mockClear(); + sendSubscribe.mockClear(); + const reconnectDelayMs = vi.fn(() => 0); + const alreadyOpenSocket = { + on: (event: string, handler: (value?: unknown) => void) => handlers.set(event, handler), + sendSubscribe, + waitForOpen, + readyState: 1, + close, + }; + connect + .mockRejectedValueOnce(new Error("initial connect failed")) + .mockResolvedValueOnce(alreadyOpenSocket); + const controller = new AbortController(); + const running = startAgentMailWebSocket({ + account, + abortSignal: controller.signal, + receive: vi.fn(async () => undefined), + catchUpSession: { run: catchUpRun }, + reconnectDelayMs, + }); + + await vi.waitFor(() => expect(connect).toHaveBeenCalledTimes(2)); + await vi.waitFor(() => expect(sendSubscribe).toHaveBeenCalled()); + handlers.get("close")?.(); + await vi.waitFor(() => expect(connect).toHaveBeenCalledTimes(3)); + expect(reconnectDelayMs.mock.calls.map(([attempt]) => attempt)).toEqual([1, 1]); + + controller.abort(); + await running; + }); + + it("rejects far-future provider timestamps before durable admission", async () => { + handlers.clear(); + const now = 1_000_000; + const dateNow = vi.spyOn(Date, "now").mockReturnValue(now); + const receive = vi.fn(async () => undefined); + const controller = new AbortController(); + const running = startAgentMailWebSocket({ + account, + abortSignal: controller.signal, + receive, + catchUpSession: { run: catchUpRun }, + }); + try { + await vi.waitFor(() => expect(handlers.has("message")).toBe(true)); + catchUpRun.mockClear(); + handlers.get("message")?.({ + type: "event", + eventType: "message.received", + message: { + inboxId: "inbox_1", + messageId: "message_future", + labels: ["received"], + timestamp: new Date(now + 365 * 24 * 60 * 60_000), + }, + }); + + await vi.waitFor(() => expect(catchUpRun).toHaveBeenCalled()); + expect(receive).not.toHaveBeenCalled(); + } finally { + controller.abort(); + await running; + dateNow.mockRestore(); + } + }); + it("durably admits message.received frames before the received label projects", async () => { handlers.clear(); const receive = vi.fn(async () => undefined); diff --git a/src/channel/src/websocket.ts b/src/channel/src/websocket.ts index 44ae86e..8f41903 100644 --- a/src/channel/src/websocket.ts +++ b/src/channel/src/websocket.ts @@ -12,6 +12,7 @@ import { AgentMailIngressCapacityError } from "./ingress.js"; import { createBackoff, waitForRetry } from "./retry.js"; import { agentMailInboxIdsEqual, + isAgentMailProviderTimestampWithinFutureSkew, resolveAgentMailTimestampMs, } from "./received-message.js"; import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; @@ -187,6 +188,14 @@ export async function startAgentMailWebSocket(params: { catchUpSupervisor.request(); return; } + const arrivedAt = Date.now(); + if (!isAgentMailProviderTimestampWithinFutureSkew(messageTimestampMs, arrivedAt)) { + params.log?.warn?.( + "AgentMail WebSocket ignored an event timestamped too far in the future", + ); + catchUpSupervisor.request(); + return; + } queuedMessageIds.add(event.message.messageId); liveQueue.push({ accountId: params.account.accountId, @@ -194,7 +203,7 @@ export async function startAgentMailWebSocket(params: { messageId: event.message.messageId, transport: "websocket", receivedAt: messageTimestampMs, - arrivedAt: Date.now(), + arrivedAt, }); runLiveWorker(); }; @@ -280,6 +289,7 @@ export async function startAgentMailWebSocket(params: { // waitForOpen() does not settle on an aborted initial connection; close the already-open race // from readyState instead. if (socket.readyState === 1) { + reconnectAttempt = 0; subscribe(); } await Promise.race([closed, aborted]); From bf9786ce124c29c79d28e0f10869dfe3bc3d2a98 Mon Sep 17 00:00:00 2001 From: Haakam Aujla Date: Thu, 23 Jul 2026 10:08:39 +0000 Subject: [PATCH 3/3] Resync capacity after partial admissions --- src/channel/src/catch-up.test.ts | 29 +++++++++ src/channel/src/catch-up.ts | 7 +- src/channel/src/durable-receive.test.ts | 50 ++++++++++++++ src/channel/src/durable-receive.ts | 86 ++++++++++++++----------- src/channel/src/ingress.ts | 5 ++ 5 files changed, 140 insertions(+), 37 deletions(-) diff --git a/src/channel/src/catch-up.test.ts b/src/channel/src/catch-up.test.ts index 0b29ee1..895e515 100644 --- a/src/channel/src/catch-up.test.ts +++ b/src/channel/src/catch-up.test.ts @@ -263,6 +263,35 @@ describe("AgentMail durable REST catch-up", () => { ); }); + it("does not invert baseline catch-up bounds when the clock moves backward", async () => { + const store = memoryStore(); + const list = vi.fn(async () => ({ count: 0, messages: [] })); + let nowMs = 2_000_000; + const session = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list } } } as never, + store: store as never, + now: () => nowMs, + }); + nowMs = 1_000_000; + + await session.run({ receive: vi.fn(), abortSignal: new AbortController().signal }); + await session.run({ + receive: vi.fn(), + abortSignal: new AbortController().signal, + sinceBaseline: true, + }); + + for (const [, query] of list.mock.calls) { + expect(query).toEqual( + expect.objectContaining({ + after: new Date(nowMs), + before: new Date(nowMs + 1), + }), + ); + } + }); + it("clamps a future message timestamp before advancing the high-water cursor", async () => { const store = memoryStore(); const nowMs = 1_000_000; diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts index a03006a..e844cff 100644 --- a/src/channel/src/catch-up.ts +++ b/src/channel/src/catch-up.ts @@ -261,7 +261,12 @@ export async function createAgentMailCatchUpSession(params: { sinceBaseline || !storedCursor.established ? storedCursor.baselineAtMs : effectiveHighWaterAtMs - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS; - const afterMs = Math.max(0, baseAfterMs, dedupeFloorMs); + // A backward clock can put a persisted baseline ahead of this run. Keep the stored baseline + // unchanged for cursor state, but never send the provider an inverted after/before range. + const afterMs = Math.min( + scanUpperBoundAtMs, + Math.max(0, baseAfterMs, dedupeFloorMs), + ); let highWaterAtMs = effectiveHighWaterAtMs; let pageCursor: string | undefined; let admitted = 0; diff --git a/src/channel/src/durable-receive.test.ts b/src/channel/src/durable-receive.test.ts index a537fb3..d74ea2c 100644 --- a/src/channel/src/durable-receive.test.ts +++ b/src/channel/src/durable-receive.test.ts @@ -108,6 +108,56 @@ describe("AgentMail durable ingress", () => { expect(pendingRows).toHaveLength(3); }); + it("resynchronizes capacity after accept persists a row and then throws", async () => { + const pendingRows = new Map(); + let failAfterPersist = true; + const journal = withAgentMailIngressCapacity( + { + accept: vi.fn(async (id: string, payload: AgentMailIngressRecord) => { + const existing = pendingRows.get(id); + if (existing) { + return { + kind: "pending", + duplicate: true, + record: { ...existing, attempts: 0 }, + }; + } + const accepted = { id, payload }; + pendingRows.set(id, accepted); + if (id === "message_3" && failAfterPersist) { + failAfterPersist = false; + throw new Error("post-enqueue prune failed"); + } + return { kind: "accepted", duplicate: false, record: accepted }; + }), + pending: vi.fn(async () => + [...pendingRows.values()].map((item) => ({ ...item, attempts: 0 })), + ), + complete: vi.fn(), + release: vi.fn(), + deletePending: vi.fn(async (id: string) => pendingRows.delete(id)), + } as never, + 3, + `partial-admission-test-${crypto.randomUUID()}`, + ); + + await journal.accept("message_1", record); + await journal.accept("message_2", record); + await expect(journal.accept("message_3", record)).rejects.toThrow( + "post-enqueue prune failed", + ); + // Duplicate redelivery does not consume capacity and must remain dispatchable. + await expect(journal.accept("message_3", record)).resolves.toMatchObject({ + kind: "pending", + }); + // The next new row forces a durable recount, rolls itself back, and preserves the cap. + await expect(journal.accept("message_4", record)).rejects.toBeInstanceOf( + AgentMailIngressCapacityError, + ); + expect(pendingRows).toHaveLength(3); + expect(pendingRows.has("message_4")).toBe(false); + }); + it("keeps an overflow row retryable when rollback deletion fails", async () => { const id = createAgentMailDurableInboundId(record); const fail = vi.fn(async () => true); diff --git a/src/channel/src/durable-receive.ts b/src/channel/src/durable-receive.ts index cbd494b..19dfc49 100644 --- a/src/channel/src/durable-receive.ts +++ b/src/channel/src/durable-receive.ts @@ -48,6 +48,21 @@ type AgentMailCapacityAdmissionState = { // storage identity, rather than coordinating only calls made through one facade. const capacityAdmissionStates = new Map(); +function resolveCapacityAdmissionState( + coordinationKey: string | undefined, +): AgentMailCapacityAdmissionState { + if (!coordinationKey) { + return { admissionChain: Promise.resolve(), pendingEstimate: null }; + } + const existing = capacityAdmissionStates.get(coordinationKey); + if (existing) { + return existing; + } + const created = { admissionChain: Promise.resolve(), pendingEstimate: null }; + capacityAdmissionStates.set(coordinationKey, created); + return created; +} + /** * Wraps the store-backed journal with an admission cap. The published SDK's queue journal only * evicts on capacity; durable email must instead reject NEW mail while keeping already-accepted @@ -64,18 +79,7 @@ export function withAgentMailIngressCapacity( // every facade for the queue; the chain never rejects so one failed admission cannot poison // later ones. This intentionally introduces brief head-of-line coupling for facades sharing one // queue; each critical section contains only the queue admission and occasional capacity scan. - const state = coordinationKey - ? (capacityAdmissionStates.get(coordinationKey) ?? { - admissionChain: Promise.resolve(), - pendingEstimate: null, - }) - : { - admissionChain: Promise.resolve(), - pendingEstimate: null, - }; - if (coordinationKey && !capacityAdmissionStates.has(coordinationKey)) { - capacityAdmissionStates.set(coordinationKey, state); - } + const state = resolveCapacityAdmissionState(coordinationKey); // Upper-bound estimate of the pending count. It only increases on new admissions (never on // completion/retention pruning), so the O(pending) scan is skipped on the common below-cap path // and only runs to re-sync from the source of truth when the estimate first reaches the cap. @@ -83,34 +87,44 @@ export function withAgentMailIngressCapacity( ...journal, accept: (id, payload, options) => { const admission = state.admissionChain.then(async () => { - const accepted = await journal.accept(id, payload, options); - // Let the queue perform its atomic id lookup first. Completed tombstones and pending - // duplicates consume no new capacity and must remain harmless even while the queue is full. - if (accepted.kind !== "accepted") { - return accepted; - } - if ( - state.pendingEstimate === null || - state.pendingEstimate >= maxPendingEntries - ) { - const pending = await journal.pending(); - state.pendingEstimate = pending.length; - if (state.pendingEstimate > maxPendingEntries) { - try { - if (await journal.deletePending(id)) { - state.pendingEstimate -= 1; + try { + const accepted = await journal.accept(id, payload, options); + // Let the queue perform its atomic id lookup first. Completed tombstones and pending + // duplicates consume no new capacity and must remain harmless even while the queue is + // full. + if (accepted.kind !== "accepted") { + return accepted; + } + if ( + state.pendingEstimate === null || + state.pendingEstimate >= maxPendingEntries + ) { + const pending = await journal.pending(); + state.pendingEstimate = pending.length; + if (state.pendingEstimate > maxPendingEntries) { + try { + if (await journal.deletePending(id)) { + state.pendingEstimate -= 1; + } + } catch { + // Never turn a capacity rejection into a terminal tombstone. If rollback deletion + // is temporarily unavailable, preserve the accepted pending row: provider + // redelivery or REST catch-up re-admits that duplicate and dispatches it once + // capacity recovers. } - } catch { - // Never turn a capacity rejection into a terminal tombstone. If rollback deletion is - // temporarily unavailable, preserve the accepted pending row: provider redelivery or - // REST catch-up re-admits that duplicate and dispatches it once capacity recovers. + throw new AgentMailIngressCapacityError(); } - throw new AgentMailIngressCapacityError(); + } else { + state.pendingEstimate += 1; } - } else { - state.pendingEstimate += 1; + return accepted; + } catch (error) { + // accept() may persist the row and then fail during its post-enqueue retention prune. + // Any failure after entering the critical section makes the cached count uncertain; force + // the next new admission to resynchronize from durable storage. + state.pendingEstimate = null; + throw error; } - return accepted; }); state.admissionChain = admission.then( () => undefined, diff --git a/src/channel/src/ingress.ts b/src/channel/src/ingress.ts index 5edef45..5512cad 100644 --- a/src/channel/src/ingress.ts +++ b/src/channel/src/ingress.ts @@ -307,6 +307,11 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro return false; } if (outcome === "completion-failed") { + // Core may already have adopted the turn, but neither the completion marker nor the + // terminal failure marker could be written. Leaving the row pending preserves recovery; + // without any durable marker, a restart cannot distinguish adoption from non-adoption + // and may replay the turn. That duplicate window is unavoidable during a total marker + // store outage. return false; } turnAbandoned = true;