diff --git a/specs/mail/sending.md b/specs/mail/sending.md
new file mode 100644
index 0000000..942a781
--- /dev/null
+++ b/specs/mail/sending.md
@@ -0,0 +1,103 @@
+# Outbound sending & the reply-token lifecycle
+
+Status: accepted (HT-15). Companion to [threading.md](./threading.md) — that spec
+decides which conversation an *inbound* message joins; this one covers how an
+*outbound* reply is minted, persisted, and sent, and is where the threading
+model's authority actually originates.
+
+## 1. Why sending is the load-bearing step
+
+Threading is *outbound-anchored* (threading.md §2): an inbound reply is threaded
+**only** on a signed reply token the engine minted into one of its own outbound
+`Message-ID`s. Nothing about inbound `In-Reply-To`/`References` is trusted on its
+own. That means every outbound message is a promise: the token it carries is the
+sole future handle on this conversation. If sending mints a token that doesn't
+match what's stored, or stores a token for a message that never went out, the
+thread breaks. So sending is held to the same "correctness outranks velocity"
+bar as the threading decision itself (CHARTER.md invariant #3).
+
+## 2. The id/token knot, and its resolution
+
+The outbound `Message-ID` must embed a token over `{conversationId, threadId}`
+(threading.md §2). But a thread's `threadId` is its storage primary key, and the
+`Message-ID` is a column stored on that same row — so the id must exist *before*
+the row is inserted. The database generating the id at insert time is circular.
+
+**Resolution (option A):** the application generates the outbound thread's UUID
+(`crypto.randomUUID()` — a CSPRNG) *before* persistence, mints the token from it,
+and inserts the row with `id` **and** `message_id` set together in one write.
+
+- **`threadId` in the token identifies the outbound thread that carries it** —
+  the specific outbound message. A later verified inbound reply therefore names
+  the exact message it is answering (useful for lineage, audit, and future
+  per-thread routing), even though `decideThreading` today routes on
+  `conversationId` alone.
+- **App-generated ids are safe in the HMAC.** The token's integrity is the
+  signature, never the unguessability of the id (threading.md §2). A v4 UUID is
+  a perfectly good identifier here; a DB-generated one would be no safer.
+- **UUIDs are token-safe.** `reply-token.ts`'s id charset is `[A-Za-z0-9_-]`,
+  which admits UUID hex-and-hyphens; UUIDs contain no `.`/`@`, the token's
+  structural delimiters. So a real store UUID mints and verifies unchanged.
+
+## 3. Outbound threads are outbox items
+
+An outbound thread carries an explicit **delivery status**: `pending`, `sent`,
+or `failed`. (Inbound threads have no delivery status — the column is `NULL`
+for them.) This makes "persisted" and "delivered" distinct facts, which is what
+keeps a mid-flight failure from lying.
+
+**Ordering — persist, then send, then mark:**
+
+1. Generate `threadId`; mint the token → `messageId`.
+2. Persist the outbound thread with `delivery_status = 'pending'` and
+   `message_id = messageId`.
+3. Call the sender provider (§4).
+4. On success → `delivery_status = 'sent'`; on failure → `'failed'`.
+
+A crash at any point leaves a truthful record: a thread stuck at `pending` means
+"we may or may not have delivered it," never a false `sent`. Send-*then*-persist
+is rejected — a crash after a successful send would lose the outbound message
+from the conversation entirely.
+
+**Retries reuse, never re-mint.** A `failed` (or orphaned `pending`) outbound
+thread is re-attempted with the **same** `threadId` and the **same**
+`Message-ID`. Minting a fresh token per attempt would spray multiple valid
+threading handles for one logical message and risk double-sends. The stable
+`Message-ID` is the idempotency anchor: a provider that de-dupes on `Message-ID`
+will not double-deliver a retried send.
+
+## 4. What a sender provider must guarantee
+
+The `EmailSender` provider (`src/providers/`) is handed a fully-formed outbound
+message and MUST transmit the engine-supplied `Message-ID` **verbatim** as the
+RFC 5322 `Message-ID` header — not generate or overwrite its own. Threading
+depends on it; a provider that cannot set `Message-ID` is unusable for
+Helpthread. `In-Reply-To` and `References` are likewise engine-set and must be
+transmitted as given.
+
+The interface can only state this contract; it cannot enforce it. Therefore
+**every real `EmailSender` adapter MUST ship with a wire-level contract test**
+asserting the exact `Message-ID`/`In-Reply-To`/`References` it emits (against
+the raw MIME or provider-API payload it produces), because an adapter whose SDK
+silently rewrites `Message-ID` would pass `sendReply` (the thread is marked
+`sent`) while every future reply fails to thread. Prefer provider APIs that
+accept raw MIME; reject any that will not carry `Message-ID` unaltered. The
+in-repo fake used by the engine tests proves only that `sendReply` *passes* the
+value to the seam — not that any given adapter preserves it on the wire.
+
+## 5. Scope of the first increment (HT-15)
+
+Deliberately narrow; each deferral below has a named later home:
+
+- **Synchronous send only** — the persist→send→mark flow runs inline. No queue
+  or retry worker yet; the `failed` status plus the stable id/`Message-ID` are
+  the seam a later delivery worker (queue provider, already interfaced) picks up.
+- **Reply to an existing conversation only.** Agent-*initiated* brand-new
+  conversations are a separate later flow.
+- **`In-Reply-To`/`References` are caller-supplied** (from the inbound message
+  being answered). Deriving the full `References` chain from stored threads is a
+  later refinement.
+- **A missing or deleted conversation is refused** — the token is minted first
+  (before `appendThread` resolves) and then discarded on refusal; only
+  persistence and sending are skipped, and the sender is never called (mirrors
+  the store's `appendThread` policy; threading.md §5).
diff --git a/src/db/migrate.test.ts b/src/db/migrate.test.ts
index d90d2c5..3ae121d 100644
--- a/src/db/migrate.test.ts
+++ b/src/db/migrate.test.ts
@@ -32,14 +32,17 @@ describe('migrate', () => {
     expect(thread.id).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/)
   })
 
-  it('records exactly one _migrations row for migration 001', async () => {
+  it('records exactly one _migrations row per migration', async () => {
     db = await createPgliteDb()
     await migrate(db)
 
     const rows = await db.query<{ id: number; name: string }>(
       'SELECT id, name FROM _migrations ORDER BY id',
     )
-    expect(rows).toEqual([{ id: 1, name: 'conversations_and_threads' }])
+    expect(rows).toEqual([
+      { id: 1, name: 'conversations_and_threads' },
+      { id: 2, name: 'add_thread_delivery_status' },
+    ])
   })
 
   it('is idempotent: a second call is a clean no-op', async () => {
@@ -48,6 +51,95 @@ describe('migrate', () => {
     await migrate(db) // must not throw (e.g. "relation already exists")
 
     const rows = await db.query<{ id: number }>('SELECT id FROM _migrations ORDER BY id')
-    expect(rows).toEqual([{ id: 1 }])
+    expect(rows).toEqual([{ id: 1 }, { id: 2 }])
+  })
+
+  it('migration 002 ties delivery_status to direction: inbound must be NULL, outbound must be pending/sent/failed', async () => {
+    db = await createPgliteDb()
+    await migrate(db)
+
+    const [conversation] = await db.query<{ id: string }>(
+      'INSERT INTO conversations (customer_email) VALUES ($1) RETURNING id',
+      ['customer@example.test'],
+    )
+
+    // Inbound → NULL is the only legal value.
+    const [nullRow] = await db.query<{ delivery_status: string | null }>(
+      `INSERT INTO threads (conversation_id, direction, from_address)
+       VALUES ($1, 'inbound', $2) RETURNING delivery_status`,
+      [conversation.id, 'customer@example.test'],
+    )
+    expect(nullRow.delivery_status).toBeNull()
+
+    // Outbound → one of the three outbox states.
+    const [pendingRow] = await db.query<{ delivery_status: string | null }>(
+      `INSERT INTO threads (conversation_id, direction, from_address, delivery_status)
+       VALUES ($1, 'outbound', $2, 'pending') RETURNING delivery_status`,
+      [conversation.id, 'support@example.test'],
+    )
+    expect(pendingRow.delivery_status).toBe('pending')
+
+    // Outbound with an out-of-domain value → rejected.
+    await expect(
+      db.query(
+        `INSERT INTO threads (conversation_id, direction, from_address, delivery_status)
+         VALUES ($1, 'outbound', $2, 'bogus')`,
+        [conversation.id, 'support@example.test'],
+      ),
+    ).rejects.toThrow()
+
+    // Cross-column invariant: an INBOUND thread may NOT carry a status...
+    await expect(
+      db.query(
+        `INSERT INTO threads (conversation_id, direction, from_address, delivery_status)
+         VALUES ($1, 'inbound', $2, 'sent')`,
+        [conversation.id, 'customer@example.test'],
+      ),
+    ).rejects.toThrow()
+
+    // ...and an OUTBOUND thread may NOT be left NULL (invisible to a delivery worker).
+    await expect(
+      db.query(
+        `INSERT INTO threads (conversation_id, direction, from_address, delivery_status)
+         VALUES ($1, 'outbound', $2, NULL)`,
+        [conversation.id, 'support@example.test'],
+      ),
+    ).rejects.toThrow()
+  })
+
+  it('migration 002 upgrades a NON-fresh 001 database with preexisting outbound rows (backfills, does not fail)', async () => {
+    db = await createPgliteDb()
+
+    // Apply ONLY migration 001, then write an outbound thread the way a
+    // pre-002 deployment would have — no delivery_status column yet.
+    await migrate(db, { throughId: 1 })
+    const [conversation] = await db.query<{ id: string }>(
+      'INSERT INTO conversations (customer_email) VALUES ($1) RETURNING id',
+      ['customer@example.test'],
+    )
+    const [outbound] = await db.query<{ id: string }>(
+      `INSERT INTO threads (conversation_id, direction, from_address)
+       VALUES ($1, 'outbound', $2) RETURNING id`,
+      [conversation.id, 'support@example.test'],
+    )
+
+    // Now apply 002 over that existing data. Without the backfill this throws
+    // (the preexisting outbound row is NULL and violates the new CHECK).
+    await expect(migrate(db)).resolves.toBeUndefined()
+
+    // The preexisting outbound row was backfilled to 'pending', and the
+    // constraint is now live (a fresh NULL outbound insert is rejected).
+    const [row] = await db.query<{ delivery_status: string | null }>(
+      'SELECT delivery_status FROM threads WHERE id = $1',
+      [outbound.id],
+    )
+    expect(row.delivery_status).toBe('pending')
+    await expect(
+      db.query(
+        `INSERT INTO threads (conversation_id, direction, from_address, delivery_status)
+         VALUES ($1, 'outbound', $2, NULL)`,
+        [conversation.id, 'support@example.test'],
+      ),
+    ).rejects.toThrow()
   })
 })
diff --git a/src/db/migrate.ts b/src/db/migrate.ts
index d3d2b26..5d2c4ad 100644
--- a/src/db/migrate.ts
+++ b/src/db/migrate.ts
@@ -61,6 +61,47 @@ CREATE TABLE threads (
 CREATE INDEX threads_conversation_id_idx ON threads (conversation_id);
 `
 
+/**
+ * Migration 002 — outbound delivery status (specs/mail/sending.md §3).
+ *
+ * An outbound thread is an outbox item: it carries `pending`/`sent`/`failed`
+ * to make "persisted" and "delivered" distinct facts (a crash mid-send must
+ * never be misreported as delivered). Inbound threads have no delivery
+ * concept, so the column stays `NULL` for them.
+ *
+ * The constraint is a CROSS-COLUMN (table-level) invariant tying status to
+ * direction, not a value-only check: an inbound row MUST be `NULL` and an
+ * outbound row MUST be one of the three states. This makes the illegal
+ * states — an inbound thread marked `'sent'`, or an outbound thread with a
+ * `NULL` status invisible to a future delivery worker — unrepresentable at
+ * the database level, not merely discouraged in application code (a
+ * table-level constraint is added with a separate `ADD CONSTRAINT` because an
+ * inline `ADD COLUMN ... CHECK` may only reference its own column).
+ */
+// NOTE on the explicit \`delivery_status IS NOT NULL\` in the outbound branch:
+// a CHECK constraint passes on TRUE *or* NULL (unknown) and only fails on
+// FALSE. Without the IS-NOT-NULL guard, an outbound row with a NULL status
+// makes \`delivery_status IN (...)\` evaluate to NULL, so the whole CHECK is
+// NULL and the row is (wrongly) ACCEPTED — the exact "outbound with no status,
+// invisible to the delivery worker" state this constraint exists to forbid.
+// The guard forces that case to FALSE so it is rejected.
+// The BACKFILL between ADD COLUMN and ADD CONSTRAINT is load-bearing, not
+// cosmetic: on a database that already ran migration 001 and stored outbound
+// threads, ADD COLUMN gives those rows a NULL delivery_status, which the new
+// direction-tied CHECK (with its IS NOT NULL guard) would then REJECT —
+// failing the whole migration on any non-fresh database. Backfilling existing
+// outbound rows to 'pending' (a truthful "delivery state unknown/unconfirmed"
+// for rows that predate delivery tracking) makes them satisfy the constraint
+// before it is added. Inbound rows correctly stay NULL.
+const MIGRATION_002_ADD_THREAD_DELIVERY_STATUS = `
+ALTER TABLE threads ADD COLUMN delivery_status text;
+UPDATE threads SET delivery_status = 'pending' WHERE direction = 'outbound' AND delivery_status IS NULL;
+ALTER TABLE threads ADD CONSTRAINT threads_delivery_status_by_direction CHECK (
+  (direction = 'inbound' AND delivery_status IS NULL)
+  OR (direction = 'outbound' AND delivery_status IS NOT NULL AND delivery_status IN ('pending','sent','failed'))
+);
+`
+
 /**
  * Every migration, in the order they must apply. `id` is the sole ordering
  * key (ascending) — array position is not relied upon, so re-sorting this
@@ -68,6 +109,11 @@ CREATE INDEX threads_conversation_id_idx ON threads (conversation_id);
  */
 const MIGRATIONS: Migration[] = [
   { id: 1, name: 'conversations_and_threads', sql: MIGRATION_001_CONVERSATIONS_AND_THREADS },
+  {
+    id: 2,
+    name: 'add_thread_delivery_status',
+    sql: MIGRATION_002_ADD_THREAD_DELIVERY_STATUS,
+  },
 ]
 
 /**
@@ -138,8 +184,18 @@ const MIGRATION_ADVISORY_LOCK_KEY = 4_137_231_984
  * only reproducible against a real multi-connection server, so it is not
  * unit-testable here. The idempotency test covers the apply-once bookkeeping;
  * true concurrent-migrate coverage waits for the Supabase-backed `Db`.)
+ *
+ * ## `throughId`
+ *
+ * `options.throughId` applies only migrations with `id <= throughId`, leaving
+ * later ones pending. Its main use is staged rollouts and testing forward
+ * UPGRADE paths — applying an earlier schema, writing data against it, then
+ * applying the next migration over that data (exactly what a real deploy does,
+ * and what a fresh-only test never exercises). Omitted, every pending
+ * migration is applied.
  */
-export async function migrate(db: Db): Promise<void> {
+export async function migrate(db: Db, options?: { throughId?: number }): Promise<void> {
+  const throughId = options?.throughId
   await db.transaction(async (tx) => {
     // Serialize concurrent migrate() runs before touching any state. A bare
     // integer key needs no table, so this is safe to take before `_migrations`
@@ -157,9 +213,10 @@ export async function migrate(db: Db): Promise<void> {
     const applied = await tx.query<{ id: number }>('SELECT id FROM _migrations')
     const appliedIds = new Set(applied.map((row) => row.id))
 
-    const pending = MIGRATIONS.filter((migration) => !appliedIds.has(migration.id)).sort(
-      (a, b) => a.id - b.id,
-    )
+    const pending = MIGRATIONS.filter(
+      (migration) =>
+        !appliedIds.has(migration.id) && (throughId === undefined || migration.id <= throughId),
+    ).sort((a, b) => a.id - b.id)
 
     for (const migration of pending) {
       for (const statement of splitStatements(migration.sql)) {
diff --git a/src/mail/send.test.ts b/src/mail/send.test.ts
new file mode 100644
index 0000000..748b93a
--- /dev/null
+++ b/src/mail/send.test.ts
@@ -0,0 +1,267 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { createPgliteDb, type Db } from '../db/client.js'
+import { migrate } from '../db/migrate.js'
+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 { decideThreading } from './thread.js'
+
+// --- fixtures ----------------------------------------------------------------
+
+const RANDOM_UUID = '00000000-0000-4000-8000-000000000000'
+
+const KEY_A: SigningKey = { keyId: 'k1', secret: 'secret-A-high-entropy-0123456789abcdef' }
+const keyring: Keyring = { current: KEY_A }
+const mailDomain = 'mail.example.test'
+
+/** Records every `OutboundEmail` it is asked to send; never fails. */
+function fakeSender(): EmailSender & { sent: OutboundEmail[] } {
+  const sent: OutboundEmail[] = []
+  return {
+    sent,
+    async send(email) {
+      sent.push(email)
+      return { providerMessageId: 'provider-1' }
+    },
+  }
+}
+
+/** Always throws — simulates a provider transport failure. */
+function failingSender(): EmailSender {
+  return {
+    async send() {
+      throw new Error('boom: provider unreachable')
+    },
+  }
+}
+
+/** Minimal ParsedEmail builder — only threading-relevant fields vary per test. */
+function inboundReplyTo(messageId: string): ParsedEmail {
+  return {
+    messageId: '<inbound-2@customer.example.test>',
+    inReplyTo: messageId,
+    references: [],
+    from: { address: 'customer@example.test' },
+    to: [{ address: 'support@example.test' }],
+    cc: [],
+    subject: 'Re: Help with my order',
+    date: null,
+    text: 'Thanks, still broken though.',
+    html: null,
+    headers: {},
+    attachments: [],
+  }
+}
+
+/** Directly flips a conversation's status for test setup. */
+async function setStatus(db: Db, conversationId: string, status: 'open' | 'closed' | 'deleted') {
+  await db.query('UPDATE conversations SET status = $1 WHERE id = $2', [status, conversationId])
+}
+
+// --- suite ---------------------------------------------------------------------
+
+describe('sendReply', () => {
+  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?',
+      },
+    })
+  }
+
+  it('happy path: persists a sent outbound thread and hands the exact messageId to the sender', async () => {
+    const { store } = await freshStore()
+    const { conversationId } = await seedConversation(store)
+    const sender = fakeSender()
+    const deps: SendReplyDeps = { store, sender, keyring, mailDomain }
+
+    const result = await sendReply(
+      {
+        conversationId,
+        from: 'support@example.test',
+        to: ['customer@example.test'],
+        subject: 'Re: Help with my order',
+        text: "We're looking into it!",
+        inReplyTo: '<inbound-1@customer.example.test>',
+      },
+      deps,
+    )
+
+    expect(result.ok).toBe(true)
+    if (!result.ok) throw new Error('unreachable')
+    expect(result.delivery).toBe('sent')
+    expect(result.threadId).toEqual(expect.any(String))
+    expect(result.messageId).toMatch(/^<ht\.k1\./)
+
+    // The fake sender received the EXACT engine-minted messageId, verbatim.
+    expect(sender.sent).toHaveLength(1)
+    expect(sender.sent[0].messageId).toBe(result.messageId)
+    expect(sender.sent[0].inReplyTo).toBe('<inbound-1@customer.example.test>')
+
+    const conversation = await store.getConversation(conversationId)
+    const outbound = conversation?.threads.find((t) => t.direction === 'outbound')
+    expect(outbound).toMatchObject({
+      id: result.threadId,
+      messageId: result.messageId,
+      deliveryStatus: 'sent',
+      bodyText: "We're looking into it!",
+    })
+  })
+
+  it('round-trip: a reply to the minted messageId threads back to the same conversation and outbound thread', async () => {
+    const { store } = await freshStore()
+    const { conversationId } = await seedConversation(store)
+    const sender = fakeSender()
+    const deps: SendReplyDeps = { store, sender, keyring, mailDomain }
+
+    const sent = await sendReply(
+      {
+        conversationId,
+        from: 'support@example.test',
+        to: ['customer@example.test'],
+        subject: 'Re: Help with my order',
+        text: "We're looking into it!",
+      },
+      deps,
+    )
+    if (!sent.ok) throw new Error('unreachable')
+
+    const inbound = inboundReplyTo(sent.messageId)
+    const decision = decideThreading(inbound, keyring)
+
+    expect(decision).toEqual({
+      kind: 'append',
+      conversationId,
+      threadId: sent.threadId,
+      forgedTokenCount: 0,
+    })
+  })
+
+  it('send failure: sendReply re-throws and the outbound thread is left failed', async () => {
+    const { store } = await freshStore()
+    const { conversationId } = await seedConversation(store)
+    const deps: SendReplyDeps = { store, sender: failingSender(), keyring, mailDomain }
+
+    await expect(
+      sendReply(
+        {
+          conversationId,
+          from: 'support@example.test',
+          to: ['customer@example.test'],
+          subject: 'Re: Help with my order',
+          text: "We're looking into it!",
+        },
+        deps,
+      ),
+    ).rejects.toThrow('boom: provider unreachable')
+
+    const conversation = await store.getConversation(conversationId)
+    const outbound = conversation?.threads.find((t) => t.direction === 'outbound')
+    expect(outbound).toMatchObject({ deliveryStatus: 'failed' })
+  })
+
+  it('send failure AND mark failure: both errors surface via AggregateError; the send cause is not lost', async () => {
+    const { store: realStore } = await freshStore()
+    const { conversationId } = await seedConversation(realStore)
+    // Wrap the real store so the 'failed' mark itself throws — the worst case
+    // where a DB blip lands right after the provider rejected.
+    const store: ConversationStore = {
+      ...realStore,
+      async setThreadDeliveryStatus() {
+        throw new Error('db down: cannot mark thread failed')
+      },
+    }
+    const deps: SendReplyDeps = { store, sender: failingSender(), keyring, mailDomain }
+
+    let caught: unknown
+    try {
+      await sendReply(
+        {
+          conversationId,
+          from: 'support@example.test',
+          to: ['customer@example.test'],
+          subject: 'Re: Help with my order',
+          text: "We're looking into it!",
+        },
+        deps,
+      )
+      throw new Error('unreachable: sendReply should have thrown')
+    } catch (err) {
+      caught = err
+    }
+
+    expect(caught).toBeInstanceOf(AggregateError)
+    const messages = (caught as AggregateError).errors.map((e) => (e as Error).message)
+    // The ORIGINAL provider failure is preserved, not swapped for the DB error.
+    expect(messages.some((m) => m.includes('boom: provider unreachable'))).toBe(true)
+    expect(messages.some((m) => m.includes('db down: cannot mark thread failed'))).toBe(true)
+  })
+
+  it('refused: a deleted conversation is refused, the sender is never called, and nothing is added', async () => {
+    const { db: rawDb, store } = await freshStore()
+    const { conversationId } = await seedConversation(store)
+    await setStatus(rawDb, conversationId, 'deleted')
+
+    const sender = fakeSender()
+    const sendSpy = vi.spyOn(sender, 'send')
+    const deps: SendReplyDeps = { store, sender, keyring, mailDomain }
+
+    const result = await sendReply(
+      {
+        conversationId,
+        from: 'support@example.test',
+        to: ['customer@example.test'],
+        subject: 'Re: Help with my order',
+        text: "We're looking into it!",
+      },
+      deps,
+    )
+
+    expect(result).toEqual({ ok: false, reason: 'conversation-deleted' })
+    expect(sendSpy).not.toHaveBeenCalled()
+
+    const conversation = await store.getConversation(conversationId)
+    expect(conversation?.threads).toHaveLength(1)
+  })
+
+  it('refused: a missing conversation is refused and the sender is never called', async () => {
+    const { store } = await freshStore()
+    const sender = fakeSender()
+    const sendSpy = vi.spyOn(sender, 'send')
+    const deps: SendReplyDeps = { store, sender, keyring, mailDomain }
+
+    const result = await sendReply(
+      {
+        conversationId: RANDOM_UUID,
+        from: 'support@example.test',
+        to: ['customer@example.test'],
+        subject: 'Re: Help with my order',
+        text: "We're looking into it!",
+      },
+      deps,
+    )
+
+    expect(result).toEqual({ ok: false, reason: 'conversation-not-found' })
+    expect(sendSpy).not.toHaveBeenCalled()
+  })
+})
diff --git a/src/mail/send.ts b/src/mail/send.ts
new file mode 100644
index 0000000..fac3cbf
--- /dev/null
+++ b/src/mail/send.ts
@@ -0,0 +1,177 @@
+/**
+ * Outbound send orchestration — mint, persist, send, mark (specs/mail/sending.md
+ * §3; companion to specs/mail/threading.md, which this closes the loop on).
+ *
+ * This is the ONE place `mintReplyMessageId` (`src/mail/reply-token.ts`) is
+ * called on the write path: every outbound reply's `Message-ID` originates
+ * here, and every later inbound reply's threading decision
+ * (`decideThreading`, `src/mail/thread.ts`) is only as trustworthy as this
+ * function's ordering.
+ *
+ * ## Ordering: persist, THEN send, THEN mark (specs/mail/sending.md §3)
+ *
+ * 1. Generate `threadId` (a CSPRNG UUID, `crypto.randomUUID()`) and mint
+ *    `messageId` from it — the id/token knot's resolution (specs/mail/sending.md
+ *    §2): the thread's own primary key must exist before the row is
+ *    inserted, because the `Message-ID` embeds it, and the `Message-ID` is a
+ *    column ON that same row.
+ * 2. Persist the outbound thread with `delivery_status = 'pending'` via
+ *    `ConversationStore.appendThread`.
+ * 3. Only once persisted, call the `EmailSender`.
+ * 4. Mark `'sent'` or `'failed'` depending on the outcome.
+ *
+ * Send-then-persist is deliberately rejected by the spec: a crash after a
+ * successful send but before persisting would lose the outbound message
+ * from the conversation entirely — an unrecoverable data loss the
+ * persist-first ordering here structurally cannot produce. The worst this
+ * ordering can do is leave a thread stuck at `'pending'` (truthful: "may or
+ * may not have been delivered"), never a false `'sent'`.
+ *
+ * ## Retries reuse, never re-mint (specs/mail/sending.md §3)
+ *
+ * When the provider `send()` call throws, this function marks the thread
+ * `'failed'` and RE-THROWS — it does not swallow the error, retry inline, or
+ * mint a fresh token. A `failed` (or crash-orphaned `pending`) thread is
+ * meant to be retried later by a queue worker (not built in this increment
+ * — specs/mail/sending.md §5) using the SAME `threadId`/`messageId` already
+ * on the row. Minting a new token per attempt would spray multiple valid
+ * threading handles for one logical message and risk a provider that
+ * de-dupes on `Message-ID` failing to catch a double-send.
+ *
+ * ## Caller responsibility: idempotency is NOT yet handled here (HT-16)
+ *
+ * This increment has no idempotency key and no "retry an existing pending/
+ * failed thread" path: each `sendReply` call mints a FRESH `threadId`/
+ * `Message-ID` and sends. So a caller that retries the same logical reply
+ * (an HTTP timeout, a double-clicked UI, a queue redelivery) will send a
+ * SECOND email. Until the delivery-worker increment adds a real dedup key
+ * (HT-16), callers MUST guarantee at-most-once invocation themselves —
+ * `sendReply` must not be wired directly behind a retrying transport.
+ *
+ * ## Assumption: ids are canonical
+ *
+ * `conversationId` is expected to be a canonical (lowercase) id as produced
+ * by the store — it is embedded verbatim into the token, so a non-canonical
+ * spelling (e.g. an upper-cased UUID) would be what `decideThreading` later
+ * recovers, even though the DB stores the canonical form. The store only ever
+ * emits canonical ids and callers pass those straight through, so this holds
+ * by construction; it is called out because the token carries the string, not
+ * a parsed UUID.
+ */
+
+import { randomUUID } from 'node:crypto'
+import type { EmailSender } from '../providers/index.js'
+import type { ConversationStore } from '../store/conversations.js'
+import { type Keyring, mintReplyMessageId } from './reply-token.js'
+
+/** Dependencies `sendReply` needs, injected so it stays testable against fakes/in-memory stores. */
+export interface SendReplyDeps {
+  store: ConversationStore
+  sender: EmailSender
+  keyring: Keyring
+  /** The domain minted into the outbound `Message-ID`'s `@domain` part (see `mintReplyMessageId`). */
+  mailDomain: string
+}
+
+/** One outbound reply to an existing conversation (specs/mail/sending.md §5: reply-only in this increment). */
+export interface SendReplyInput {
+  conversationId: string
+  from: string
+  to: string[]
+  cc?: string[]
+  subject: string
+  text?: string
+  html?: string
+  /** `In-Reply-To` of the inbound message being answered — caller-supplied (specs/mail/sending.md §5). */
+  inReplyTo?: string
+  /** `References` chain of the inbound message being answered — caller-supplied (specs/mail/sending.md §5). */
+  references?: string[]
+}
+
+/**
+ * The outcome of {@link sendReply}. Modeled as an explicit discriminated
+ * result rather than throw/catch for the REFUSAL cases (missing/deleted
+ * conversation) — mirroring `ConversationStore.appendThread`'s `AppendResult`
+ * — because a reply aimed at a conversation that no longer accepts mail is
+ * expected, not exceptional. A provider SEND failure is different: that
+ * throws (see the module doc's "retries reuse, never re-mint" note), because
+ * by that point the thread is already durably persisted and the failure is
+ * the caller's problem to react to, not a routine control-flow branch.
+ */
+export type SendReplyResult =
+  | { ok: true; threadId: string; messageId: string; delivery: 'sent' }
+  | { ok: false; reason: 'conversation-not-found' | 'conversation-deleted' }
+
+/**
+ * Send a reply to an existing conversation, per the persist→send→mark
+ * ordering in the module doc. See there for the full ordering and retry
+ * rationale.
+ *
+ * Refusal (missing or deleted conversation): the token is minted before the
+ * `appendThread` call resolves, then discarded when refusal is detected —
+ * harmless, since it was never persisted or handed to the sender. The
+ * `EmailSender` is NEVER invoked in a refusal case.
+ */
+export async function sendReply(
+  input: SendReplyInput,
+  deps: SendReplyDeps,
+): Promise<SendReplyResult> {
+  const { store, sender, keyring, mailDomain } = deps
+
+  const threadId = randomUUID()
+  const messageId = mintReplyMessageId(
+    { conversationId: input.conversationId, threadId, mailDomain },
+    keyring,
+  )
+
+  const appended = await store.appendThread(input.conversationId, {
+    id: threadId,
+    direction: 'outbound',
+    messageId,
+    inReplyTo: input.inReplyTo ?? null,
+    fromAddress: input.from,
+    bodyText: input.text ?? null,
+    bodyHtml: input.html ?? null,
+    deliveryStatus: 'pending',
+  })
+
+  if (!appended.ok) {
+    // Nothing was persisted; the minted token above is discarded unused.
+    return {
+      ok: false,
+      reason: appended.reason === 'not-found' ? 'conversation-not-found' : 'conversation-deleted',
+    }
+  }
+
+  try {
+    await sender.send({
+      messageId,
+      inReplyTo: input.inReplyTo,
+      references: input.references,
+      from: input.from,
+      to: input.to,
+      cc: input.cc,
+      subject: input.subject,
+      text: input.text,
+      html: input.html,
+    })
+  } catch (sendErr) {
+    // Mark 'failed' best-effort, but never let a failure of the MARK bury the
+    // original send failure — its cause is what a caller/operator needs to
+    // act on. If the mark ALSO throws (e.g. a transient DB error right after
+    // the provider rejected), surface both rather than silently swapping one
+    // for the other.
+    try {
+      await store.setThreadDeliveryStatus(threadId, 'failed')
+    } catch (markErr) {
+      throw new AggregateError(
+        [sendErr, markErr],
+        'send failed, and marking the outbound thread failed also failed',
+      )
+    }
+    throw sendErr
+  }
+
+  await store.setThreadDeliveryStatus(threadId, 'sent')
+  return { ok: true, threadId, messageId, delivery: 'sent' }
+}
diff --git a/src/providers/README.md b/src/providers/README.md
index 15d47a1..731c0d8 100644
--- a/src/providers/README.md
+++ b/src/providers/README.md
@@ -5,17 +5,19 @@ calls a platform directly: queueing, scheduled and durable work, blob
 storage, and inbound email all sit behind thin provider interfaces the
 project owns, with today's implementations (Vercel Queues, Vercel Cron and
 Workflows, Supabase Storage, Gmail push) as adapters"** rather than
-assumptions baked into engine code.
+assumptions baked into engine code. Outbound email (`EmailSender`,
+specs/mail/sending.md §4) follows the same discipline, even though it
+predates a name-check in that charter sentence.
 
 ## The rule
 
 **Engine core imports only from `src/providers`** — the interfaces and types
 defined in this directory — never a platform SDK (`@vercel/*`,
 `@supabase/*`, `googleapis`, etc.) directly. If an engine module needs to
-enqueue work, schedule an action, store a blob, or read an inbound email, it
-takes a dependency on the relevant interface (`QueueProvider`,
-`SchedulerProvider`, `BlobStore`, `InboundEmailProvider`) — never on the
-package that implements it.
+enqueue work, schedule an action, store a blob, read an inbound email, or
+send an outbound email, it takes a dependency on the relevant interface
+(`QueueProvider`, `SchedulerProvider`, `BlobStore`, `InboundEmailProvider`,
+`EmailSender`) — never on the package that implements it.
 
 Concrete implementations — **adapters** — live in `src/providers/adapters/<name>/`
 (e.g. `src/providers/adapters/vercel-queues/`). This task defines the
@@ -45,8 +47,8 @@ seams the charter names.
 
 Because engine code depends on these interfaces rather than concrete SDKs,
 the engine's test suite runs against **in-memory fakes** of `QueueProvider`,
-`SchedulerProvider`, `BlobStore`, and `InboundEmailProvider` — no cloud
-account, network call, or platform emulator required to exercise queueing,
-scheduling, storage, or inbound-mail logic. Adapters get their own
-integration tests against the real platform; engine logic does not need
-those to run.
+`SchedulerProvider`, `BlobStore`, `InboundEmailProvider`, and `EmailSender`
+— no cloud account, network call, or platform emulator required to exercise
+queueing, scheduling, storage, inbound-mail, or outbound-send logic.
+Adapters get their own integration tests against the real platform; engine
+logic does not need those to run.
diff --git a/src/providers/email-sender.ts b/src/providers/email-sender.ts
new file mode 100644
index 0000000..a39faa1
--- /dev/null
+++ b/src/providers/email-sender.ts
@@ -0,0 +1,85 @@
+/**
+ * `EmailSender` — the seam for transmitting outbound mail.
+ *
+ * See `src/providers/README.md` for the pattern this fits into. This is the
+ * `send()` half of specs/mail/sending.md; `src/mail/send.ts` (`sendReply`)
+ * is the engine-side orchestration that calls it as step 3 of the
+ * persist→send→mark ordering (specs/mail/sending.md §3).
+ *
+ * ## The `Message-ID` contract is load-bearing
+ *
+ * Outbound-anchored threading (specs/mail/threading.md §2, specs/mail/sending.md
+ * §1) only works if the `Message-ID` a customer's reply eventually echoes
+ * back in `In-Reply-To`/`References` is EXACTLY the signed-token id the
+ * engine minted (`mintReplyMessageId`, `src/mail/reply-token.ts`) — not a
+ * provider-generated substitute. So every `EmailSender` implementation MUST
+ * transmit `OutboundEmail.messageId` **verbatim** as the RFC 5322
+ * `Message-ID` header, and MUST NOT generate or overwrite it with a
+ * provider-assigned id. `inReplyTo` and `references`, when present, are
+ * likewise engine-set (specs/mail/sending.md §5: caller-supplied from the
+ * inbound message being answered) and must be transmitted as given, not
+ * reinterpreted. A provider SDK that cannot set `Message-ID` explicitly
+ * (some transactional-email APIs only expose a "reply-to" concept and mint
+ * their own `Message-ID` unconditionally) is unusable for Helpthread and
+ * must not be adapted to this interface — there is no fallback path that
+ * preserves threading correctness.
+ *
+ * `EmailSendResult.providerMessageId` is a SEPARATE, optional field for the
+ * provider's own internal delivery id (e.g. for looking up delivery status
+ * or bounce webhooks in that provider's dashboard/API later). It carries no
+ * threading authority and is never compared against `messageId` — the two
+ * ids serve entirely different purposes and must not be confused.
+ */
+
+/** One fully-formed outbound email, ready to transmit. */
+export interface OutboundEmail {
+  /**
+   * Engine-minted `Message-ID` (the signed reply token, WITH angle
+   * brackets — see `mintReplyMessageId`). The provider MUST send this
+   * verbatim as the RFC 5322 `Message-ID` header; see the module doc.
+   */
+  messageId: string
+
+  /** `In-Reply-To` header value, if this is a reply, verbatim (engine-set — see the module doc). */
+  inReplyTo?: string
+
+  /** `References` header values, verbatim, in the order they should appear (engine-set — see the module doc). */
+  references?: string[]
+
+  from: string
+  to: string[]
+  cc?: string[]
+  subject: string
+
+  /** Plain-text body. At least one of `text`/`html` should be provided. */
+  text?: string
+
+  /** HTML body. At least one of `text`/`html` should be provided. */
+  html?: string
+}
+
+/**
+ * The result of a successful send. `providerMessageId` — the provider's own
+ * internal id for this delivery, if it returns one — carries no threading
+ * authority; see the module doc for why it is kept separate from
+ * `OutboundEmail.messageId`.
+ */
+export interface EmailSendResult {
+  providerMessageId?: string
+}
+
+/**
+ * Provider for transmitting one outbound email. One implementation per
+ * provider (Postmark, SES, Resend, ...). See the module doc for the
+ * `Message-ID` contract every implementation must uphold.
+ */
+export interface EmailSender {
+  /**
+   * 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).
+   */
+  send(email: OutboundEmail): Promise<EmailSendResult>
+}
diff --git a/src/providers/index.ts b/src/providers/index.ts
index d669e17..780c104 100644
--- a/src/providers/index.ts
+++ b/src/providers/index.ts
@@ -5,6 +5,7 @@
  */
 
 export type { BlobStore } from './blob.js'
+export type { EmailSender, EmailSendResult, OutboundEmail } from './email-sender.js'
 export type {
   InboundEmailProvider,
   NormalizedInboundAttachment,
diff --git a/src/store/conversations.test.ts b/src/store/conversations.test.ts
index 9ef820c..03b8a8d 100644
--- a/src/store/conversations.test.ts
+++ b/src/store/conversations.test.ts
@@ -217,4 +217,38 @@ describe('createConversationStore', () => {
     )
     expect(tables).toHaveLength(1)
   })
+
+  it('setThreadDeliveryStatus flips an outbound thread from pending to sent', async () => {
+    const { store } = await freshStore()
+    const { conversationId } = await store.createConversation(newConversation())
+    const appended = await store.appendThread(conversationId, newThread())
+    expect(appended.ok).toBe(true)
+    const threadId = (appended as { threadId: string }).threadId
+
+    // Outbound threads default to 'pending' on insert.
+    let conversation = await store.getConversation(conversationId)
+    expect(conversation?.threads.find((t) => t.id === threadId)?.deliveryStatus).toBe('pending')
+
+    await store.setThreadDeliveryStatus(threadId, 'sent')
+
+    conversation = await store.getConversation(conversationId)
+    expect(conversation?.threads.find((t) => t.id === threadId)?.deliveryStatus).toBe('sent')
+  })
+
+  it('setThreadDeliveryStatus throws for a nonexistent thread id (no silent no-op)', async () => {
+    const { store } = await freshStore()
+    await expect(store.setThreadDeliveryStatus(RANDOM_UUID, 'sent')).rejects.toThrow()
+  })
+
+  it('setThreadDeliveryStatus refuses to mark an INBOUND thread (direction-scoped)', async () => {
+    const { store } = await freshStore()
+    // createConversation's first thread is inbound; its id must not be markable.
+    const { conversationId, threadId } = await store.createConversation(newConversation())
+    const inboundThreadId = threadId
+    await expect(store.setThreadDeliveryStatus(inboundThreadId, 'sent')).rejects.toThrow()
+
+    // And it really wasn't touched.
+    const conversation = await store.getConversation(conversationId)
+    expect(conversation?.threads.find((t) => t.id === inboundThreadId)?.deliveryStatus).toBeNull()
+  })
 })
diff --git a/src/store/conversations.ts b/src/store/conversations.ts
index 0766b12..dcd64c7 100644
--- a/src/store/conversations.ts
+++ b/src/store/conversations.ts
@@ -60,6 +60,16 @@ import type { Db, Queryable } from '../db/client.js'
 
 /** One message to be persisted as a new thread — inbound customer mail, or outbound agent/assistant mail. */
 export interface NewThread {
+  /**
+   * Caller-supplied thread id (a v4 UUID), for outbound threads whose id
+   * must be known BEFORE the row is inserted — the outbound `Message-ID`
+   * embeds a signed token over `{conversationId, threadId}`
+   * (specs/mail/sending.md §2's "id/token knot"), so `mintReplyMessageId`
+   * must run before this insert, not after it. When omitted, the database's
+   * `gen_random_uuid()` default generates the id (the inbound path, which
+   * has no such circularity).
+   */
+  id?: string
   direction: 'inbound' | 'outbound'
   /**
    * The RFC `Message-ID` of this message, verbatim. For an inbound
@@ -75,6 +85,14 @@ export interface NewThread {
   fromAddress: string
   bodyText?: string | null
   bodyHtml?: string | null
+  /**
+   * Outbox status for an OUTBOUND thread: `'pending'` immediately after
+   * mint-and-persist, `'sent'`/`'failed'` once the send attempt resolves
+   * (specs/mail/sending.md §3). Inbound threads leave this `null` (or
+   * omitted) — delivery status is not a meaningful concept for mail we
+   * received, and the column stays `NULL` for those rows.
+   */
+  deliveryStatus?: 'pending' | 'sent' | 'failed' | null
 }
 
 /** Input to {@link ConversationStore.createConversation}: a new conversation plus its first thread. */
@@ -94,6 +112,8 @@ export interface StoredThread {
   fromAddress: string
   bodyText: string | null
   bodyHtml: string | null
+  /** Outbox status — `null` for inbound threads, `'pending'|'sent'|'failed'` for outbound ones. See {@link NewThread.deliveryStatus}. */
+  deliveryStatus: 'pending' | 'sent' | 'failed' | null
   createdAt: Date
 }
 
@@ -148,6 +168,16 @@ export interface ConversationStore {
   getConversation(
     conversationId: string,
   ): Promise<(StoredConversation & { threads: StoredThread[] }) | null>
+
+  /**
+   * Update an outbound thread's outbox status in place (specs/mail/sending.md
+   * §3's persist→send→mark ordering — this is the "mark" step). Callers
+   * (`src/mail/send.ts`) invoke this AFTER the send attempt resolves, moving
+   * a thread from `'pending'` to `'sent'` or `'failed'`. Not transactional
+   * with anything else — this is a single-row status flip by primary key,
+   * and the row was already durably persisted before the send was attempted.
+   */
+  setThreadDeliveryStatus(threadId: string, status: 'pending' | 'sent' | 'failed'): Promise<void>
 }
 
 /** Raw `conversations` row shape, before mapping to {@link StoredConversation}. */
@@ -170,11 +200,12 @@ interface ThreadRow {
   from_address: string
   body_text: string | null
   body_html: string | null
+  delivery_status: string | null
   created_at: Date | string
 }
 
 const THREAD_COLUMNS =
-  'id, conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html, created_at'
+  'id, conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html, delivery_status, created_at'
 
 /**
  * Create a {@link ConversationStore} backed by `db`. Every operation opens
@@ -251,18 +282,76 @@ export function createConversationStore(db: Db): ConversationStore {
         threads: threadRows.map(toStoredThread),
       }
     },
+
+    async setThreadDeliveryStatus(threadId, status) {
+      // Scope to outbound rows and confirm exactly one was updated. Without
+      // `direction = 'outbound'` an inbound thread id could be marked
+      // 'sent'/'failed' (violating the direction↔status invariant the schema
+      // now enforces); without `RETURNING`, a wrong id or a row deleted
+      // between send and mark would no-op silently and let a caller believe a
+      // send was recorded when it wasn't. Both surface loudly instead.
+      const updated = await db.query<{ id: string }>(
+        "UPDATE threads SET delivery_status = $1 WHERE id = $2 AND direction = 'outbound' RETURNING id",
+        [status, threadId],
+      )
+      if (updated.length === 0) {
+        throw new Error(
+          `setThreadDeliveryStatus: no outbound thread with id ${threadId} (wrong id, an inbound thread, or the row was deleted)`,
+        )
+      }
+    },
   }
 }
 
-/** Shared insert used by both `createConversation`'s first thread and `appendThread`. */
+/**
+ * Shared insert used by both `createConversation`'s first thread and
+ * `appendThread`.
+ *
+ * When `thread.id` is supplied (the outbound-send path, specs/mail/sending.md
+ * §2), the `id` column is set explicitly to that caller-generated UUID. When
+ * omitted (the inbound path), the `id` column is left out of the INSERT
+ * entirely so the schema's `gen_random_uuid()` default fires — passing an
+ * explicit `id` in every case would either require the caller to always
+ * generate one (defeating the point of a DB default) or special-case a
+ * `null`/`undefined` id column value, which is not what "no id supplied"
+ * means here.
+ */
 async function insertThread(
   tx: Queryable,
   conversationId: string,
   thread: NewThread,
 ): Promise<string> {
+  // Derive delivery_status from direction so the row always satisfies the
+  // schema's direction↔status CHECK (migration 002): an outbound thread
+  // defaults to 'pending' (its outbox starting state) unless the caller set a
+  // status; an inbound thread is forced to NULL regardless of any status
+  // passed, since delivery status is meaningless for received mail.
+  const deliveryStatus =
+    thread.direction === 'outbound' ? (thread.deliveryStatus ?? 'pending') : null
+
+  if (thread.id !== undefined) {
+    const [row] = await tx.query<{ id: string }>(
+      `INSERT INTO threads (id, conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html, delivery_status)
+       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+       RETURNING id`,
+      [
+        thread.id,
+        conversationId,
+        thread.direction,
+        thread.messageId,
+        thread.inReplyTo ?? null,
+        thread.fromAddress,
+        thread.bodyText ?? null,
+        thread.bodyHtml ?? null,
+        deliveryStatus,
+      ],
+    )
+    return row.id
+  }
+
   const [row] = await tx.query<{ id: string }>(
-    `INSERT INTO threads (conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html)
-     VALUES ($1, $2, $3, $4, $5, $6, $7)
+    `INSERT INTO threads (conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html, delivery_status)
+     VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
      RETURNING id`,
     [
       conversationId,
@@ -272,6 +361,7 @@ async function insertThread(
       thread.fromAddress,
       thread.bodyText ?? null,
       thread.bodyHtml ?? null,
+      deliveryStatus,
     ],
   )
   return row.id
@@ -312,6 +402,7 @@ function toStoredThread(row: ThreadRow): StoredThread {
     fromAddress: row.from_address,
     bodyText: row.body_text,
     bodyHtml: row.body_html,
+    deliveryStatus: row.delivery_status as StoredThread['deliveryStatus'],
     createdAt: toDate(row.created_at),
   }
 }