feat(mail): couple the delivery lease to the sender's enforced timeout (HT-22)

specs/mail/sending.md

@@ -115,7 +115,13 @@ send that outlives its own lease can be re-claimed and retried by another
 attempt while the original call is still in flight — a genuine concurrent
 double-send, not merely a race over which of two callers marks the outcome.
 Every `EmailSender` used behind these retry paths must therefore bound its
-own call time well below this lease (see §4).
+own call time well below this lease (see §4). This is enforced mechanically,
+not by convention: the `EmailSender` contract requires each implementation to
+declare the bound it itself enforces (`maxSendMs`,
+`src/providers/email-sender.ts`), and both retry paths assert
+`maxSendMs < leaseMs` before claiming a row
+(`assertLeaseExceedsSenderBound`, `src/mail/send.ts`) — a violating
+lease/timeout combination throws up front, claiming and sending nothing.
 
 **Delivery is at-least-once, not at-most-once — and nothing above changes
 that.** The idempotency key, the envelope snapshot, and the lease all close
@@ -188,7 +194,12 @@ too.** §3a's lease only holds "at most one attempt in flight per row" if the
 provider's `send()` reliably returns well inside the lease window — an
 adapter whose HTTP call has no timeout (or one comparable to or longer than
 the lease) can outlive its own claim and collide with a re-claimed retry.
-See each adapter's own timeout documentation for its bound.
+The contract makes this precondition checkable: every `EmailSender` declares
+`maxSendMs` — the bound it really enforces (a mechanical timeout on its own
+call, e.g. the Gmail adapter's `timeoutMs` feeding `AbortSignal.timeout`),
+not an estimate — and the retry paths refuse to claim under a lease that
+does not strictly exceed it (§3a). See each adapter's own timeout
+documentation for its bound.
 
 ## 5. Scope
 

src/api/index.test.ts

@@ -21,6 +21,7 @@ function createFakeSender(): { sender: EmailSender; sent: OutboundEmail[] } {
   const sent: OutboundEmail[] = []
   return {
     sender: {
+      maxSendMs: 30_000,
       async send(email) {
         sent.push(email)
         return {}
@@ -33,6 +34,7 @@ function createFakeSender(): { sender: EmailSender; sent: OutboundEmail[] } {
 /** An `EmailSender` that always rejects — for exercising the `502 send_failed` path. */
 function createThrowingSender(): EmailSender {
   return {
+    maxSendMs: 30_000,
     async send() {
       throw new Error('provider rejected the message (must never leak to the client)')
     },

src/mail/delivery-worker.test.ts

@@ -22,6 +22,7 @@ function fakeSender(): EmailSender & { sent: OutboundEmail[] } {
   const sent: OutboundEmail[] = []
   return {
     sent,
+    maxSendMs: 30_000,
     async send(email) {
       sent.push(email)
       return { providerMessageId: 'provider-1' }
@@ -268,6 +269,7 @@ describe('runDeliveryWorker', () => {
 
     const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {})
     const sender: EmailSender = {
+      maxSendMs: 30_000,
       async send() {
         throw new Error('still down')
       },
@@ -304,6 +306,7 @@ describe('runDeliveryWorker', () => {
     const seedDeps: SendReplyDeps = {
       store,
       sender: {
+        maxSendMs: 30_000,
         async send() {
           throw new Error('boom')
         },
@@ -333,6 +336,7 @@ describe('runDeliveryWorker', () => {
     })
     let workerSendCalls = 0
     const workerSender: EmailSender = {
+      maxSendMs: 30_000,
       async send() {
         workerSendCalls++
         await workerSendGate
@@ -363,4 +367,21 @@ describe('runDeliveryWorker', () => {
     expect(thread?.deliveryStatus).toBe('sent')
     expect(thread?.messageId).toBe(seeded.messageId)
   })
+
+  it("a leaseMs that does not strictly exceed the sender's maxSendMs throws up front — before anything is listed or claimed", async () => {
+    const { store } = await freshStore()
+    const listSpy = vi.spyOn(store, 'listDeliverableThreads')
+    const claimSpy = vi.spyOn(store, 'claimThreadForDelivery')
+    const sender = fakeSender() // maxSendMs: 30_000
+
+    // Equality is a violation too — the lease must STRICTLY exceed the
+    // sender's enforced bound (specs/mail/sending.md §3a).
+    await expect(
+      runDeliveryWorker({ store, sender }, { leaseMs: sender.maxSendMs }),
+    ).rejects.toThrow(/must strictly exceed/)
+
+    expect(listSpy).not.toHaveBeenCalled()
+    expect(claimSpy).not.toHaveBeenCalled()
+    expect(sender.sent).toHaveLength(0)
+  })
 })

src/mail/delivery-worker.ts

@@ -40,7 +40,11 @@
 
 import type { EmailSender } from '../providers/index.js'
 import type { ConversationStore } from '../store/conversations.js'
-import { attemptDeliveryOfClaimedThread, DEFAULT_LEASE_MS } from './send.js'
+import {
+  assertLeaseExceedsSenderBound,
+  attemptDeliveryOfClaimedThread,
+  DEFAULT_LEASE_MS,
+} from './send.js'
 
 /** Default age a `'pending'` row must reach before this worker considers it stuck rather than merely in flight. */
 const DEFAULT_STALE_AFTER_MS = 5 * 60_000
@@ -58,7 +62,13 @@ export interface DeliveryWorkerDeps {
 export interface DeliveryWorkerOptions {
   /** How old a `'pending'` row must be before it's a retry candidate (default {@link DEFAULT_STALE_AFTER_MS}). */
   staleAfterMs?: number
-  /** Lease duration held while a candidate is being attempted (default {@link DEFAULT_LEASE_MS}, shared with `sendReply`'s own retry-claim). */
+  /**
+   * Lease duration held while a candidate is being attempted (default
+   * {@link DEFAULT_LEASE_MS}, shared with `sendReply`'s own retry-claim).
+   * Must strictly exceed the sender's enforced per-`send()` bound
+   * (`EmailSender.maxSendMs`) — asserted up front, before anything is
+   * claimed (see `assertLeaseExceedsSenderBound`, `src/mail/send.ts`).
+   */
   leaseMs?: number
   /** Hard cap on rows attempted in this one call (default {@link DEFAULT_BATCH_SIZE}). */
   batchSize?: number
@@ -91,6 +101,10 @@ export async function runDeliveryWorker(
   const leaseMs = options?.leaseMs ?? DEFAULT_LEASE_MS
   const batchSize = options?.batchSize ?? DEFAULT_BATCH_SIZE
 
+  // Fail loudly on a lease/sender-timeout misconfiguration BEFORE listing or
+  // claiming anything — see this helper's doc comment in send.ts.
+  assertLeaseExceedsSenderBound(deps.sender, leaseMs)
+
   const candidates = await deps.store.listDeliverableThreads({ staleAfterMs, batchSize })
 
   let sent = 0

src/mail/send.test.ts

@@ -5,7 +5,7 @@ import type { EmailSender, OutboundEmail } from '../providers/index.js'
 import { type ConversationStore, createConversationStore } from '../store/conversations.js'
 import type { ParsedEmail } from './parse.js'
 import type { Keyring, SigningKey } from './reply-token.js'
-import { type SendReplyDeps, sendReply } from './send.js'
+import { DEFAULT_LEASE_MS, type SendReplyDeps, sendReply } from './send.js'
 import { decideThreading } from './thread.js'
 
 // --- fixtures ----------------------------------------------------------------
@@ -21,6 +21,7 @@ function fakeSender(): EmailSender & { sent: OutboundEmail[] } {
   const sent: OutboundEmail[] = []
   return {
     sent,
+    maxSendMs: 30_000,
     async send(email) {
       sent.push(email)
       return { providerMessageId: 'provider-1' }
@@ -31,6 +32,7 @@ function fakeSender(): EmailSender & { sent: OutboundEmail[] } {
 /** Always throws — simulates a provider transport failure. */
 function failingSender(): EmailSender {
   return {
+    maxSendMs: 30_000,
     async send() {
       throw new Error('boom: provider unreachable')
     },
@@ -475,6 +477,7 @@ describe('sendReply idempotency (HT-16)', () => {
     })
     let sendCallCount = 0
     const sender: EmailSender = {
+      maxSendMs: 30_000,
       async send() {
         sendCallCount++
         await sendGate
@@ -563,3 +566,72 @@ describe('sendReply idempotency (HT-16)', () => {
     expect(sender.sent).toHaveLength(0) // never re-sent — already delivered
   })
 })
+
+describe('lease / sender-bound coupling', () => {
+  let db: Db | undefined
+
+  afterEach(async () => {
+    await db?.close()
+    db = undefined
+  })
+
+  async function freshStore(): Promise<{ db: Db; store: ConversationStore }> {
+    db = await createPgliteDb()
+    await migrate(db)
+    return { db, store: createConversationStore(db) }
+  }
+
+  async function seedConversation(store: ConversationStore) {
+    return store.createConversation({
+      subject: 'Help with my order',
+      customerEmail: 'customer@example.test',
+      firstMessage: {
+        direction: 'inbound',
+        messageId: '<inbound-1@customer.example.test>',
+        fromAddress: 'customer@example.test',
+        bodyText: 'Where is my order?',
+      },
+    })
+  }
+
+  const input = (conversationId: string) => ({
+    conversationId,
+    from: 'support@example.test',
+    to: ['customer@example.test'],
+    subject: 'Re: Help with my order',
+    text: "We're looking into it!",
+  })
+
+  it('keyed path: a sender whose maxSendMs does not stay strictly below the lease throws BEFORE claiming or sending', async () => {
+    const { store } = await freshStore()
+    const { conversationId } = await seedConversation(store)
+    // Equality is deliberately a violation too — the lease must STRICTLY
+    // exceed the bound (specs/mail/sending.md §3a).
+    const sender = { ...fakeSender(), maxSendMs: DEFAULT_LEASE_MS }
+    const sendSpy = vi.spyOn(sender, 'send')
+    const claimSpy = vi.spyOn(store, 'claimThreadForDelivery')
+    const deps: SendReplyDeps = { store, sender, keyring, mailDomain }
+
+    await expect(
+      sendReply({ ...input(conversationId), idempotencyKey: 'k-violating' }, deps),
+    ).rejects.toThrow(/must strictly exceed/)
+
+    expect(claimSpy).not.toHaveBeenCalled()
+    expect(sendSpy).not.toHaveBeenCalled()
+  })
+
+  it('no-key path: the assertion does not apply — there is no lease to violate', async () => {
+    // A fresh no-key send never claims a lease, so a sender whose bound
+    // exceeds DEFAULT_LEASE_MS is not a misconfiguration ON THIS PATH; the
+    // retry paths (keyed claim above, worker sweep — see
+    // delivery-worker.test.ts) are where the invariant is enforced.
+    const { store } = await freshStore()
+    const { conversationId } = await seedConversation(store)
+    const sender = { ...fakeSender(), maxSendMs: DEFAULT_LEASE_MS * 2 }
+    const deps: SendReplyDeps = { store, sender, keyring, mailDomain }
+
+    const result = await sendReply(input(conversationId), deps)
+    expect(result).toMatchObject({ ok: true, delivery: 'sent' })
+    expect(sender.sent).toHaveLength(1)
+  })
+})

src/mail/send.ts

@@ -132,12 +132,43 @@ import { type Keyring, mintReplyMessageId } from './reply-token.js'
  * measured worst case, because none has been measured here. Any
  * `EmailSender` used behind these retry paths (§4) MUST bound its own
  * `send()` call well below this lease — via its own request timeout — so
- * this margin is never actually spent. Raising this constant without also
- * checking every adapter's timeout against it re-opens the hole it exists
- * to close.
+ * this margin is never actually spent.
+ *
+ * This relationship is enforced mechanically, not by convention: every
+ * `EmailSender` declares the bound it enforces (`maxSendMs`,
+ * `src/providers/email-sender.ts`), and both retry paths assert
+ * `maxSendMs < leaseMs` via {@link assertLeaseExceedsSenderBound} before
+ * claiming a row. Changing this constant, an adapter's timeout, or a
+ * worker's `leaseMs` option into a violating combination therefore throws
+ * at the call site instead of silently re-opening the hole.
  */
 export const DEFAULT_LEASE_MS = 120_000
 
+/**
+ * Assert the invariant {@link DEFAULT_LEASE_MS}'s doc comment exists to
+ * hold: the delivery lease strictly exceeds the sender's own enforced
+ * per-`send()` bound (`EmailSender.maxSendMs`). Called by both retry paths
+ * — `sendReply`'s keyed claim and `runDeliveryWorker`'s sweep
+ * (`src/mail/delivery-worker.ts`) — BEFORE any row is claimed, so a
+ * violating configuration fails loudly up front: nothing is claimed,
+ * nothing is sent, and the throw names both numbers.
+ *
+ * A violation is a wiring bug (an adapter timeout raised to/past the lease,
+ * or a lease tuned down below an adapter's timeout), never an expected
+ * runtime outcome — hence a throw, not a discriminated result, matching
+ * `sendReply`'s "only throw on genuinely unexpected faults" contract.
+ */
+export function assertLeaseExceedsSenderBound(sender: EmailSender, leaseMs: number): void {
+  if (!(sender.maxSendMs < leaseMs)) {
+    throw new Error(
+      `delivery lease (${leaseMs}ms) must strictly exceed the sender's enforced send() bound ` +
+        `(maxSendMs: ${sender.maxSendMs}ms), or a re-claimed retry can race a still-in-flight ` +
+        `send into a concurrent double-send (specs/mail/sending.md §3a) — ` +
+        `raise the lease or lower the sender's timeout`,
+    )
+  }
+}
+
 /** Dependencies `sendReply` needs, injected so it stays testable against fakes/in-memory stores. */
 export interface SendReplyDeps {
   store: ConversationStore
@@ -293,6 +324,7 @@ export async function sendReply(
   // pre-existing from an earlier attempt, both converge here: claim the
   // delivery lease before sending, so a concurrent duplicate call (same key)
   // or the delivery worker cannot also be sending this row right now.
+  assertLeaseExceedsSenderBound(sender, DEFAULT_LEASE_MS)
   const claimed = await store.claimThreadForDelivery(thread.id, DEFAULT_LEASE_MS)
   if (claimed === null) {
     // The claim can fail for two different reasons, and conflating them

src/providers/adapters/gmail/sender.test.ts

@@ -54,6 +54,22 @@ describe('createGmailEmailSender', () => {
     await expect(sender.send(email)).rejects.toThrow(/timeout|timed out|aborted/i)
   })
 
+  it('declares maxSendMs equal to the timeout it actually enforces (default and custom)', () => {
+    // `maxSendMs` is what the engine's retry paths assert against the
+    // delivery lease (see EmailSender.maxSendMs's doc) — it must be the SAME
+    // number as the AbortSignal.timeout bound, or the assertion checks a
+    // fiction. The abort test above proves timeoutMs is really enforced;
+    // this pins the declaration to it.
+    const { fetchImpl } = fakeFetch(200, { id: 'gmail-123' })
+    const getAccessToken = async () => 'token'
+
+    const defaulted = createGmailEmailSender({ getAccessToken, fetchImpl })
+    expect(defaulted.maxSendMs).toBe(30_000)
+
+    const custom = createGmailEmailSender({ getAccessToken, fetchImpl, timeoutMs: 5_000 })
+    expect(custom.maxSendMs).toBe(5_000)
+  })
+
   it('happy path: POSTs the encoded raw MIME to the send endpoint and returns providerMessageId', async () => {
     const { fetchImpl, calls } = fakeFetch(200, { id: 'gmail-123' })
     const getAccessToken = vi.fn(async () => 'token-abc-123')

src/providers/adapters/gmail/sender.ts

@@ -71,6 +71,12 @@ export interface GmailEmailSenderOptions {
    * delivered-but-reported-failed window every network sender has; the
    * HT-16 idempotency work is where retry-safety lands). This is the safe
    * direction: never report a delivery that can't be confirmed.
+   *
+   * This value is also declared as the adapter's `EmailSender.maxSendMs`,
+   * which the engine's retry paths assert is strictly below the delivery
+   * lease (`DEFAULT_LEASE_MS`, `src/mail/send.ts`) before claiming a row —
+   * so raising it to or past the lease makes those paths throw rather than
+   * silently risk a concurrent double-send.
    */
   timeoutMs?: number
 }
@@ -117,6 +123,11 @@ export function createGmailEmailSender(options: GmailEmailSenderOptions): EmailS
   const endpoint = `${GMAIL_API_BASE}/users/${encodeURIComponent(userId)}/messages/send`
 
   return {
+    // The same value that bounds the HTTP call below (`AbortSignal.timeout`)
+    // — one variable, so the declared bound and the enforced one cannot
+    // drift apart. See `EmailSender.maxSendMs`'s doc for what consumes this.
+    maxSendMs: timeoutMs,
+
     async send(email: OutboundEmail): Promise<EmailSendResult> {
       const raw = buildRawMessage(email)
       const encoded = Buffer.from(raw, 'utf8').toString('base64url')

src/providers/email-sender.ts

@@ -74,12 +74,32 @@ export interface EmailSendResult {
  * `Message-ID` contract every implementation must uphold.
  */
 export interface EmailSender {
+  /**
+   * The upper bound, in milliseconds, that this implementation ITSELF
+   * enforces on one `send()` call — a real, mechanical timeout (e.g. an
+   * `AbortSignal.timeout` on the underlying HTTP request), not an estimate
+   * or an aspiration. Every `send()` call MUST settle (resolve or reject)
+   * within this many milliseconds.
+   *
+   * Why the contract carries this: the delivery lease (`DEFAULT_LEASE_MS`,
+   * `src/mail/send.ts`) must strictly exceed the worst-case `send()`
+   * duration, or a re-claimed retry can race a still-in-flight send into a
+   * concurrent double-send (specs/mail/sending.md §3a, §4). The engine's
+   * retry paths assert `maxSendMs < leaseMs` before claiming a row, so an
+   * adapter whose bound is missing or too large fails loudly at the call
+   * site instead of silently re-opening that hole. Declaring a value the
+   * implementation does not actually enforce defeats the check — set it
+   * from the same variable that configures the real timeout (see the Gmail
+   * adapter's `timeoutMs`).
+   */
+  readonly maxSendMs: number
+
   /**
    * Send `email`. Resolves with an {@link EmailSendResult} on success.
    * Rejects (throws) on any failure to hand the message to the provider —
    * `src/mail/send.ts`'s `sendReply` treats a rejection as a delivery
    * failure and marks the outbound thread `'failed'` accordingly
-   * (specs/mail/sending.md §3).
+   * (specs/mail/sending.md §3). Must settle within {@link maxSendMs}.
    */
   send(email: OutboundEmail): Promise<EmailSendResult>
 }