diff --git a/.testcoverage.yml b/.testcoverage.yml
index 2c0de03a..9baa06de 100644
--- a/.testcoverage.yml
+++ b/.testcoverage.yml
@@ -29,3 +29,7 @@ override:
   # Outbound send on River — the at-least-once send worker + accept-tx enqueue.
   - threshold: 75
     path: ^internal/outboundsend$
+  # Inbound processing on River — the at-least-once worker + reconciler + retention.
+  # (The relay-side ProcessIntake adapter is covered cross-package by relay tests.)
+  - threshold: 65
+    path: ^internal/inboundprocess$
diff --git a/cmd/e2a/main.go b/cmd/e2a/main.go
index 47083aa9..6899d258 100644
--- a/cmd/e2a/main.go
+++ b/cmd/e2a/main.go
@@ -28,6 +28,7 @@ import (
 	"github.com/Mnexa-AI/e2a/internal/limits"
 	"github.com/Mnexa-AI/e2a/internal/oauth"
 	"github.com/Mnexa-AI/e2a/internal/outbound"
+	"github.com/Mnexa-AI/e2a/internal/inboundprocess"
 	"github.com/Mnexa-AI/e2a/internal/outboundsend"
 	"github.com/Mnexa-AI/e2a/internal/relay"
 	"github.com/Mnexa-AI/e2a/internal/senderidentity"
@@ -250,6 +251,18 @@ func main() {
 		registrars = append(registrars, outboundJobs)
 	}
 
+	// Async inbound pipeline (inbound-message-pipeline-river.md), gated by
+	// E2A_INBOUND_MODE=async. The InboundProcessWorker registers on the shared River
+	// client; the SMTP accept-tx enqueues an inbound_process job in the same tx as
+	// the intake row. The Processor (the relay Server) is injected later via
+	// SetProcessor — the relay is built below, after the River client. nil ⇒ the
+	// synchronous inline path (unchanged).
+	var inboundJobs *inboundprocess.Jobs
+	if cfg.Inbound.Mode == "async" {
+		inboundJobs = inboundprocess.NewJobs(store)
+		registrars = append(registrars, inboundJobs)
+	}
+
 	var senderMgr *senderidentity.Manager
 	if region := cfg.SenderIdentity.SESRegion; region != "" {
 		provider, perr := senderidentity.NewSESProviderFromConfig(ctx, region)
@@ -515,6 +528,27 @@ func main() {
 	smtpServer.SetEnforcer(enforcer)
 	smtpServer.SetOutbox(webhookOutbox)
 
+	// Wire the async inbound pipeline into the relay (E2A_INBOUND_MODE=async): the
+	// Processor is the relay Server itself; the accept-tx enqueues via the shared
+	// client. Done here because the relay is constructed after the River client — the
+	// worker late-binds the Processor (SetProcessor) and tolerates the brief startup
+	// window before it's set.
+	if inboundJobs != nil {
+		inboundJobs.SetProcessor(smtpServer)
+		inboundJobs.SetEnqueuer(jobsClient)
+		smtpServer.SetInboundEnqueuer(inboundJobs)
+		// Cutover: enqueue any accepted-but-unenqueued intake rows (pre-async rows at
+		// the mode-flip, or rows stranded by a crash between insert and enqueue).
+		// Idempotent (process_job_id IS NULL guard). Runs after SetProcessor so the
+		// re-driven jobs find a wired processor.
+		if n, cerr := inboundJobs.ReconcilePending(ctx, pool); cerr != nil {
+			log.Printf("[inbound-process] cutover: %v", cerr)
+		} else if n > 0 {
+			log.Printf("[inbound-process] cutover enqueued %d stranded intakes", n)
+		}
+		log.Printf("[inbound-process] engine=river (async accept, E2A_INBOUND_MODE=async)")
+	}
+
 	// Graceful shutdown
 	sigCh := make(chan os.Signal, 1)
 	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
diff --git a/docs/design/inbound-message-pipeline-river.md b/docs/design/inbound-message-pipeline-river.md
new file mode 100644
index 00000000..81661149
--- /dev/null
+++ b/docs/design/inbound-message-pipeline-river.md
@@ -0,0 +1,250 @@
+# Inbound message pipeline → persist-first on River
+
+Status: **DRAFT for review** (2026-07-07). Author: design pass off the as-built trace.
+Companion to `async-message-pipeline.md` (outbound) and
+`webhook-delivery-river-migration.md`. This is the **inbound** counterpart: push the
+durability boundary to the SMTP edge, the way outbound pushed it to the API edge.
+
+---
+
+## 1. Problem statement
+
+Today the inbound path does **all** processing inline in the SMTP session — MIME
+parse, SPF/DKIM, sender gate, content screening (incl. an up-to-10s Gemini call),
+HMAC signing, conversation lookup, and the `messages` insert — and only then returns
+`250 OK`. Two consequences:
+
+1. **Silent loss.** `deliverToAgent` swallows a persist error (logs + returns), and
+   `deliverMessages` counts it delivered anyway, so a failed insert still returns
+   **`250 OK`** (`internal/relay/server.go:522`). The sending MTA treats `250` as
+   done and never retries → the message is **lost**. `docs/events.md` advertises
+   `email.received` as "at-least-once end-to-end," which is **false** at this edge.
+2. **LLM latency on the SMTP critical path.** A multi-second Gemini screen inside the
+   session risks the sender's SMTP timeout and couples receipt to an external API.
+
+There is also **no inbound dedup**: the message id is random per DATA call, so an MTA
+retry after a slow/failed `250` produces a **second `messages` row + second
+`email.received`**, delivered twice.
+
+Outbound already solved the symmetric problem (persist-first accept-tx → River
+`QueueOutbound`, #388). Inbound should mirror it.
+
+## 2. Goals and non-goals
+
+**Goals**
+- **At-least-once from the SMTP edge inward.** A `250 OK` means the message is
+  durably accepted; anything we can't accept returns a `4xx` so the sender retries.
+- **Move parse / SPF-DKIM / screening / persist / deliver OFF the SMTP session** into
+  a River worker on a new `QueueInbound`.
+- **Best-effort dedup** of MTA retries (collapse the lost-ack duplicate).
+- **Preserve** all existing behavior: RCPT-time hard rejects (550/552), SPF/DKIM
+  semantics (against the connecting IP + envelope), inbound screening/HITL
+  (flag/review/block), conversation threading, `email.received`/flagged/blocked/
+  pending_review events, WS live-tail, per-recipient fan-out.
+- **Flag-gated cutover** (`E2A_INBOUND_MODE`, default `sync`) with the sync path
+  byte-for-byte unchanged; a startup + live reconciler for stranded intake.
+
+**Non-goals**
+- Changing the SPF/DKIM/screening *logic* (only *where* it runs).
+- A raw-object/S3 store — raw MIME lives in the intake row (bytea), as today it lives
+  in `messages.raw_message`.
+- Exactly-once (impossible over SMTP); dedup is a quality property, not a correctness
+  invariant.
+- The outbound/webhook lanes (already migrated).
+
+## 3. Relevant context and constraints
+
+- e2a's relay is the **direct MX** (`docs/deployment.md:42`), `emersion/go-smtp`. The
+  retrying MTA is the *sender's*, not SES. `4xx` → sender queues + retries ~4–5 days;
+  `250` → done, no retry; `5xx` → bounce to author.
+- **RCPT-time gating stays synchronous and pre-DATA** — unknown/unverified recipient
+  → `550`, owner over cap → `552` (`server.go:194-220`). We must keep rejecting bad
+  recipients *before* accepting a body; only post-DATA processing moves async.
+- **SPF needs the connecting IP**, available only in-session. The intake row MUST
+  capture `remote_ip` + `envelope_from` so the worker can run `emailauth.Check`
+  later. (`auth_verdict` is already stored for the same reason.)
+- Delivery fan-out is **already** decoupled (outbox → `OutboxWorker` → River
+  `webhookdelivery`); we reuse it verbatim — the worker's `messages`-tx publishes
+  `email.received` to the same outbox.
+- Shared River foundation exists (`internal/jobs` Registrar/Enqueuer, named queues).
+  Add `QueueInbound`.
+- Per-recipient model: today one `messages` row per resolved RCPT TO.
+
+## 4. Proposed design
+
+### 4.1 Three-layer model (consistent with webhook/outbound)
+
+- **L1 — `inbound_intake`** (NEW): the immutable durable fact — "these raw bytes
+  arrived for this recipient from this IP." Written in the SMTP session; the only
+  thing that gates `250`.
+- **L2 — `messages`** (unchanged shape): the processed result, written by the worker.
+- **L3 — `river_job` on `QueueInbound`**: the execution (parse/screen/persist/deliver).
+
+### 4.2 SMTP session (pre-`250`) — shrunk to a durable accept
+
+RCPT gating unchanged. In `session.Data`, when `E2A_INBOUND_MODE=async`:
+
+```
+read body (already bounded 10MB)
+content_hash = sha256(raw)
+ONE TX:
+  for each resolved recipient:
+     INSERT INTO inbound_intake (id, recipient, envelope_from, remote_ip,
+        raw_message, message_id, content_hash, status='accepted', created_at)
+     ON CONFLICT (recipient, message_id, content_hash) DO NOTHING
+     RETURNING id
+     if a row was inserted:            -- not a duplicate
+        job_id = EnqueueInboundProcessTx(tx, intake_id)   -- River InsertTx on QueueInbound
+        UPDATE inbound_intake SET process_job_id=job_id WHERE id=intake_id
+COMMIT
+if commit ok  -> 250 OK            (durably accepted, incl. idempotent duplicate)
+if commit err -> 451 4.3.0         (transient — sender retries)
+```
+
+- **Dedup:** `UNIQUE (recipient, message_id, content_hash)`. A duplicate (lost-ack
+  retry) hits `ON CONFLICT DO NOTHING`, enqueues nothing, and still returns `250` —
+  idempotent accept. If `message_id` is empty (sender omitted it), the hash + envelope
+  still key it; two genuinely-distinct bodies won't collapse.
+- **Atomicity:** intake insert + River enqueue + job-id stamp in one tx (River
+  `InsertTx` on the same `pgx.Tx`), exactly like the outbound accept-tx. A committed
+  `accepted` intake row always has a job.
+- **`451` correctness rule:** return `4xx` **only** when the tx did not commit. A
+  failure *after* commit but before `250` must still `250` (else double-accept — the
+  dedup key would collapse it anyway, but the rule keeps intent clean).
+
+### 4.3 River worker (post-`250`) — `internal/inboundprocess`
+
+`InboundProcessWorker.Work(intake_id)` — one attempt per job attempt, River owns retry:
+
+```
+load intake row; nil -> no-op (pruned)
+if intake.status='processed' -> no-op         (idempotent re-drive)
+parse MIME (net/mail)                          — off the SMTP path now
+emailauth.Check(intake.remote_ip, envelope_from, raw)
+HMAC sign (headers.Signer)
+sender gate (inboundpolicy.EvaluateIngestion)
+content screen (piguard, incl. Gemini)         — off the SMTP path now
+conversation resolve
+ONE TX:
+  CreateInboundMessageInTx(...)   -> messages row (status per screen: sent/pending_review/review_rejected)
+  outbox.PublishTx(email.received | flagged | blocked | pending_review)  (suppressed when held)
+  UPDATE inbound_intake SET status='processed', message_fk=<msg id> WHERE id=intake_id
+COMMIT
+best-effort: writeProtectionEvents; WS hub.Send (when !Hold)
+```
+
+- **Idempotency:** the worker's terminal tx flips `intake.status='processed'` **in the
+  same tx** as the `messages` insert. A re-drive sees `processed` → no-op. So a crash
+  after commit never double-creates.
+- **Screening/HITL:** identical logic (`screenInbound`), just relocated. Held →
+  `messages` row with hold status + `approval_expires_at`, `email.received`
+  suppressed, only `email.pending_review`/`blocked` emitted, WS skipped. Unchanged
+  externally.
+- **Delivery:** the `email.received` outbox row drives the existing
+  `OutboxWorker`→River webhook path with no change. WS stays best-effort.
+- **Error classification** (mirror outbound §8): a transient parse/DB error → River
+  retry (bounded envelope). A permanently-unparseable body → terminal (mark intake
+  `failed`, emit nothing / an `email.rejected`? — see open questions). No provider
+  outage concept here (no upstream call except Gemini, which fails *open* to allow).
+
+### 4.4 Reconciler + retention
+
+- **Startup cutover reconciler:** on boot in async mode, enqueue a job for every
+  `inbound_intake` with `status='accepted' AND process_job_id IS NULL` (mirror
+  `outboundsend.ReconcilePending`). Handles rows accepted by a build that crashed
+  pre-enqueue and the mode-flip moment.
+- **Live periodic reconciler** (`QueueMaintenance`) — **DEFERRED (follow-up, matching
+  outboundsend)**: would re-enqueue `accepted` intake whose job is terminal/absent,
+  closing the rare "job discarded without processing" strand. Not shipped: the
+  accept-tx is atomic so the NULL-job set is ~empty in steady state, and the startup
+  pass re-drives on the next deploy. Only the startup cutover ships.
+- **Retention:** prune `status='processed'` intake older than N days (raw is also in
+  `messages.raw_message`). A periodic sweep; keep a short window (e.g. 3 days) for
+  debugging + re-drive. `failed` intake retained longer for inspection.
+
+### 4.5 Cutover
+
+`E2A_INBOUND_MODE` (`config.InboundConfig`, env override), default **`sync`**. The
+relay's `Data` branches on it: `async` → §4.2; `sync` → the existing inline path,
+untouched. Enable per-deploy in e2a-ops after canary. No wire/API change — inbound
+processing is server-internal, so no spec/SDK regen.
+
+## 5. Edge cases and failure handling
+
+- **Transient accept failure** → `451`, sender retries (§4.2). No loss.
+- **Lost `250` ack** (committed, ack dropped) → sender retry → dedup `ON CONFLICT` →
+  idempotent `250`, no second job. No duplicate.
+- **Worker crash mid-processing** → River re-drives; `processed` guard makes it
+  idempotent; before the terminal tx nothing is externally visible, so re-run is safe.
+- **Unparseable / malformed MIME** → terminal fail (don't infinite-retry a body that
+  will never parse); mark intake `failed`. *Open q: notify anyone?*
+- **Gemini timeout/outage** → screen fails **open** (allow + deliver), as today — an
+  LLM outage must not block mail. Now it also doesn't block the SMTP session.
+- **Multi-recipient** → N intake rows + N jobs in one tx; partial isn't possible (one
+  tx). Each recipient dedups independently.
+- **Held message** → worker writes hold-status row, suppresses `email.received`; a
+  re-drive is idempotent via `processed`.
+- **Agent deleted between accept and processing** → worker load resolves no agent →
+  terminal no-op (message dropped; the recipient is gone). Matches today's cascade.
+- **Intake grows unbounded** → retention sweep (§4.4).
+- **Clock/`remote_ip` capture** → store `remote_ip` as text at accept; SPF uses it
+  verbatim. IPv6 + proxied (PROXY protocol / XCLIENT) — capture the same source the
+  sync path uses today (verify in impl).
+
+## 6. Scalability and extensibility notes
+
+- Decoupling the Gemini call from the session removes the dominant SMTP-latency /
+  timeout risk and lets screening scale independently (River concurrency cap), not
+  bounded by SMTP connection slots.
+- `QueueInbound` is independently pausable/tunable (`river.QueuePause`,
+  `MaxWorkers`) — an inbound spike can't starve outbound/webhook and vice-versa.
+- The intake table is the natural home for future needs: raw re-processing (re-run
+  screening after a model upgrade), inbound replay/redelivery API (symmetric to
+  webhook redelivery), and a spam/greylisting layer at accept time.
+- Extensible dedup: the key can gain a scope column without a rewrite.
+
+## 7. Verification strategy
+
+- **Unit:** dedup key (same `message_id`+recipient+hash → one row, one job); the
+  `451`-on-tx-failure vs `250`-on-commit rule; worker idempotency (re-drive of a
+  `processed` intake = no-op); screen relocation preserves hold/deliver outcomes;
+  accept-tx atomicity (intake + job together or not at all).
+- **Coverage floor** for `internal/inboundprocess` (mirror `outboundsend` @75).
+- **Local e2e (over real SMTP, the required gate):** send via SMTP → `250` → worker →
+  `messages` row + `email.received` at a webhook capture; a held-scan case →
+  `pending_review` + suppressed `received`; a **transient-failure injection** (point
+  at a broken DB / force the intake insert to fail) → **`451`, no message**, then
+  recovery on retry; a **duplicate send** (same `Message-ID` twice) → one message;
+  sync-mode (flag off) unchanged; reconciler cutover (plant an `accepted` intake, no
+  job → boot → processed).
+- **Regression:** the swallowed-`250`-on-persist-failure path must now be a `451`.
+- **Independent + adversarial review** on the pushed branch, as with #388.
+
+## 8. Open questions
+
+1. **Intake table vs minimal-`messages`-row-first.** — **DECIDED (2026-07-07):
+   separate `inbound_intake` L1 table.** Rationale: inbound cannot produce a complete
+   `messages` row at accept (subject/sender/auth/conversation all come out of the
+   worker's parse+screen), so a minimal `messages` row would pollute the product's
+   central read model with genuinely-incomplete `receiving` rows and push an
+   exclude-incomplete obligation onto every reader (API list/detail, web inbox,
+   threading, review queue, exports) — a silent-leak failure mode. Unlike outbound,
+   whose `accepted` row is *complete*-but-undelivered (so persist-first into
+   `messages` was fine there), inbound must accept bytes before it knows what they
+   are. The separate table's costs (one migration, a bounded raw-duplication window
+   erased by pruning) are contained and mechanical.
+2. **Terminal-failure UX.** When a body is permanently unparseable or the worker
+   exhausts retries, do we (a) silently drop + mark intake `failed`, (b) emit a new
+   `email.rejected` webhook event, or (c) attempt an SMTP-time reject? Since we've
+   already `250`'d, (c) is out; leaning (a) for v1 with a `failed` intake for ops
+   visibility, (b) as a follow-up. Agree?
+3. **Dedup window.** Unbounded (unique index forever) vs scoped to a retention window
+   (a very late resend after prune would re-deliver). Leaning **unique index on live
+   intake + prune after 3d** — a >3d resend is effectively a new message. OK?
+4. **`remote_ip` under proxy.** Confirm the sync path's IP source (direct vs PROXY
+   protocol / XCLIENT) so the intake captures the same value SPF uses today.
+5. **Scope of slice 1.** Recommend slice 1 = the engine-agnostic honesty fix
+   (`deliverToAgent` propagates the persist error → `Data` returns `451`), shippable
+   immediately and independent of the queue — then slices 2+ build the intake table,
+   worker, cutover. Agree?
+```
diff --git a/internal/config/config.go b/internal/config/config.go
index 92497b71..2f2e7238 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -41,6 +41,7 @@ type Config struct {
 	Signing          SigningConfig          `yaml:"signing"`
 	OutboundSMTP     OutboundSMTPConfig     `yaml:"outbound_smtp"`
 	Outbound         OutboundConfig         `yaml:"outbound"`
+	Inbound          InboundConfig          `yaml:"inbound"`
 	SenderIdentity   SenderIdentityConfig   `yaml:"sender_identity"`
 	DeliveryFeedback DeliveryFeedbackConfig `yaml:"delivery_feedback"`
 	Limits           LimitsConfig           `yaml:"limits"`
@@ -135,6 +136,18 @@ type OutboundConfig struct {
 	Mode string `yaml:"mode"`
 }
 
+// InboundConfig selects the inbound processing model (inbound-message-pipeline-
+// river.md). Mode="sync" (the default) is the historical path: the SMTP session runs
+// parse/screen/persist/deliver inline before 250. Mode="async" opts into the
+// queue-first River pipeline: the session durably accepts the raw MIME to
+// inbound_intake + enqueues a processing job atomically before 250, and the
+// internal/inboundprocess worker does the processing off the SMTP critical path.
+// Override with E2A_INBOUND_MODE. Any value other than "async" is treated as "sync"
+// (fail-safe to the unchanged path).
+type InboundConfig struct {
+	Mode string `yaml:"mode"`
+}
+
 // DeliveryFeedbackConfig controls outbound delivery feedback (decision 9 /
 // Slice 4b). When SESConfigurationSet is set, outbound mail is tagged so SES
 // publishes delivery/bounce/complaint events; SNSTopicARNs is the fail-closed
@@ -221,6 +234,7 @@ func Load(path string) (*Config, error) {
 			CacheTTLSeconds:  60,
 		},
 		Outbound: OutboundConfig{Mode: "sync"},
+		Inbound:  InboundConfig{Mode: "sync"},
 		Env:      "development",
 	}
 
@@ -273,6 +287,9 @@ func Load(path string) (*Config, error) {
 	if v := os.Getenv("E2A_OUTBOUND_MODE"); v != "" {
 		cfg.Outbound.Mode = v
 	}
+	if v := os.Getenv("E2A_INBOUND_MODE"); v != "" {
+		cfg.Inbound.Mode = v
+	}
 	if v := os.Getenv("E2A_OUTBOUND_SMTP_REQUIRE_TLS"); v != "" {
 		if b, err := strconv.ParseBool(v); err == nil {
 			cfg.OutboundSMTP.RequireTLS = &b
diff --git a/internal/identity/inbound_intake_store.go b/internal/identity/inbound_intake_store.go
new file mode 100644
index 00000000..0341d02e
--- /dev/null
+++ b/internal/identity/inbound_intake_store.go
@@ -0,0 +1,148 @@
+package identity
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// --- Inbound intake (queue-first inbound pipeline, Layer 1) ---
+//
+// inbound_intake is the durable landing pad the SMTP session writes before 250 (see
+// inbound-message-pipeline-river.md). The internal/inboundprocess River worker reads
+// a row by id, parses/screens/persists a messages row, and marks the intake
+// processed. These methods are the store surface for both sides; the worker's Store
+// adapter (in internal/agent) bridges onto them.
+
+// IntakeStatus values (mirrors migration 056's CHECK).
+const (
+	IntakeStatusAccepted  = "accepted"
+	IntakeStatusProcessed = "processed"
+	IntakeStatusFailed    = "failed"
+)
+
+// NewInboundIntakeID mints an intake row id.
+func NewInboundIntakeID() string {
+	return "intk_" + generateID()
+}
+
+// InboundIntake is the worker's view of an accepted inbound message — the raw MIME
+// plus the connection facts (envelope + remote IP) it needs to run SPF/DKIM and
+// screening, which cannot be recomputed outside the SMTP session.
+type InboundIntake struct {
+	ID           string
+	Recipient    string
+	EnvelopeFrom string
+	RemoteIP     string
+	Raw          []byte
+	MessageID    string // sender's RFC 5322 Message-ID
+	ContentHash  string
+	Status       string
+	CreatedAt    time.Time
+}
+
+// InsertInboundIntakeTx writes an accepted intake row inside the accept-tx. It
+// returns inserted=false (no error) when the row is a duplicate — the dedup unique
+// index (recipient, message_id, content_hash) suppressed it — so the caller knows
+// NOT to enqueue a second job and to still answer 250 (idempotent accept).
+func (s *Store) InsertInboundIntakeTx(ctx context.Context, tx pgx.Tx, id, recipient, envelopeFrom, remoteIP, messageID, contentHash string, raw []byte) (inserted bool, err error) {
+	var returnedID string
+	err = tx.QueryRow(ctx,
+		`INSERT INTO inbound_intake (id, recipient, envelope_from, remote_ip, raw_message, message_id, content_hash, status)
+		 VALUES ($1, $2, $3, $4, $5, $6, $7, 'accepted')
+		 ON CONFLICT (recipient, message_id, content_hash) DO NOTHING
+		 RETURNING id`,
+		id, recipient, envelopeFrom, remoteIP, raw, messageID, contentHash,
+	).Scan(&returnedID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return false, nil // duplicate — dedup index suppressed the insert
+	}
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+// StampInboundIntakeJobIDTx records the River job id on the intake row, in the same
+// accept-tx as the insert + enqueue (so a committed accepted row always has its job).
+func (s *Store) StampInboundIntakeJobIDTx(ctx context.Context, tx pgx.Tx, intakeID string, jobID int64) error {
+	_, err := tx.Exec(ctx, `UPDATE inbound_intake SET process_job_id = $2 WHERE id = $1`, intakeID, jobID)
+	return err
+}
+
+// LoadInboundIntake reads an intake row for the worker. Returns (nil, nil) when the
+// row is gone (pruned) so the worker treats it as a no-op.
+func (s *Store) LoadInboundIntake(ctx context.Context, intakeID string) (*InboundIntake, error) {
+	var it InboundIntake
+	err := s.pool.QueryRow(ctx,
+		`SELECT id, recipient, envelope_from, remote_ip, raw_message, message_id, content_hash, status, created_at
+		   FROM inbound_intake WHERE id = $1`,
+		intakeID,
+	).Scan(&it.ID, &it.Recipient, &it.EnvelopeFrom, &it.RemoteIP, &it.Raw, &it.MessageID, &it.ContentHash, &it.Status, &it.CreatedAt)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &it, nil
+}
+
+// MarkInboundIntakeProcessedTx flips the intake to 'processed' and links the created
+// messages row, in the worker's terminal tx (same tx as the messages insert + outbox
+// publish). This is the worker's idempotency gate: a re-drive that finds 'processed'
+// no-ops instead of re-creating the message.
+func (s *Store) MarkInboundIntakeProcessedTx(ctx context.Context, tx pgx.Tx, intakeID, messageFK string) error {
+	// Guard on status='accepted' (defense-in-depth): the worker's Work() already
+	// no-ops a non-accepted row before processing, but scoping the flip here means a
+	// second concurrent processor of the same intake commits nothing rather than
+	// re-linking a different messages row.
+	tag, err := tx.Exec(ctx,
+		`UPDATE inbound_intake SET status = 'processed', message_fk = $2, processed_at = now()
+		  WHERE id = $1 AND status = 'accepted'`,
+		intakeID, messageFK)
+	if err != nil {
+		return err
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrIntakeAlreadyProcessed
+	}
+	return nil
+}
+
+// ErrIntakeAlreadyProcessed signals that the intake was not in 'accepted' state when
+// the worker tried to flip it — another attempt already processed it. The caller
+// rolls back its persist tx (no duplicate message/event) and treats it as done.
+var ErrIntakeAlreadyProcessed = errors.New("inbound intake already processed")
+
+// ErrRecipientGone signals that the recipient no longer resolves to an agent (deleted
+// between accept and processing). It is NOT a transient error — the async worker
+// marks the intake terminally (so it doesn't linger 'accepted' forever) and the sync
+// path skips the recipient. Distinct sentinel so callers don't retry.
+var ErrRecipientGone = errors.New("recipient no longer resolves to an agent")
+
+// MarkInboundIntakeFailed marks an intake terminally failed (unparseable body /
+// exhausted retries) with a diagnostic detail. Own transaction — a terminal record
+// for ops visibility; the message is dropped (we already 250'd).
+func (s *Store) MarkInboundIntakeFailed(ctx context.Context, intakeID, detail string) error {
+	_, err := s.pool.Exec(ctx,
+		`UPDATE inbound_intake SET status = 'failed', detail = $2, processed_at = now() WHERE id = $1`,
+		intakeID, detail)
+	return err
+}
+
+// PruneProcessedIntake deletes processed intake rows whose terminal time is older
+// than olderThan. Safe because the raw MIME also lives in messages.raw_message once
+// processed; failed rows are deliberately retained for inspection. Returns the count
+// pruned. Called by the inbound retention periodic (QueueMaintenance).
+func (s *Store) PruneProcessedIntake(ctx context.Context, olderThan time.Duration) (int64, error) {
+	cutoff := time.Now().Add(-olderThan)
+	tag, err := s.pool.Exec(ctx,
+		`DELETE FROM inbound_intake WHERE status = 'processed' AND processed_at < $1`, cutoff)
+	if err != nil {
+		return 0, err
+	}
+	return tag.RowsAffected(), nil
+}
diff --git a/internal/identity/inbound_intake_store_test.go b/internal/identity/inbound_intake_store_test.go
new file mode 100644
index 00000000..d587c6c9
--- /dev/null
+++ b/internal/identity/inbound_intake_store_test.go
@@ -0,0 +1,121 @@
+package identity_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/Mnexa-AI/e2a/internal/identity"
+	"github.com/Mnexa-AI/e2a/internal/testutil"
+	"github.com/jackc/pgx/v5"
+)
+
+// TestInboundIntake_InsertLoadDedup covers the accept-side surface: insert reports
+// whether a row landed, the dedup key (recipient, message_id, content_hash)
+// suppresses a re-send (inserted=false, no error), and load round-trips the raw +
+// facts / returns (nil,nil) when gone.
+func TestInboundIntake_InsertLoadDedup(t *testing.T) {
+	pool := testutil.TestDB(t)
+	store := identity.NewStore(pool)
+	ctx := context.Background()
+
+	raw := []byte("From: a@b.test\r\nTo: bot@x.test\r\nSubject: hi\r\n\r\nbody")
+	insert := func(iid, recipient, msgID, hash string) bool {
+		var inserted bool
+		if err := store.WithTx(ctx, func(tx pgx.Tx) error {
+			var e error
+			inserted, e = store.InsertInboundIntakeTx(ctx, tx, iid, recipient, "a@b.test", "1.2.3.4", msgID, hash, raw)
+			return e
+		}); err != nil {
+			t.Fatalf("insert: %v", err)
+		}
+		return inserted
+	}
+
+	first := identity.NewInboundIntakeID()
+	if !insert(first, "bot@x.test", "<m1@b.test>", "hash1") {
+		t.Fatal("first insert should report inserted=true")
+	}
+	if insert(identity.NewInboundIntakeID(), "bot@x.test", "<m1@b.test>", "hash1") {
+		t.Fatal("duplicate (same recipient+message_id+content_hash) must report inserted=false")
+	}
+	if !insert(identity.NewInboundIntakeID(), "bot@x.test", "<m1@b.test>", "hash2") {
+		t.Fatal("different content_hash must insert")
+	}
+	if !insert(identity.NewInboundIntakeID(), "other@x.test", "<m1@b.test>", "hash1") {
+		t.Fatal("different recipient must insert")
+	}
+
+	it, err := store.LoadInboundIntake(ctx, first)
+	if err != nil || it == nil {
+		t.Fatalf("load: err=%v it=%v", err, it)
+	}
+	if it.Recipient != "bot@x.test" || it.MessageID != "<m1@b.test>" || it.ContentHash != "hash1" ||
+		string(it.Raw) != string(raw) || it.Status != identity.IntakeStatusAccepted || it.RemoteIP != "1.2.3.4" {
+		t.Fatalf("loaded intake mismatch: %+v", it)
+	}
+	miss, err := store.LoadInboundIntake(ctx, "intk_missing")
+	if err != nil || miss != nil {
+		t.Fatalf("missing load should be (nil,nil), got it=%v err=%v", miss, err)
+	}
+}
+
+// TestInboundIntake_StampProcessAndFail covers the worker-side terminal transitions:
+// stamp the job id in the accept-tx, flip to processed + link the messages row, and
+// the failed terminal path with a diagnostic detail.
+func TestInboundIntake_StampProcessAndFail(t *testing.T) {
+	pool := testutil.TestDB(t)
+	store := identity.NewStore(pool)
+	ctx := context.Background()
+
+	id := identity.NewInboundIntakeID()
+	if err := store.WithTx(ctx, func(tx pgx.Tx) error {
+		_, e := store.InsertInboundIntakeTx(ctx, tx, id, "bot@x.test", "a@b.test", "1.2.3.4", "<m@b.test>", "h", []byte("raw"))
+		return e
+	}); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+
+	if err := store.WithTx(ctx, func(tx pgx.Tx) error {
+		return store.StampInboundIntakeJobIDTx(ctx, tx, id, 4242)
+	}); err != nil {
+		t.Fatalf("stamp: %v", err)
+	}
+	var jobID int64
+	if err := pool.QueryRow(ctx, `SELECT process_job_id FROM inbound_intake WHERE id=$1`, id).Scan(&jobID); err != nil {
+		t.Fatalf("read job id: %v", err)
+	}
+	if jobID != 4242 {
+		t.Fatalf("job id = %d, want 4242", jobID)
+	}
+
+	if err := store.WithTx(ctx, func(tx pgx.Tx) error {
+		return store.MarkInboundIntakeProcessedTx(ctx, tx, id, "msg_result123")
+	}); err != nil {
+		t.Fatalf("mark processed: %v", err)
+	}
+	var status, fk string
+	if err := pool.QueryRow(ctx, `SELECT status, message_fk FROM inbound_intake WHERE id=$1`, id).Scan(&status, &fk); err != nil {
+		t.Fatalf("read processed: %v", err)
+	}
+	if status != identity.IntakeStatusProcessed || fk != "msg_result123" {
+		t.Fatalf("after process: status=%q fk=%q", status, fk)
+	}
+
+	id2 := identity.NewInboundIntakeID()
+	if err := store.WithTx(ctx, func(tx pgx.Tx) error {
+		_, e := store.InsertInboundIntakeTx(ctx, tx, id2, "bot@x.test", "a@b.test", "1.2.3.4", "<m2@b.test>", "h2", []byte("raw2"))
+		return e
+	}); err != nil {
+		t.Fatalf("insert 2: %v", err)
+	}
+	if err := store.MarkInboundIntakeFailed(ctx, id2, "unparseable MIME"); err != nil {
+		t.Fatalf("mark failed: %v", err)
+	}
+	var s2, d2 string
+	if err := pool.QueryRow(ctx, `SELECT status, detail FROM inbound_intake WHERE id=$1`, id2).Scan(&s2, &d2); err != nil {
+		t.Fatalf("read failed: %v", err)
+	}
+	if s2 != identity.IntakeStatusFailed || d2 != "unparseable MIME" {
+		t.Fatalf("after fail: status=%q detail=%q", s2, d2)
+	}
+}
diff --git a/internal/inboundprocess/jobs.go b/internal/inboundprocess/jobs.go
new file mode 100644
index 00000000..820fc012
--- /dev/null
+++ b/internal/inboundprocess/jobs.go
@@ -0,0 +1,162 @@
+package inboundprocess
+
+import (
+	"context"
+	"errors"
+	"log"
+	"sync"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/riverqueue/river"
+
+	"github.com/Mnexa-AI/e2a/internal/identity"
+	"github.com/Mnexa-AI/e2a/internal/jobs"
+)
+
+// reconcileBatch bounds one cutover scan of stranded accepted intake rows — mirrors
+// outboundsend.reconcileBatch. In steady state the stranded set is ~empty (the
+// accept-tx stamps process_job_id atomically); this caps how many rows one pass
+// re-drives if a systemic enqueue failure ever left a backlog.
+const reconcileBatch = 1000
+
+// Jobs is the inbound-processing integration on the shared River client: a
+// jobs.Registrar (contributes InboundProcessWorker) plus the transactional enqueue
+// entry point the SMTP accept-tx calls. The shared client + the Processor are both
+// injected AFTER construction (two-phase wiring) — the client via SetEnqueuer, the
+// Processor (the relay Server) via SetProcessor, because the relay is built after
+// jobs.New. Jobs itself is the worker's Processor, late-binding to the concrete one.
+type Jobs struct {
+	store Store
+	enq   jobs.Enqueuer
+
+	mu        sync.RWMutex
+	processor Processor
+}
+
+// NewJobs builds the integration with just its store (no client, no processor yet).
+func NewJobs(store Store) *Jobs {
+	return &Jobs{store: store}
+}
+
+// SetEnqueuer injects the shared client so EnqueueInboundProcessTx can insert jobs.
+func (j *Jobs) SetEnqueuer(e jobs.Enqueuer) { j.enq = e }
+
+// SetProcessor injects the concrete Processor (the relay Server), built after
+// jobs.New. Guarded so the River worker goroutines read it race-free.
+func (j *Jobs) SetProcessor(p Processor) {
+	j.mu.Lock()
+	j.processor = p
+	j.mu.Unlock()
+}
+
+// ProcessIntake makes Jobs itself the worker's Processor, delegating to the concrete
+// one set via SetProcessor. Until that is wired (the brief startup window before the
+// relay is built) it returns a retryable error, so a pending job simply retries
+// rather than panicking on a nil processor.
+func (j *Jobs) ProcessIntake(ctx context.Context, it *identity.InboundIntake) error {
+	j.mu.RLock()
+	p := j.processor
+	j.mu.RUnlock()
+	if p == nil {
+		return errors.New("inbound processor not wired yet — retrying")
+	}
+	return p.ProcessIntake(ctx, it)
+}
+
+// RegisterJobs adds the InboundProcessWorker (with Jobs as the late-binding
+// Processor) + the RetentionWorker, and schedules the retention sweep as a periodic
+// on QueueMaintenance. Implements jobs.Registrar.
+func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob {
+	river.AddWorker(w, NewInboundProcessWorker(j.store, j))
+	river.AddWorker(w, &RetentionWorker{pruner: j.store})
+	return []*river.PeriodicJob{
+		river.NewPeriodicJob(
+			river.PeriodicInterval(retentionInterval),
+			func() (river.JobArgs, *river.InsertOpts) {
+				return InboundRetentionArgs{}, &river.InsertOpts{Queue: jobs.QueueMaintenance}
+			},
+			&river.PeriodicJobOpts{RunOnStart: false},
+		),
+	}
+}
+
+// ReconcilePending enqueues an inbound_process job for every accepted intake row that
+// has no job yet (process_job_id IS NULL). Run ONCE at startup as the cutover.
+//
+// Because the accept-tx is a single transaction (intake insert + job enqueue + job-id
+// stamp commit together), a committed accepted row in steady state ALWAYS has its
+// job — so this set is normally empty. It exists to enqueue (a) any accepted rows at
+// the moment the mode is first flipped on, and (b) rows stranded by a crash between
+// insert and enqueue. Idempotent: the per-row FOR UPDATE + process_job_id IS NULL
+// guard means a re-run (or a concurrent replica) never double-enqueues.
+//
+// NOTE (follow-up, matching outboundsend): no live periodic reconciler yet. The one
+// residual it would close is an intake left accepted-with-a-terminal/dead-job if the
+// worker's MarkInboundIntakeFailed somehow never lands — rare, and the startup pass
+// re-drives NULL-job rows on the next deploy.
+func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error) {
+	rows, err := pool.Query(ctx,
+		`SELECT id FROM inbound_intake
+		  WHERE status='accepted' AND process_job_id IS NULL
+		  LIMIT $1`, reconcileBatch)
+	if err != nil {
+		return 0, err
+	}
+	var ids []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			rows.Close()
+			return 0, err
+		}
+		ids = append(ids, id)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return 0, err
+	}
+
+	n := 0
+	for _, id := range ids {
+		if err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
+			// Re-check under a row lock: another process (or a prior run) may have
+			// enqueued it already. Skip if process_job_id is now set.
+			var jobID *int64
+			if err := tx.QueryRow(ctx,
+				`SELECT process_job_id FROM inbound_intake WHERE id=$1 FOR UPDATE`, id,
+			).Scan(&jobID); err != nil {
+				return err
+			}
+			if jobID != nil {
+				return nil // already enqueued
+			}
+			newJobID, err := j.EnqueueInboundProcessTx(ctx, tx, id)
+			if err != nil {
+				return err
+			}
+			if err := j.store.StampInboundIntakeJobIDTx(ctx, tx, id, newJobID); err != nil {
+				return err
+			}
+			n++
+			return nil
+		}); err != nil {
+			log.Printf("[inbound-reconcile] enqueue %s: %v", id, err)
+		}
+	}
+	return n, nil
+}
+
+// EnqueueInboundProcessTx inserts the inbound_process job in the caller's accept-tx
+// (the same tx as the inbound_intake insert), returning the River job id to stamp on
+// the intake row so a committed accepted row always has its job.
+func (j *Jobs) EnqueueInboundProcessTx(ctx context.Context, tx pgx.Tx, intakeID string) (int64, error) {
+	res, err := j.enq.InsertTx(ctx, tx, InboundProcessArgs{IntakeID: intakeID}, &river.InsertOpts{
+		Queue:       jobs.QueueInbound,
+		MaxAttempts: MaxInboundAttempts,
+	})
+	if err != nil {
+		return 0, err
+	}
+	return res.Job.ID, nil
+}
diff --git a/internal/inboundprocess/reconcile_test.go b/internal/inboundprocess/reconcile_test.go
new file mode 100644
index 00000000..698e1435
--- /dev/null
+++ b/internal/inboundprocess/reconcile_test.go
@@ -0,0 +1,139 @@
+package inboundprocess_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/rivertype"
+
+	"github.com/Mnexa-AI/e2a/internal/identity"
+	"github.com/Mnexa-AI/e2a/internal/inboundprocess"
+	"github.com/Mnexa-AI/e2a/internal/testutil"
+)
+
+// TestPruneProcessedIntake covers retention: an old processed row is pruned; a recent
+// processed row and an accepted (unprocessed) row of any age are kept.
+func TestPruneProcessedIntake(t *testing.T) {
+	pool := testutil.TestDB(t)
+	store := identity.NewStore(pool)
+	ctx := context.Background()
+
+	insert := func(id, msgID, hash string) {
+		if err := store.WithTx(ctx, func(tx pgx.Tx) error {
+			_, e := store.InsertInboundIntakeTx(ctx, tx, id, "bot@prune.test", "a@b.test", "1.2.3.4", msgID, hash, []byte("raw"))
+			return e
+		}); err != nil {
+			t.Fatalf("insert %s: %v", id, err)
+		}
+	}
+	process := func(id, msgFK string) {
+		if err := store.WithTx(ctx, func(tx pgx.Tx) error {
+			return store.MarkInboundIntakeProcessedTx(ctx, tx, id, msgFK)
+		}); err != nil {
+			t.Fatalf("process %s: %v", id, err)
+		}
+	}
+
+	oldID := identity.NewInboundIntakeID()
+	insert(oldID, "<old@b.test>", "h1")
+	process(oldID, "msg_1")
+	if _, err := pool.Exec(ctx, `UPDATE inbound_intake SET processed_at = now() - interval '100 hours' WHERE id=$1`, oldID); err != nil {
+		t.Fatalf("backdate: %v", err)
+	}
+
+	recentID := identity.NewInboundIntakeID()
+	insert(recentID, "<recent@b.test>", "h2")
+	process(recentID, "msg_2")
+
+	accID := identity.NewInboundIntakeID()
+	insert(accID, "<acc@b.test>", "h3") // stays accepted
+
+	n, err := store.PruneProcessedIntake(ctx, 72*time.Hour)
+	if err != nil {
+		t.Fatalf("prune: %v", err)
+	}
+	if n < 1 {
+		t.Fatalf("should prune the old processed row, got %d", n)
+	}
+
+	exists := func(id string) bool {
+		var c int
+		_ = pool.QueryRow(ctx, `SELECT count(*) FROM inbound_intake WHERE id=$1`, id).Scan(&c)
+		return c > 0
+	}
+	if exists(oldID) {
+		t.Error("old processed row should be pruned")
+	}
+	if !exists(recentID) {
+		t.Error("recent processed row must be kept")
+	}
+	if !exists(accID) {
+		t.Error("accepted (unprocessed) row must be kept regardless of age")
+	}
+}
+
+// fakeEnq is a minimal jobs.Enqueuer that hands back an increasing job id without a
+// live River client — enough to exercise ReconcilePending's stamp path.
+type fakeEnq struct{ n int64 }
+
+func (f *fakeEnq) Insert(_ context.Context, _ river.JobArgs, _ *river.InsertOpts) (*rivertype.JobInsertResult, error) {
+	f.n++
+	return &rivertype.JobInsertResult{Job: &rivertype.JobRow{ID: f.n}}, nil
+}
+
+func (f *fakeEnq) InsertTx(_ context.Context, _ pgx.Tx, _ river.JobArgs, _ *river.InsertOpts) (*rivertype.JobInsertResult, error) {
+	f.n++
+	return &rivertype.JobInsertResult{Job: &rivertype.JobRow{ID: f.n}}, nil
+}
+
+// TestReconcilePending covers the startup cutover: an accepted intake row with no job
+// (a crash between insert and enqueue, or a pre-async row) gets a job stamped, and a
+// re-run does not re-enqueue it (idempotent via the process_job_id IS NULL guard).
+func TestReconcilePending(t *testing.T) {
+	pool := testutil.TestDB(t)
+	store := identity.NewStore(pool)
+	ctx := context.Background()
+
+	// Plant an accepted intake with NO job (stamp step skipped).
+	id := identity.NewInboundIntakeID()
+	if err := store.WithTx(ctx, func(tx pgx.Tx) error {
+		_, e := store.InsertInboundIntakeTx(ctx, tx, id, "bot@recon.test", "a@b.test", "1.2.3.4", "<recon@b.test>", "h", []byte("raw"))
+		return e
+	}); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+
+	j := inboundprocess.NewJobs(store)
+	j.SetEnqueuer(&fakeEnq{})
+
+	n, err := j.ReconcilePending(ctx, pool)
+	if err != nil {
+		t.Fatalf("reconcile: %v", err)
+	}
+	if n < 1 {
+		t.Fatalf("reconcile should enqueue at least the stranded row, got %d", n)
+	}
+	var jobID *int64
+	if err := pool.QueryRow(ctx, `SELECT process_job_id FROM inbound_intake WHERE id=$1`, id).Scan(&jobID); err != nil {
+		t.Fatalf("read job id: %v", err)
+	}
+	if jobID == nil {
+		t.Fatal("reconcile must stamp a job id on the stranded row")
+	}
+	stamped := *jobID
+
+	// Idempotent: a re-run must not re-enqueue this row (its job id is unchanged).
+	if _, err := j.ReconcilePending(ctx, pool); err != nil {
+		t.Fatalf("reconcile 2: %v", err)
+	}
+	var jobID2 *int64
+	if err := pool.QueryRow(ctx, `SELECT process_job_id FROM inbound_intake WHERE id=$1`, id).Scan(&jobID2); err != nil {
+		t.Fatalf("read job id 2: %v", err)
+	}
+	if jobID2 == nil || *jobID2 != stamped {
+		t.Fatalf("re-run must not re-enqueue an already-stamped row: was %d, now %v", stamped, jobID2)
+	}
+}
diff --git a/internal/inboundprocess/retention.go b/internal/inboundprocess/retention.go
new file mode 100644
index 00000000..fc5b4e02
--- /dev/null
+++ b/internal/inboundprocess/retention.go
@@ -0,0 +1,47 @@
+package inboundprocess
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/riverqueue/river"
+)
+
+const (
+	// retentionInterval is how often the processed-intake sweep runs.
+	retentionInterval = 6 * time.Hour
+	// retentionWindow keeps processed intake for a short debugging / re-drive window.
+	// The raw MIME also lives in messages.raw_message once processed, so pruning past
+	// this is non-destructive. Failed rows are retained (not pruned) for inspection.
+	retentionWindow = 72 * time.Hour
+)
+
+// InboundRetentionArgs drives one retention sweep (no payload — the worker prunes the
+// whole processed-and-old set).
+type InboundRetentionArgs struct{}
+
+func (InboundRetentionArgs) Kind() string { return "inbound_retention" }
+
+// Pruner is the retention surface (a subset of Store).
+type Pruner interface {
+	PruneProcessedIntake(ctx context.Context, olderThan time.Duration) (int64, error)
+}
+
+// RetentionWorker prunes processed intake older than retentionWindow. Scheduled as a
+// River periodic on QueueMaintenance (mirrors the webhook auto-disable janitor).
+type RetentionWorker struct {
+	river.WorkerDefaults[InboundRetentionArgs]
+	pruner Pruner
+}
+
+func (w *RetentionWorker) Work(ctx context.Context, _ *river.Job[InboundRetentionArgs]) error {
+	n, err := w.pruner.PruneProcessedIntake(ctx, retentionWindow)
+	if err != nil {
+		return err // retryable — the periodic re-runs next interval regardless
+	}
+	if n > 0 {
+		log.Printf("[inbound-retention] pruned %d processed intake rows older than %s", n, retentionWindow)
+	}
+	return nil
+}
diff --git a/internal/inboundprocess/retention_internal_test.go b/internal/inboundprocess/retention_internal_test.go
new file mode 100644
index 00000000..bd9632e4
--- /dev/null
+++ b/internal/inboundprocess/retention_internal_test.go
@@ -0,0 +1,32 @@
+package inboundprocess
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/riverqueue/river"
+)
+
+type fakePruner struct {
+	calls int
+	n     int64
+}
+
+func (f *fakePruner) PruneProcessedIntake(_ context.Context, _ time.Duration) (int64, error) {
+	f.calls++
+	return f.n, nil
+}
+
+// TestRetentionWorker_Work covers the retention sweep body: it calls the pruner once
+// with the retention window and does not error on a non-zero prune count.
+func TestRetentionWorker_Work(t *testing.T) {
+	p := &fakePruner{n: 3}
+	w := &RetentionWorker{pruner: p}
+	if err := w.Work(context.Background(), &river.Job[InboundRetentionArgs]{}); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+	if p.calls != 1 {
+		t.Errorf("prune calls = %d, want 1", p.calls)
+	}
+}
diff --git a/internal/inboundprocess/worker.go b/internal/inboundprocess/worker.go
new file mode 100644
index 00000000..7cd89921
--- /dev/null
+++ b/internal/inboundprocess/worker.go
@@ -0,0 +1,136 @@
+// Package inboundprocess is Layer 3 of the queue-first inbound pipeline
+// (docs/design/inbound-message-pipeline-river.md): the River execution stage that
+// takes an accepted inbound_intake row and runs the full processing chain — parse,
+// SPF/DKIM, HMAC sign, ingestion gate, content screening, persist, and event publish
+// — off the SMTP critical path. It mirrors internal/outboundsend: a River Worker on
+// the shared `inbound` queue, with River owning claim / retry / rescue.
+//
+// At-least-once from the SMTP edge: the intake row is committed (and the job
+// enqueued) in the same tx before 250, so an accepted message is never lost. The
+// worker's terminal write flips intake.status='processed' ATOMICALLY with the
+// messages insert + event publish (inside processInbound's tx), so a crash re-drive
+// finds 'processed' and no-ops — the idempotency gate.
+//
+// One processing pass per job attempt — River owns the multi-attempt envelope via
+// NextRetry, so Work() stays a single ProcessIntake call, not an internal loop.
+package inboundprocess
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/riverqueue/river"
+
+	"github.com/Mnexa-AI/e2a/internal/identity"
+)
+
+// inboundRetryBackoffs is the per-attempt delay for a failed processing attempt.
+// River drives it via NextRetry, indexed by attempt. Inbound processing errors are
+// transient (DB persist / agent resolve), so the tail is a bounded retry, not the
+// outage/permanent split the outbound sender needs (there is no upstream provider
+// call except the fail-open Gemini screen).
+var inboundRetryBackoffs = []time.Duration{
+	30 * time.Second,
+	2 * time.Minute,
+	10 * time.Minute,
+	1 * time.Hour,
+	4 * time.Hour,
+}
+
+// MaxInboundAttempts caps the retry tail before the intake is marked failed.
+const MaxInboundAttempts = 6
+
+// InboundProcessArgs drives one inbound processing job. Args carry only the intake
+// id; the worker re-reads the inbound_intake row (the source of truth) each attempt.
+type InboundProcessArgs struct {
+	IntakeID string `json:"intake_id"`
+}
+
+func (InboundProcessArgs) Kind() string { return "inbound_process" }
+
+// Processor runs the full inbound chain for one intake row and marks the intake
+// processed atomically with the messages insert. Implemented over the relay Server
+// (which owns processInbound) in the binary.
+type Processor interface {
+	ProcessIntake(ctx context.Context, intake *identity.InboundIntake) error
+}
+
+// Store is the intake surface the worker + reconciler need. Implemented over
+// internal/identity.
+type Store interface {
+	// LoadInboundIntake returns the row, or (nil, nil) if pruned — a no-op.
+	LoadInboundIntake(ctx context.Context, intakeID string) (*identity.InboundIntake, error)
+	// MarkInboundIntakeFailed records a terminal failure after the retry tail.
+	MarkInboundIntakeFailed(ctx context.Context, intakeID, detail string) error
+	// StampInboundIntakeJobIDTx records the job id on a reconciled row.
+	StampInboundIntakeJobIDTx(ctx context.Context, tx pgx.Tx, intakeID string, jobID int64) error
+	// PruneProcessedIntake deletes processed rows older than olderThan (retention).
+	PruneProcessedIntake(ctx context.Context, olderThan time.Duration) (int64, error)
+}
+
+// InboundProcessWorker processes an accepted intake row. Mirrors
+// outboundsend.SendWorker.
+type InboundProcessWorker struct {
+	river.WorkerDefaults[InboundProcessArgs]
+	store     Store
+	processor Processor
+}
+
+func NewInboundProcessWorker(store Store, processor Processor) *InboundProcessWorker {
+	return &InboundProcessWorker{store: store, processor: processor}
+}
+
+// NextRetry overrides River's default backoff with the decided inbound envelope.
+func (w *InboundProcessWorker) NextRetry(job *river.Job[InboundProcessArgs]) time.Time {
+	i := job.Attempt
+	if i < 0 || i >= len(inboundRetryBackoffs) {
+		return time.Time{} // fall back to River's default at the tail
+	}
+	return time.Now().Add(inboundRetryBackoffs[i])
+}
+
+func (w *InboundProcessWorker) Work(ctx context.Context, job *river.Job[InboundProcessArgs]) error {
+	it, err := w.store.LoadInboundIntake(ctx, job.Args.IntakeID)
+	if err != nil {
+		return err // DB error — retryable
+	}
+	if it == nil {
+		return nil // intake pruned — nothing to do
+	}
+	if it.Status != identity.IntakeStatusAccepted {
+		return nil // already processed/failed — idempotent re-drive no-op
+	}
+
+	if err := w.processor.ProcessIntake(ctx, it); err != nil {
+		if errors.Is(err, identity.ErrIntakeAlreadyProcessed) {
+			return nil // a concurrent/prior attempt already processed it — done
+		}
+		if errors.Is(err, identity.ErrRecipientGone) {
+			// Recipient's agent was deleted between accept and processing. Mark the
+			// intake terminally (not a retry) so it doesn't linger 'accepted' forever
+			// with orphaned raw MIME; the message is dropped (nothing to deliver).
+			if ferr := w.store.MarkInboundIntakeFailed(ctx, it.ID, "recipient agent no longer exists"); ferr != nil {
+				log.Printf("[inbound-process] mark-gone for %s: %v", it.ID, ferr)
+			}
+			return nil
+		}
+		// Processing errors are transient (DB persist / agent resolve). Retry within
+		// the bounded envelope; on exhaustion, mark the intake failed for ops
+		// visibility — we already returned 250, so the message is dropped, not lost
+		// silently. (A genuinely-undeliverable body doesn't error here: parsing is
+		// best-effort and screening fails open.)
+		if job.Attempt >= MaxInboundAttempts {
+			if ferr := w.store.MarkInboundIntakeFailed(ctx, it.ID, err.Error()); ferr != nil {
+				log.Printf("[inbound-process] CRITICAL: mark-failed for %s errored: %v", it.ID, ferr)
+			}
+			return fmt.Errorf("inbound processing failed (final attempt %d): %w", job.Attempt, err)
+		}
+		return fmt.Errorf("inbound processing attempt %d failed: %w", job.Attempt, err)
+	}
+	// Success — ProcessIntake flipped intake.status to 'processed' in its own tx.
+	return nil
+}
diff --git a/internal/inboundprocess/worker_test.go b/internal/inboundprocess/worker_test.go
new file mode 100644
index 00000000..66869e28
--- /dev/null
+++ b/internal/inboundprocess/worker_test.go
@@ -0,0 +1,157 @@
+package inboundprocess_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/rivertype"
+
+	"github.com/Mnexa-AI/e2a/internal/identity"
+	"github.com/Mnexa-AI/e2a/internal/inboundprocess"
+)
+
+type fakeStore struct {
+	intake  *identity.InboundIntake
+	loadErr error
+	failed  []string // intake ids marked failed
+}
+
+func (f *fakeStore) LoadInboundIntake(_ context.Context, _ string) (*identity.InboundIntake, error) {
+	return f.intake, f.loadErr
+}
+
+func (f *fakeStore) MarkInboundIntakeFailed(_ context.Context, id, _ string) error {
+	f.failed = append(f.failed, id)
+	return nil
+}
+
+func (f *fakeStore) StampInboundIntakeJobIDTx(_ context.Context, _ pgx.Tx, _ string, _ int64) error {
+	return nil
+}
+
+func (f *fakeStore) PruneProcessedIntake(_ context.Context, _ time.Duration) (int64, error) {
+	return 0, nil
+}
+
+type fakeProcessor struct {
+	err   error
+	calls int
+}
+
+func (f *fakeProcessor) ProcessIntake(_ context.Context, _ *identity.InboundIntake) error {
+	f.calls++
+	return f.err
+}
+
+func job(attempt int) *river.Job[inboundprocess.InboundProcessArgs] {
+	return &river.Job[inboundprocess.InboundProcessArgs]{
+		JobRow: &rivertype.JobRow{Attempt: attempt, MaxAttempts: inboundprocess.MaxInboundAttempts, Kind: inboundprocess.InboundProcessArgs{}.Kind()},
+		Args:   inboundprocess.InboundProcessArgs{IntakeID: "intk_1"},
+	}
+}
+
+func accepted() *identity.InboundIntake {
+	return &identity.InboundIntake{ID: "intk_1", Status: identity.IntakeStatusAccepted, Recipient: "bot@x.test", Raw: []byte("raw")}
+}
+
+func TestWorker_Success(t *testing.T) {
+	st := &fakeStore{intake: accepted()}
+	pr := &fakeProcessor{}
+	w := inboundprocess.NewInboundProcessWorker(st, pr)
+	if err := w.Work(context.Background(), job(0)); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+	if pr.calls != 1 {
+		t.Errorf("ProcessIntake calls = %d, want 1", pr.calls)
+	}
+	if len(st.failed) != 0 {
+		t.Errorf("no MarkFailed on success, got %v", st.failed)
+	}
+}
+
+func TestWorker_GoneIsNoOp(t *testing.T) {
+	st := &fakeStore{intake: nil} // pruned between accept and processing
+	pr := &fakeProcessor{}
+	w := inboundprocess.NewInboundProcessWorker(st, pr)
+	if err := w.Work(context.Background(), job(0)); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+	if pr.calls != 0 {
+		t.Error("a gone intake must not be processed")
+	}
+}
+
+func TestWorker_AlreadyProcessedIsNoOp(t *testing.T) {
+	it := accepted()
+	it.Status = identity.IntakeStatusProcessed
+	st := &fakeStore{intake: it}
+	pr := &fakeProcessor{}
+	w := inboundprocess.NewInboundProcessWorker(st, pr)
+	if err := w.Work(context.Background(), job(0)); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+	if pr.calls != 0 {
+		t.Error("an already-processed intake must not be re-processed (idempotency gate)")
+	}
+}
+
+// TestWorker_RecipientGoneMarksTerminal: a gone recipient (deleted agent) is dropped
+// terminally (intake marked failed), NOT retried — no error, so the job completes and
+// the intake doesn't linger 'accepted' with orphaned raw MIME.
+func TestWorker_RecipientGoneMarksTerminal(t *testing.T) {
+	st := &fakeStore{intake: accepted()}
+	pr := &fakeProcessor{err: identity.ErrRecipientGone}
+	w := inboundprocess.NewInboundProcessWorker(st, pr)
+	if err := w.Work(context.Background(), job(0)); err != nil {
+		t.Fatalf("a gone recipient should drop (no error), got %v", err)
+	}
+	if len(st.failed) != 1 || st.failed[0] != "intk_1" {
+		t.Errorf("a gone recipient must mark the intake terminal, got %v", st.failed)
+	}
+}
+
+func TestWorker_TransientRetryDoesNotMarkFailed(t *testing.T) {
+	st := &fakeStore{intake: accepted()}
+	pr := &fakeProcessor{err: errors.New("db blip")}
+	w := inboundprocess.NewInboundProcessWorker(st, pr)
+	if err := w.Work(context.Background(), job(0)); err == nil {
+		t.Fatal("a transient processing error should return a retryable error")
+	}
+	if len(st.failed) != 0 {
+		t.Errorf("an early attempt must NOT MarkFailed, got %v", st.failed)
+	}
+}
+
+// TestJobs_ProcessIntakeLateBinding covers the late-bound Processor: before
+// SetProcessor, Jobs.ProcessIntake returns a retryable error (no panic); after, it
+// delegates to the concrete processor.
+func TestJobs_ProcessIntakeLateBinding(t *testing.T) {
+	j := inboundprocess.NewJobs(&fakeStore{})
+	if err := j.ProcessIntake(context.Background(), accepted()); err == nil {
+		t.Fatal("an unwired processor should return a retryable error, not panic")
+	}
+	pr := &fakeProcessor{}
+	j.SetProcessor(pr)
+	if err := j.ProcessIntake(context.Background(), accepted()); err != nil {
+		t.Fatalf("wired ProcessIntake: %v", err)
+	}
+	if pr.calls != 1 {
+		t.Errorf("wired ProcessIntake should delegate once, got %d", pr.calls)
+	}
+}
+
+func TestWorker_FinalAttemptMarksFailed(t *testing.T) {
+	st := &fakeStore{intake: accepted()}
+	pr := &fakeProcessor{err: errors.New("db blip")}
+	w := inboundprocess.NewInboundProcessWorker(st, pr)
+	if err := w.Work(context.Background(), job(inboundprocess.MaxInboundAttempts)); err == nil {
+		t.Fatal("the final attempt should still return an error so River discards the job")
+	}
+	if len(st.failed) != 1 || st.failed[0] != "intk_1" {
+		t.Errorf("the final attempt must MarkFailed the intake, got %v", st.failed)
+	}
+}
diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go
index 9f2ee69c..f09a00c9 100644
--- a/internal/jobs/jobs.go
+++ b/internal/jobs/jobs.go
@@ -44,6 +44,7 @@ type Registrar interface {
 // Config sizes the per-queue worker pools. Zero values get sane defaults.
 type Config struct {
 	OutboundWorkers    int // QueueOutbound concurrency (default 8)
+	InboundWorkers     int // QueueInbound concurrency (default 8)
 	WebhookWorkers     int // QueueWebhook concurrency (default 16)
 	MaintenanceWorkers int // QueueMaintenance concurrency (default 2)
 	DefaultWorkers     int // QueueDefault concurrency (default 5)
@@ -53,6 +54,9 @@ func (c Config) withDefaults() Config {
 	if c.OutboundWorkers <= 0 {
 		c.OutboundWorkers = 8
 	}
+	if c.InboundWorkers <= 0 {
+		c.InboundWorkers = 8
+	}
 	if c.WebhookWorkers <= 0 {
 		c.WebhookWorkers = 16
 	}
@@ -99,7 +103,7 @@ func New(pool *pgxpool.Pool, cfg Config, registrars ...Registrar) (*Client, erro
 	}
 
 	rc, err := river.NewClient(riverpgxv5.New(pool), &river.Config{
-		Queues:       defaultQueueConfig(cfg.OutboundWorkers, cfg.WebhookWorkers, cfg.MaintenanceWorkers, cfg.DefaultWorkers),
+		Queues:       defaultQueueConfig(cfg.OutboundWorkers, cfg.InboundWorkers, cfg.WebhookWorkers, cfg.MaintenanceWorkers, cfg.DefaultWorkers),
 		Workers:      workers,
 		PeriodicJobs: periodic,
 	})
diff --git a/internal/jobs/queues.go b/internal/jobs/queues.go
index 60f7ab59..112c9c7a 100644
--- a/internal/jobs/queues.go
+++ b/internal/jobs/queues.go
@@ -11,6 +11,10 @@ import "github.com/riverqueue/river"
 const (
 	// QueueOutbound carries outbound send jobs (API → SES).
 	QueueOutbound = "outbound"
+	// QueueInbound carries inbound message-processing jobs (accepted raw MIME →
+	// parse/screen/persist/deliver). Isolated so an inbound spike (incl. the
+	// per-message Gemini screen) can't starve outbound sends or webhook delivery.
+	QueueInbound = "inbound"
 	// QueueWebhook carries customer webhook-delivery jobs. Isolated from outbound
 	// so a slow/failing endpoint's backlog never delays sends.
 	QueueWebhook = "webhook"
@@ -25,9 +29,10 @@ const (
 // MaxWorkers is deliberately generous for I/O-bound work (SMTP / HTTP); tune
 // against real throughput. Every queue a domain enqueues into MUST appear here or
 // its jobs sit unworked.
-func defaultQueueConfig(outbound, webhook, maintenance, deflt int) map[string]river.QueueConfig {
+func defaultQueueConfig(outbound, inbound, webhook, maintenance, deflt int) map[string]river.QueueConfig {
 	return map[string]river.QueueConfig{
 		QueueOutbound:    {MaxWorkers: outbound},
+		QueueInbound:     {MaxWorkers: inbound},
 		QueueWebhook:     {MaxWorkers: webhook},
 		QueueMaintenance: {MaxWorkers: maintenance},
 		QueueDefault:     {MaxWorkers: deflt},
diff --git a/internal/relay/inbound_async_test.go b/internal/relay/inbound_async_test.go
new file mode 100644
index 00000000..566759a9
--- /dev/null
+++ b/internal/relay/inbound_async_test.go
@@ -0,0 +1,114 @@
+package relay_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+
+	"github.com/Mnexa-AI/e2a/internal/config"
+	"github.com/Mnexa-AI/e2a/internal/headers"
+	"github.com/Mnexa-AI/e2a/internal/identity"
+	"github.com/Mnexa-AI/e2a/internal/relay"
+	"github.com/Mnexa-AI/e2a/internal/testutil"
+	"github.com/Mnexa-AI/e2a/internal/usage"
+	"github.com/Mnexa-AI/e2a/internal/webhookpub"
+	"github.com/Mnexa-AI/e2a/internal/ws"
+)
+
+// fakeInboundEnq stands in for the River enqueuer so the accept-tx can be exercised
+// without a live River client — it just hands back an increasing job id.
+type fakeInboundEnq struct{ calls int }
+
+func (f *fakeInboundEnq) EnqueueInboundProcessTx(_ context.Context, _ pgx.Tx, _ string) (int64, error) {
+	f.calls++
+	return int64(f.calls), nil
+}
+
+// TestInbound_AsyncAcceptAndDedup exercises the queue-first accept-tx (slice 4):
+// with E2A_INBOUND_MODE=async, a send lands ONE inbound_intake row (accepted, with a
+// job id) and NO messages row (processing is deferred to the worker), and a resend of
+// the same message dedups — no second intake row, no second enqueue.
+func TestInbound_AsyncAcceptAndDedup(t *testing.T) {
+	pool := testutil.TestDB(t)
+	store := identity.NewStore(pool)
+	ctx := context.Background()
+
+	const domain = "async-inbound.example.com"
+	const agentEmail = "bot@" + domain
+	user, err := store.CreateOrGetUser(ctx, "owner@"+domain, "O", "g-async-inbound")
+	if err != nil {
+		t.Fatalf("user: %v", err)
+	}
+	if _, err := store.ClaimOrCreateDomain(ctx, domain, user.ID); err != nil {
+		t.Fatalf("domain: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE domains SET verified=true WHERE domain=$1`, domain); err != nil {
+		t.Fatalf("verify domain: %v", err)
+	}
+	if _, err := store.CreateAgent(ctx, agentEmail, domain, "", "", "", user.ID); err != nil {
+		t.Fatalf("agent: %v", err)
+	}
+
+	port, err := freePort()
+	if err != nil {
+		t.Fatalf("freePort: %v", err)
+	}
+	cfg := &config.Config{
+		SMTP:    config.SMTPConfig{ListenAddr: "127.0.0.1:" + port, Domain: domain},
+		Inbound: config.InboundConfig{Mode: "async"},
+		Env:     "development",
+	}
+	signer := headers.NewSigner("test-relay-hmac-key-32-bytes-long!")
+	server := relay.NewServer(cfg, store, signer, usage.NewNoopUsageTracker(), ws.NewHub())
+	server.SetOutbox(webhookpub.NewOutbox(pool, webhookpub.StaticFlag(true)))
+	enq := &fakeInboundEnq{}
+	server.SetInboundEnqueuer(enq)
+	go func() { _ = server.ListenAndServe() }()
+	t.Cleanup(func() { _ = server.Close() })
+	waitForSMTP(t, cfg.SMTP.ListenAddr)
+
+	body := "From: sender@ext.test\r\nTo: " + agentEmail + "\r\nMessage-ID: <dedup-1@ext.test>\r\nSubject: hi\r\n\r\nhello"
+	sendSMTP(t, cfg.SMTP.ListenAddr, "sender@ext.test", agentEmail, body)
+
+	// One accepted intake row with a stamped job id.
+	var n int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM inbound_intake WHERE recipient=$1`, agentEmail).Scan(&n); err != nil {
+		t.Fatalf("count intake: %v", err)
+	}
+	if n != 1 {
+		t.Fatalf("want 1 intake row after async accept, got %d", n)
+	}
+	var status string
+	var jobID int64
+	if err := pool.QueryRow(ctx, `SELECT status, process_job_id FROM inbound_intake WHERE recipient=$1`, agentEmail).Scan(&status, &jobID); err != nil {
+		t.Fatalf("read intake: %v", err)
+	}
+	if status != identity.IntakeStatusAccepted || jobID == 0 {
+		t.Fatalf("intake status=%q job=%d; want accepted + a stamped job", status, jobID)
+	}
+	// Processing is deferred — no messages row yet.
+	var msgs int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM messages WHERE agent_id=$1`, agentEmail).Scan(&msgs); err != nil {
+		t.Fatalf("count messages: %v", err)
+	}
+	if msgs != 0 {
+		t.Fatalf("no messages row should exist yet (async defers processing), got %d", msgs)
+	}
+	if enq.calls != 1 {
+		t.Fatalf("enqueuer should fire once, got %d", enq.calls)
+	}
+
+	// Resend the SAME message (lost-ack MTA retry) → dedup: still one intake, no
+	// second enqueue.
+	sendSMTP(t, cfg.SMTP.ListenAddr, "sender@ext.test", agentEmail, body)
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM inbound_intake WHERE recipient=$1`, agentEmail).Scan(&n); err != nil {
+		t.Fatalf("recount intake: %v", err)
+	}
+	if n != 1 {
+		t.Fatalf("dedup: want 1 intake row after resend, got %d", n)
+	}
+	if enq.calls != 1 {
+		t.Fatalf("dedup: the enqueuer must not fire on the duplicate, calls=%d", enq.calls)
+	}
+}
diff --git a/internal/relay/inbound_persist_fail_test.go b/internal/relay/inbound_persist_fail_test.go
new file mode 100644
index 00000000..a93371ce
--- /dev/null
+++ b/internal/relay/inbound_persist_fail_test.go
@@ -0,0 +1,133 @@
+package relay_test
+
+import (
+	"context"
+	"net/smtp"
+	"strings"
+	"testing"
+
+	"github.com/Mnexa-AI/e2a/internal/config"
+	"github.com/Mnexa-AI/e2a/internal/headers"
+	"github.com/Mnexa-AI/e2a/internal/identity"
+	"github.com/Mnexa-AI/e2a/internal/relay"
+	"github.com/Mnexa-AI/e2a/internal/testutil"
+	"github.com/Mnexa-AI/e2a/internal/usage"
+	"github.com/Mnexa-AI/e2a/internal/webhookpub"
+	"github.com/Mnexa-AI/e2a/internal/ws"
+)
+
+// TestInbound_PersistFailure_Returns451 is the regression for the silent-loss fix
+// (slice 1): when the inbound messages insert fails, the SMTP session must return a
+// transient 451 so the sending MTA retries — NOT a 250 that silently drops the mail.
+// Before the fix, deliverToAgent swallowed the persist error and deliverMessages
+// returned nil → go-smtp wrote 250 for a message that was never stored.
+//
+// We force the insert to fail with a trigger scoped to THIS test's agent id (so it
+// can't affect any other row) and dropped before the pool closes. Safe on the shared
+// test DB because `make test` runs packages with -p 1 and relay tests don't
+// t.Parallel(), so nothing else inserts into messages during the window.
+func TestInbound_PersistFailure_Returns451(t *testing.T) {
+	pool := testutil.TestDB(t)
+	store := identity.NewStore(pool)
+	ctx := context.Background()
+
+	const domain = "persistfail.example.com"
+	const agentEmail = "bot@" + domain
+	user, err := store.CreateOrGetUser(ctx, "owner@"+domain, "O", "g-persistfail")
+	if err != nil {
+		t.Fatalf("user: %v", err)
+	}
+	if _, err := store.ClaimOrCreateDomain(ctx, domain, user.ID); err != nil {
+		t.Fatalf("domain: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE domains SET verified=true WHERE domain=$1`, domain); err != nil {
+		t.Fatalf("verify domain: %v", err)
+	}
+	if _, err := store.CreateAgent(ctx, agentEmail, domain, "", "", "", user.ID); err != nil {
+		t.Fatalf("agent: %v", err)
+	}
+
+	// Force the inbound insert for THIS agent to fail. Scoped by agent_id so other
+	// rows pass through (RETURN NEW); dropped before TruncateAll/pool.Close (t.Cleanup
+	// is LIFO, and TestDB registered its cleanup first).
+	if _, err := pool.Exec(ctx, `
+		CREATE OR REPLACE FUNCTION test_fail_persist_slice1() RETURNS trigger AS $f$
+		BEGIN
+			IF NEW.agent_id = '`+agentEmail+`' THEN
+				RAISE EXCEPTION 'forced persist failure (slice-1 regression test)';
+			END IF;
+			RETURN NEW;
+		END; $f$ LANGUAGE plpgsql;`); err != nil {
+		t.Fatalf("create trigger fn: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `
+		DROP TRIGGER IF EXISTS test_fail_persist_slice1 ON messages;
+		CREATE TRIGGER test_fail_persist_slice1 BEFORE INSERT ON messages
+			FOR EACH ROW EXECUTE FUNCTION test_fail_persist_slice1();`); err != nil {
+		t.Fatalf("create trigger: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(),
+			`DROP TRIGGER IF EXISTS test_fail_persist_slice1 ON messages; DROP FUNCTION IF EXISTS test_fail_persist_slice1();`)
+	})
+
+	port, err := freePort()
+	if err != nil {
+		t.Fatalf("freePort: %v", err)
+	}
+	cfg := &config.Config{
+		SMTP: config.SMTPConfig{ListenAddr: "127.0.0.1:" + port, Domain: domain},
+		Env:  "development",
+	}
+	signer := headers.NewSigner("test-relay-hmac-key-32-bytes-long!")
+	server := relay.NewServer(cfg, store, signer, usage.NewNoopUsageTracker(), ws.NewHub())
+	server.SetOutbox(webhookpub.NewOutbox(pool, webhookpub.StaticFlag(true)))
+	go func() { _ = server.ListenAndServe() }()
+	t.Cleanup(func() { _ = server.Close() })
+	waitForSMTP(t, cfg.SMTP.ListenAddr)
+
+	body := "From: sender@ext.test\r\nTo: " + agentEmail + "\r\nSubject: hi\r\n\r\nhello"
+	derr := sendSMTPCaptureErr(cfg.SMTP.ListenAddr, "sender@ext.test", agentEmail, body)
+
+	// The core assertion: the persist failure surfaces as a transient 451, not a 250.
+	if derr == nil {
+		t.Fatal("DATA returned success (250) despite a persist failure — silent loss regression")
+	}
+	if !strings.Contains(derr.Error(), "451") {
+		t.Fatalf("expected a 451 transient error, got: %v", derr)
+	}
+	// And nothing was persisted for the agent.
+	var n int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM messages WHERE agent_id=$1`, agentEmail).Scan(&n); err != nil {
+		t.Fatalf("count messages: %v", err)
+	}
+	if n != 0 {
+		t.Fatalf("expected 0 persisted messages after a failed insert, got %d", n)
+	}
+}
+
+// sendSMTPCaptureErr runs a full SMTP send and RETURNS the DATA-completion error
+// (surfaced at the final "." on w.Close()) instead of failing the test — so a caller
+// can assert the server's transient/permanent response code.
+func sendSMTPCaptureErr(addr, from, to, body string) error {
+	c, err := smtp.Dial(addr)
+	if err != nil {
+		return err
+	}
+	defer c.Close()
+	if err := c.Mail(from); err != nil {
+		return err
+	}
+	if err := c.Rcpt(to); err != nil {
+		return err
+	}
+	w, err := c.Data()
+	if err != nil {
+		return err
+	}
+	if _, err := w.Write([]byte(body)); err != nil {
+		return err
+	}
+	// The server's Data handler result (250 / 451) surfaces here, at the final ".".
+	return w.Close()
+}
diff --git a/internal/relay/inbound_process.go b/internal/relay/inbound_process.go
new file mode 100644
index 00000000..58662b82
--- /dev/null
+++ b/internal/relay/inbound_process.go
@@ -0,0 +1,93 @@
+package relay
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"log"
+	"net"
+
+	"github.com/emersion/go-smtp"
+	"github.com/jackc/pgx/v5"
+
+	"github.com/Mnexa-AI/e2a/internal/identity"
+)
+
+// acceptInbound is the queue-first accept-tx (E2A_INBOUND_MODE=async): it durably
+// lands the raw MIME in inbound_intake and enqueues one River processing job per
+// NEW recipient, all in ONE transaction, BEFORE returning 250. This is the only work
+// gating 250 — parse/screen/persist/deliver run later in the worker.
+//
+// A committed accept ⇒ 250 (responsibility taken; River delivers at-least-once). A
+// commit failure ⇒ 451, so the sending MTA retries the whole message instead of us
+// losing it. The dedup key (recipient, message_id, content_hash) makes a lost-ack
+// retry an idempotent no-op accept — the duplicate re-uses the row and enqueues no
+// second job.
+func (s *session) acceptInbound(ctx context.Context, body []byte, info threadInfo) error {
+	contentHash := contentHashHex(body)
+	envelopeFrom := extractEmail(s.from)
+	remoteIP := ""
+	if s.remoteIP != nil {
+		remoteIP = s.remoteIP.String() // stored as text; nil would serialize to "<nil>"
+	}
+	messageID := info.MessageID // sender's RFC 5322 Message-ID (may be "")
+
+	err := s.relay.store.WithTx(ctx, func(tx pgx.Tx) error {
+		for _, rcpt := range s.recipients {
+			intakeID := identity.NewInboundIntakeID()
+			inserted, e := s.relay.store.InsertInboundIntakeTx(ctx, tx, intakeID, rcpt, envelopeFrom, remoteIP, messageID, contentHash, body)
+			if e != nil {
+				return e
+			}
+			if !inserted {
+				// Duplicate (lost-ack MTA retry): the row + its job already exist —
+				// enqueue nothing, still answer 250. Idempotent accept.
+				log.Printf("[%s] inbound dedup hit for %s (msgid=%q) — no re-enqueue", s.id, rcpt, messageID)
+				continue
+			}
+			jobID, e := s.relay.inboundEnq.EnqueueInboundProcessTx(ctx, tx, intakeID)
+			if e != nil {
+				return e
+			}
+			if e := s.relay.store.StampInboundIntakeJobIDTx(ctx, tx, intakeID, jobID); e != nil {
+				return e
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		// Nothing committed (one tx, all-or-nothing) — 451 so the sender retries.
+		log.Printf("[%s] inbound accept-tx failed → 451 (sender will retry): %v", s.id, err)
+		return &smtp.SMTPError{Code: 451, EnhancedCode: smtp.EnhancedCode{4, 3, 0}, Message: "temporary failure accepting message; please retry"}
+	}
+	log.Printf("[%s] inbound accepted (async) recipients=%d msgid=%q", s.id, len(s.recipients), messageID)
+	return nil
+}
+
+// contentHashHex is the dedup content fingerprint — sha256 of the raw MIME, hex.
+func contentHashHex(body []byte) string {
+	sum := sha256.Sum256(body)
+	return hex.EncodeToString(sum[:])
+}
+
+// ProcessIntake runs the full inbound chain for an accepted inbound_intake row — the
+// async River worker's entry point (internal/inboundprocess.Processor). It rebuilds
+// the connection context from the persisted row (the worker has no live session) and
+// calls the shared processInbound with a hook that flips the intake to 'processed'
+// ATOMICALLY with the messages insert + event publish — the worker's idempotency
+// gate. The stored remote_ip text is parsed back to net.IP for SPF; an unparseable
+// value yields nil, which emailauth.Check treats as an unauthenticated source
+// (fail-safe, not a crash).
+func (srv *Server) ProcessIntake(ctx context.Context, it *identity.InboundIntake) error {
+	in := inboundInput{
+		Body:         it.Raw,
+		EnvelopeFrom: it.EnvelopeFrom,
+		RemoteIP:     net.ParseIP(it.RemoteIP),
+		Recipient:    it.Recipient,
+		TraceID:      it.ID,
+	}
+	hook := func(ctx context.Context, tx pgx.Tx, messageID string) error {
+		return srv.store.MarkInboundIntakeProcessedTx(ctx, tx, it.ID, messageID)
+	}
+	return srv.processInbound(ctx, in, hook)
+}
diff --git a/internal/relay/inbound_process_test.go b/internal/relay/inbound_process_test.go
new file mode 100644
index 00000000..16e8bfd4
--- /dev/null
+++ b/internal/relay/inbound_process_test.go
@@ -0,0 +1,101 @@
+package relay_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+
+	"github.com/Mnexa-AI/e2a/internal/config"
+	"github.com/Mnexa-AI/e2a/internal/headers"
+	"github.com/Mnexa-AI/e2a/internal/identity"
+	"github.com/Mnexa-AI/e2a/internal/relay"
+	"github.com/Mnexa-AI/e2a/internal/testutil"
+	"github.com/Mnexa-AI/e2a/internal/usage"
+	"github.com/Mnexa-AI/e2a/internal/webhookpub"
+	"github.com/Mnexa-AI/e2a/internal/ws"
+)
+
+// TestInbound_ProcessIntake_RealPath exercises the ACTUAL async worker Processor
+// (relay.Server.ProcessIntake → processInbound with the MarkInboundIntakeProcessedTx
+// hook), not a fake: an accepted intake is processed into a messages row with the
+// intake flipped to 'processed' atomically, and a re-drive is a no-op that creates no
+// second message (the status-guard idempotency crux).
+func TestInbound_ProcessIntake_RealPath(t *testing.T) {
+	pool := testutil.TestDB(t)
+	store := identity.NewStore(pool)
+	ctx := context.Background()
+
+	const domain = "process-real.example.com"
+	const agentEmail = "bot@" + domain
+	user, err := store.CreateOrGetUser(ctx, "owner@"+domain, "O", "g-process-real")
+	if err != nil {
+		t.Fatalf("user: %v", err)
+	}
+	if _, err := store.ClaimOrCreateDomain(ctx, domain, user.ID); err != nil {
+		t.Fatalf("domain: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE domains SET verified=true WHERE domain=$1`, domain); err != nil {
+		t.Fatalf("verify: %v", err)
+	}
+	if _, err := store.CreateAgent(ctx, agentEmail, domain, "", "", "", user.ID); err != nil {
+		t.Fatalf("agent: %v", err)
+	}
+
+	cfg := &config.Config{SMTP: config.SMTPConfig{Domain: domain}, Env: "development"}
+	signer := headers.NewSigner("test-relay-hmac-key-32-bytes-long!")
+	server := relay.NewServer(cfg, store, signer, usage.NewNoopUsageTracker(), ws.NewHub())
+	server.SetOutbox(webhookpub.NewOutbox(pool, webhookpub.StaticFlag(true)))
+
+	// Plant an accepted intake row (as acceptInbound would).
+	raw := []byte("From: alice@sender.test\r\nTo: " + agentEmail + "\r\nMessage-ID: <rp1@sender.test>\r\nSubject: real path\r\n\r\nbody")
+	id := identity.NewInboundIntakeID()
+	if err := store.WithTx(ctx, func(tx pgx.Tx) error {
+		_, e := store.InsertInboundIntakeTx(ctx, tx, id, agentEmail, "alice@sender.test", "1.2.3.4", "<rp1@sender.test>", "hash-rp", raw)
+		return e
+	}); err != nil {
+		t.Fatalf("plant intake: %v", err)
+	}
+
+	it, err := store.LoadInboundIntake(ctx, id)
+	if err != nil || it == nil {
+		t.Fatalf("load intake: %v", err)
+	}
+	if err := server.ProcessIntake(ctx, it); err != nil {
+		t.Fatalf("ProcessIntake: %v", err)
+	}
+
+	// messages row created + intake flipped processed + linked, atomically.
+	var subject, status string
+	if err := pool.QueryRow(ctx, `SELECT subject, status FROM messages WHERE agent_id=$1 AND direction='inbound'`, agentEmail).Scan(&subject, &status); err != nil {
+		t.Fatalf("messages row: %v", err)
+	}
+	if subject != "real path" {
+		t.Errorf("subject = %q, want %q", subject, "real path")
+	}
+	var intakeStatus string
+	var fk *string
+	if err := pool.QueryRow(ctx, `SELECT status, message_fk FROM inbound_intake WHERE id=$1`, id).Scan(&intakeStatus, &fk); err != nil {
+		t.Fatalf("read intake: %v", err)
+	}
+	if intakeStatus != identity.IntakeStatusProcessed || fk == nil {
+		t.Fatalf("intake status=%q fk=%v; want processed + linked", intakeStatus, fk)
+	}
+
+	// Re-drive: ProcessIntake on the now-processed intake. The hook's status guard
+	// (WHERE status='accepted' → 0 rows) aborts the persist tx with
+	// ErrIntakeAlreadyProcessed, so NO second messages row is created.
+	it2, _ := store.LoadInboundIntake(ctx, id)
+	rerr := server.ProcessIntake(ctx, it2)
+	if !errors.Is(rerr, identity.ErrIntakeAlreadyProcessed) {
+		t.Fatalf("re-drive should return ErrIntakeAlreadyProcessed, got %v", rerr)
+	}
+	var msgCount int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM messages WHERE agent_id=$1`, agentEmail).Scan(&msgCount); err != nil {
+		t.Fatalf("count messages: %v", err)
+	}
+	if msgCount != 1 {
+		t.Fatalf("re-drive must not create a second message, got %d", msgCount)
+	}
+}
diff --git a/internal/relay/server.go b/internal/relay/server.go
index daaeaf93..55bdda2d 100644
--- a/internal/relay/server.go
+++ b/internal/relay/server.go
@@ -7,6 +7,7 @@ import (
 	"crypto/tls"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"io"
 	"log"
 	"mime"
@@ -52,8 +53,25 @@ type Server struct {
 	// X-E2A-Conversation-ID directly; external senders fall back to the
 	// In-Reply-To lookup so they cannot forge conversation IDs.
 	outboundFromDomain string
+	// inboundAsync routes inbound through the queue-first River pipeline
+	// (E2A_INBOUND_MODE=async): the session durably accepts to inbound_intake +
+	// enqueues a processing job before 250 instead of processing inline. A nil
+	// inboundEnq forces the synchronous path regardless (fail-safe).
+	inboundAsync bool
+	inboundEnq   InboundEnqueuer
 }
 
+// InboundEnqueuer inserts the inbound_process job in the SMTP accept-tx (the same
+// transaction as the inbound_intake insert). *inboundprocess.Jobs satisfies it.
+// Injected via SetInboundEnqueuer; nil keeps inbound on the synchronous path.
+type InboundEnqueuer interface {
+	EnqueueInboundProcessTx(ctx context.Context, tx pgx.Tx, intakeID string) (int64, error)
+}
+
+// SetInboundEnqueuer wires the shared River client's inbound enqueuer, enabling the
+// queue-first accept path when E2A_INBOUND_MODE=async.
+func (s *Server) SetInboundEnqueuer(e InboundEnqueuer) { s.inboundEnq = e }
+
 // SetOutbox wires the transactional outbox. The inbound trigger commits the
 // messages row and the webhook_events outbox row in a single transaction (per
 // design §4.2); the drain fans out and enqueues River delivery jobs.
@@ -76,6 +94,7 @@ func NewServer(cfg *config.Config, store *identity.Store, signer *headers.Signer
 		screen:             buildScreenEngine(),
 		smtpDomain:         cfg.SMTP.Domain,
 		outboundFromDomain: cfg.OutboundSMTP.FromDomain,
+		inboundAsync:       cfg.Inbound.Mode == "async",
 	}
 
 	be := &backend{relay: s}
@@ -175,10 +194,6 @@ type session struct {
 	from       string
 	recipients []string
 	remoteIP   net.IP
-	// Extracted from inbound email for threading
-	inboundMsgID      string
-	inboundSubject    string
-	inboundThreadInfo threadInfo
 }
 
 func (s *session) AuthPlain(username, password string) error {
@@ -234,54 +249,110 @@ func (s *session) Data(r io.Reader) error {
 		return err
 	}
 
-	// Extract threading info from inbound email
+	// Extract threading info once, for the DATA log + the async accept path (the
+	// per-recipient processing recomputes its own threadInfo from the raw body).
 	threadInfo := extractThreadInfo(body)
-	s.inboundMsgID = threadInfo.MessageID
-	s.inboundSubject = threadInfo.Subject
-	s.inboundThreadInfo = threadInfo
-
-	// Prefer From header (human-readable) over SMTP envelope (may be SES bounce address)
-	senderEmail := extractEmail(s.from)
+	senderEmail := extractEmail(s.from) // prefer From header for the log
 	if threadInfo.From != "" {
 		senderEmail = threadInfo.From
 	}
 	log.Printf("[%s] [%s] DATA recipients=%v size=%d bytes", s.id, senderEmail, s.recipients, len(body))
 
-	// Deliver directly — no human identity lookup needed
-	return s.deliverMessages(ctx, senderEmail, body)
+	// Queue-first async path (E2A_INBOUND_MODE=async): durably accept the raw MIME to
+	// inbound_intake + enqueue a River processing job, all before 250 — processing
+	// happens off the SMTP critical path. Falls back to the synchronous inline path
+	// when the enqueuer isn't wired (fail-safe).
+	if s.relay.inboundAsync && s.relay.inboundEnq != nil {
+		return s.acceptInbound(ctx, body, threadInfo)
+	}
+	return s.deliverMessages(ctx, body)
 }
 
-// deliverMessages runs SPF/DKIM/DMARC checks and delivers to agents via webhook.
-func (s *session) deliverMessages(ctx context.Context, senderEmail string, body []byte) error {
-	// Authenticate against the TRUE SMTP envelope MAIL FROM (s.from), not the
-	// display senderEmail (which prefers the From header). SPF is an
-	// envelope-identity check (RFC 7208), and DMARC alignment must compare the
-	// real envelope domain against the From-header domain — passing the
-	// From-derived senderEmail here would make SPF-alignment a tautology
-	// (adversarial review F5). DKIM + From extraction come from the body.
-	domainAuth := emailauth.Check(s.remoteIP, extractEmail(s.from), body)
-	log.Printf("[%s] [%s] domain auth from %s (envelope %s): %s", s.id, senderEmail, s.remoteIP, s.from, domainAuth.Summary())
-
-	delivered := 0
+// deliverMessages resolves each recipient and processes the message synchronously
+// (the E2A_INBOUND_MODE=sync path): it calls processInbound inline with no
+// post-persist hook. A persist failure surfaces as a 451 so the sending MTA retries
+// (never a silent 250). The async path enqueues to River instead (see the accept-tx).
+func (s *session) deliverMessages(ctx context.Context, body []byte) error {
 	for _, rcpt := range s.recipients {
-		agent, err := s.relay.resolveAgent(ctx, rcpt)
-		if err != nil {
-			// Should not happen — Rcpt() already validated, but guard defensively
-			log.Printf("[%s] [%s] skipping %s: agent resolution failed: %v", s.id, senderEmail, rcpt, err)
-			continue
+		in := inboundInput{
+			Body:         body,
+			EnvelopeFrom: extractEmail(s.from),
+			RemoteIP:     s.remoteIP,
+			Recipient:    rcpt,
+			TraceID:      s.id,
+		}
+		if derr := s.relay.processInbound(ctx, in, nil); derr != nil {
+			if errors.Is(derr, identity.ErrRecipientGone) {
+				continue // recipient's agent is gone — skip it (historical skip+continue)
+			}
+			// Persist failed — return a transient SMTP error so the sending MTA
+			// retries the whole message (RFC 5321 §4.5.4.1) instead of us silently
+			// losing it under a 250. Multi-recipient caveat: a retry re-delivers to
+			// already-succeeded recipients (the sync path has no dedup) — duplicate
+			// beats loss, and the queue-first path (E2A_INBOUND_MODE=async) dedups.
+			log.Printf("[%s] persist failed for %s → 451 (sender will retry): %v", s.id, rcpt, derr)
+			return &smtp.SMTPError{Code: 451, EnhancedCode: smtp.EnhancedCode{4, 3, 0}, Message: "temporary failure storing message; please retry"}
 		}
-
-		s.deliverToAgent(ctx, agent, senderEmail, rcpt, body, domainAuth)
-		delivered++
 	}
-
-	log.Printf("[%s] [%s] session complete: delivered=%d/%d", s.id, senderEmail, delivered, len(s.recipients))
 	return nil
 }
 
-// deliverToAgent signs auth headers and delivers a single message to an agent.
-// Push agents get webhook delivery; poll agents get the message stored for retrieval.
-func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdentity, senderEmail, rcpt string, body []byte, domainAuth *emailauth.Result) {
+// inboundInput is the connection-derived data processInbound needs — the same facts
+// the live SMTP session holds and the intake row persists for the async worker (which
+// cannot recompute EnvelopeFrom/RemoteIP after the session closes).
+type inboundInput struct {
+	Body         []byte
+	EnvelopeFrom string // SMTP MAIL FROM
+	RemoteIP     net.IP // connecting IP (SPF); the worker parses the stored text form
+	Recipient    string // RCPT TO
+	TraceID      string // log correlation (session id / intake id)
+}
+
+// postPersistHook runs INSIDE processInbound's persist transaction, after the
+// messages insert + event publish, given the new message id. The async worker passes
+// MarkInboundIntakeProcessedTx so the intake flips to 'processed' atomically with the
+// message (its idempotency gate); the sync path passes nil.
+type postPersistHook func(ctx context.Context, tx pgx.Tx, messageID string) error
+
+// processInbound runs the full inbound chain — parse, SPF/DKIM, HMAC sign, ingestion
+// gate, content screening, persist, and event publish — for one recipient. It is the
+// SINGLE implementation shared by the synchronous SMTP session and the async River
+// worker (internal/inboundprocess); the post-persist hook is the only difference.
+//
+// Returns a non-nil error ONLY when the message could not be durably persisted (the
+// caller maps that to a 451 / a River retry). A recipient that no longer resolves to
+// an agent is a no-op (nil) — the mailbox is gone. Screening/metering fail open.
+func (srv *Server) processInbound(ctx context.Context, in inboundInput, hook postPersistHook) error {
+	// Recompute the connection-derived context from the raw bytes — identical whether
+	// we are in the live session or replaying a persisted intake row.
+	threadInfo := extractThreadInfo(in.Body)
+	senderEmail := extractEmail(in.EnvelopeFrom)
+	if threadInfo.From != "" {
+		senderEmail = threadInfo.From
+	}
+	// SPF/DKIM/DMARC against the TRUE envelope MAIL FROM (RFC 7208), not the From
+	// header — else SPF-alignment is a tautology (adversarial review F5).
+	domainAuth := emailauth.Check(in.RemoteIP, extractEmail(in.EnvelopeFrom), in.Body)
+	log.Printf("[%s] [%s] domain auth from %s (envelope %s): %s", in.TraceID, senderEmail, in.RemoteIP, in.EnvelopeFrom, domainAuth.Summary())
+
+	agent, err := srv.resolveAgent(ctx, in.Recipient)
+	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+		return err // genuine transient resolve failure — retryable (451 / River)
+	}
+	if errors.Is(err, pgx.ErrNoRows) || agent == nil {
+		// Recipient's agent was deleted between accept and processing — NOT a transient
+		// error. Return the ErrRecipientGone sentinel: the async worker marks the
+		// intake terminally (so it doesn't linger 'accepted' with orphaned raw MIME)
+		// and the sync path skips the recipient. Returning a plain error here would
+		// burn the whole retry envelope (~5.5h) and re-meter the undeliverable message
+		// on every attempt. (GetAgentByID returns ErrNoRows, not (nil,nil), for a gone
+		// agent — the nil check is defensive.)
+		log.Printf("[%s] recipient %s no longer resolves to an agent — dropping", in.TraceID, in.Recipient)
+		return identity.ErrRecipientGone
+	}
+	rcpt := in.Recipient
+	body := in.Body
+
 	// Generate the message ID up-front so it can be bound into the HMAC
 	// canonical. Recipients verify by reconstructing the canonical with
 	// the message_id from the payload — substituting the ID without
@@ -302,7 +373,7 @@ func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdent
 		MessageID:   messageID,
 		BodyHash:    headers.HashBody(body),
 	}
-	authHeaders := s.relay.signer.Sign(authPayload)
+	authHeaders := srv.signer.Sign(authPayload)
 
 	// Display sender for webhook / stored message prefers the first Reply-To
 	// when set, so recipients reply to the intended mailbox (matches how
@@ -310,8 +381,8 @@ func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdent
 	// address. The full Reply-To list is shipped separately on the
 	// webhook payload so downstream consumers can see all addresses.
 	displaySender := senderEmail
-	if len(s.inboundThreadInfo.ReplyTo) > 0 {
-		displaySender = s.inboundThreadInfo.ReplyTo[0]
+	if len(threadInfo.ReplyTo) > 0 {
+		displaySender = threadInfo.ReplyTo[0]
 	}
 
 	// Inbound trust policy ingestion gate (decision 10 / Slice 7a). Evaluate the
@@ -322,23 +393,18 @@ func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdent
 	//
 	// senderResolvable fails the gate closed for shared-relay "via e2a" mail,
 	// which authenticates but carries no per-agent identity (#299).
-	policyDecision := inboundpolicy.EvaluateIngestion(agent.InboundPolicy, agent.InboundAllowlist, senderEmail, s.senderResolvable(senderEmail))
+	policyDecision := inboundpolicy.EvaluateIngestion(agent.InboundPolicy, agent.InboundAllowlist, senderEmail, srv.senderResolvable(senderEmail))
 
 	// Content screening (Slice 4): run the per-agent inbound scan and record the
 	// audit trail (protection_events) + the denormalized verdict. Detection +
 	// annotation only here — review/block holds are a later slice, so the message
 	// still delivers.
-	screenRes := s.relay.screenInbound(ctx, agent, messageID, senderEmail, body, domainAuth, policyDecision)
-
-	// Record inbound usage (fail-open — never block inbound email)
-	if agent.UserID != "" {
-		s.relay.usage.RecordAndCheck(ctx, agent.UserID, agent.ID, agent.Domain, "inbound")
-	}
+	screenRes := srv.screenInbound(ctx, agent, messageID, senderEmail, body, domainAuth, policyDecision)
 
 	lookup := func(ctx context.Context, ids []string) (string, error) {
-		return s.relay.store.LookupConversationID(ctx, agent.ID, ids)
+		return srv.store.LookupConversationID(ctx, agent.ID, ids)
 	}
-	conversationID := resolveConversationID(ctx, s.inboundThreadInfo, s.envelopeFromTrusted(), lookup)
+	conversationID := resolveConversationID(ctx, threadInfo, srv.envelopeFromTrusted(in.EnvelopeFrom), lookup)
 
 	// All inbound is persisted to the pollable inbox. There is no per-agent
 	// webhook delivery anymore (push is via /v1/webhooks subscriptions), so
@@ -352,7 +418,7 @@ func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdent
 	// ON CONFLICT (id) DO NOTHING. Idempotency by construction; see
 	// design §5.1.
 	event := webhookpub.NewEvent(webhookpub.EventEmailReceived, agent.UserID, buildEmailReceivedPayload(
-		messageID, conversationID, displaySender, senderEmail, rcpt, s.inboundSubject, s.inboundThreadInfo, authHeaders, agent,
+		messageID, conversationID, displaySender, senderEmail, rcpt, threadInfo.Subject, threadInfo, authHeaders, agent,
 	))
 	event.AgentID = agent.ID
 	event.ConversationID = conversationID
@@ -381,9 +447,9 @@ func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdent
 			// reply-routing addresses separately so the signal is complete.
 			"from":           senderEmail,
 			"display_sender": displaySender,
-			"reply_to":       s.inboundThreadInfo.ReplyTo,
+			"reply_to":       threadInfo.ReplyTo,
 			"recipient":      rcpt,
-			"subject":        s.inboundSubject,
+			"subject":        threadInfo.Subject,
 			"policy":         agent.InboundPolicy,
 			"reason":         policyDecision.Reason,
 		})
@@ -408,7 +474,7 @@ func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdent
 			"direction":       "inbound",
 			"from":            senderEmail,
 			"recipient":       rcpt,
-			"subject":         s.inboundSubject,
+			"subject":         threadInfo.Subject,
 			"reason":          screenRes.Reason,
 			"reason_source":   screenRes.Denorm.ReviewReason,
 		})
@@ -433,7 +499,7 @@ func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdent
 			"direction":           "inbound",
 			"from":                senderEmail,
 			"recipient":           rcpt,
-			"subject":             s.inboundSubject,
+			"subject":             threadInfo.Subject,
 			"reason":              screenRes.Reason,
 			"reason_source":       screenRes.Denorm.ReviewReason,
 			"approval_expires_at": screenRes.Denorm.ApprovalExpiresAt,
@@ -470,16 +536,15 @@ func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdent
 	authVerdictJSON, _ := json.Marshal(domainAuth)
 
 	var inboundMsg *identity.Message
-	var err error
-	if s.relay.outbox != nil {
-		err = s.relay.store.WithTx(ctx, func(tx pgx.Tx) error {
+	if srv.outbox != nil {
+		err = srv.store.WithTx(ctx, func(tx pgx.Tx) error {
 			var txErr error
-			inboundMsg, txErr = s.relay.store.CreateInboundMessageInTx(
+			inboundMsg, txErr = srv.store.CreateInboundMessageInTx(
 				ctx, tx, messageID, agent.ID, displaySender, rcpt,
-				s.inboundMsgID, s.inboundSubject, conversationID,
+				threadInfo.MessageID, threadInfo.Subject, conversationID,
 				deliveryStatus, body, authHeaders, authVerdictJSON,
 				policyDecision.Flagged, policyDecision.Reason,
-				s.inboundThreadInfo.To, s.inboundThreadInfo.CC, s.inboundThreadInfo.ReplyTo,
+				threadInfo.To, threadInfo.CC, threadInfo.ReplyTo,
 				screenRes.Denorm,
 			)
 			if txErr != nil {
@@ -488,52 +553,79 @@ func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdent
 			// Held messages (review/block) are persisted but NOT delivered — suppress
 			// the email.received push (the agent only ever sees released messages).
 			if !screenRes.Hold {
-				if txErr = s.relay.outbox.PublishTx(ctx, tx, event); txErr != nil {
+				if txErr = srv.outbox.PublishTx(ctx, tx, event); txErr != nil {
 					return txErr
 				}
 			}
 			if flaggedEvent != nil {
-				if txErr = s.relay.outbox.PublishTx(ctx, tx, *flaggedEvent); txErr != nil {
+				if txErr = srv.outbox.PublishTx(ctx, tx, *flaggedEvent); txErr != nil {
 					return txErr
 				}
 			}
 			if blockedEvent != nil {
-				if txErr = s.relay.outbox.PublishTx(ctx, tx, *blockedEvent); txErr != nil {
+				if txErr = srv.outbox.PublishTx(ctx, tx, *blockedEvent); txErr != nil {
 					return txErr
 				}
 			}
 			if pendingReviewEvent != nil {
-				if txErr = s.relay.outbox.PublishTx(ctx, tx, *pendingReviewEvent); txErr != nil {
+				if txErr = srv.outbox.PublishTx(ctx, tx, *pendingReviewEvent); txErr != nil {
+					return txErr
+				}
+			}
+			// Async worker: flip the intake to 'processed' ATOMICALLY with the message
+			// + events, so a crash re-drive finds 'processed' and no-ops (the worker's
+			// idempotency gate). nil on the synchronous path.
+			if hook != nil {
+				if txErr = hook(ctx, tx, messageID); txErr != nil {
 					return txErr
 				}
 			}
 			return nil
 		})
 	} else {
-		inboundMsg, err = s.relay.store.CreateInboundMessage(
+		// No-outbox legacy path (sync only — the async worker always runs with the
+		// outbox wired). hook is not supported here; async requires srv.outbox.
+		inboundMsg, err = srv.store.CreateInboundMessage(
 			ctx, messageID, agent.ID, displaySender, rcpt,
-			s.inboundMsgID, s.inboundSubject, conversationID,
+			threadInfo.MessageID, threadInfo.Subject, conversationID,
 			deliveryStatus, body, authHeaders, authVerdictJSON,
 			policyDecision.Flagged, policyDecision.Reason,
-			s.inboundThreadInfo.To, s.inboundThreadInfo.CC, s.inboundThreadInfo.ReplyTo,
+			threadInfo.To, threadInfo.CC, threadInfo.ReplyTo,
 			screenRes.Denorm,
 		)
 	}
 	if err != nil {
-		log.Printf("[%s] [%s] failed to record inbound message: %v", s.id, senderEmail, err)
-		return
+		if errors.Is(err, identity.ErrIntakeAlreadyProcessed) {
+			// Benign at-least-once re-drive: a prior attempt already processed this
+			// intake, so the persist tx rolled back (no duplicate). Not a failure —
+			// the async worker treats this sentinel as done. Don't log it as an error.
+			return err
+		}
+		// Do NOT swallow — surface to deliverMessages so the SMTP session returns a
+		// 451 and the sender retries, instead of a 250 that silently drops the mail.
+		log.Printf("[%s] [%s] failed to record inbound message: %v", in.TraceID, senderEmail, err)
+		return err
 	}
 	_ = inboundMsg
 
+	// Record inbound usage AFTER the message is durably persisted (fail-open — never
+	// block inbound email). The async worker retries a failed attempt, so metering
+	// before the persist tx would over-count — billing an undeliverable message once
+	// per attempt. Post-persist + the worker's already-processed no-op gate ⇒ once per
+	// delivered message. Mirrors the outbound send path.
+	if agent.UserID != "" {
+		srv.usage.RecordAndCheck(ctx, agent.UserID, agent.ID, agent.Domain, "inbound")
+	}
+
 	// Append the screening audit rows (gate + scan violations) best-effort. Soft-ref
 	// + deterministic ids make this idempotent under MTA retry, so it's safe outside
 	// the message transaction.
-	s.relay.writeProtectionEvents(ctx, messageID, screenRes.Events)
+	srv.writeProtectionEvents(ctx, messageID, screenRes.Events)
 
 	slug, _, _ := strings.Cut(rcpt, "@")
 
 	log.Printf("[mail:%s] dir=inbound from=%s to=%s slug=%s conv_id=%s subject=%q verified=%t",
-		messageID, displaySender, rcpt, slug, conversationID, s.inboundSubject, domainAuth.DomainAuthenticated())
+		messageID, displaySender, rcpt, slug, conversationID, threadInfo.Subject, domainAuth.DomainAuthenticated())
 
 	// Inbound events (email.received + flagged/blocked/pending_review variants)
 	// are written to the outbox (webhook_events) above, in the message tx; the
@@ -544,12 +636,13 @@ func (s *session) deliverToAgent(ctx context.Context, agent *identity.AgentIdent
 	// /v1/webhooks subscriber resource (driven by the outbox drain) is the
 	// durable push path; WS is an opportunistic live-tail on top of it,
 	// available to every agent regardless of how it's configured.
-	if s.relay.hub != nil && !screenRes.Hold && s.relay.hub.IsConnected(agent.ID) {
+	if srv.hub != nil && !screenRes.Hold && srv.hub.IsConnected(agent.ID) {
 		notification := buildWSNotification(inboundMsg)
-		if s.relay.hub.Send(agent.ID, notification) {
+		if srv.hub.Send(agent.ID, notification) {
 			log.Printf("[mail:%s] ws_notify=sent slug=%s", messageID, slug)
 		}
 	}
+	return nil
 }
 
 // The envelope wrapper ({event, id, created_at, data}) is added by the publisher
@@ -615,9 +708,6 @@ func splitEmail(addr string) (local, domain string) {
 func (s *session) Reset() {
 	s.from = ""
 	s.recipients = nil
-	s.inboundMsgID = ""
-	s.inboundSubject = ""
-	s.inboundThreadInfo = threadInfo{}
 }
 
 func (s *session) Logout() error {
@@ -663,12 +753,12 @@ func resolveConversationID(ctx context.Context, info threadInfo, trusted bool, l
 // send.e2a.dev but the envelope comes back as
 // <bounce-id>@mail.send.e2a.dev). We own all subdomains of our outbound
 // domain, so trusting them is the same trust boundary.
-func (s *session) envelopeFromTrusted() bool {
-	if s.relay.outboundFromDomain == "" {
+func (srv *Server) envelopeFromTrusted(envelopeFrom string) bool {
+	if srv.outboundFromDomain == "" {
 		return false
 	}
-	envDomain := strings.ToLower(extractDomain(s.from))
-	trusted := strings.ToLower(s.relay.outboundFromDomain)
+	envDomain := strings.ToLower(extractDomain(envelopeFrom))
+	trusted := strings.ToLower(srv.outboundFromDomain)
 	return envDomain == trusted || strings.HasSuffix(envDomain, "."+trusted)
 }
 
@@ -686,12 +776,12 @@ func (s *session) envelopeFromTrusted() bool {
 //
 // Matches the exact relay domain OR any subdomain, mirroring envelopeFromTrusted:
 // a verified agent always sends from its own custom domain, never the relay's.
-func (s *session) senderResolvable(senderEmail string) bool {
-	if s.relay.outboundFromDomain == "" {
+func (srv *Server) senderResolvable(senderEmail string) bool {
+	if srv.outboundFromDomain == "" {
 		return true
 	}
 	dom := strings.ToLower(extractDomain(senderEmail))
-	relay := strings.ToLower(s.relay.outboundFromDomain)
+	relay := strings.ToLower(srv.outboundFromDomain)
 	return dom != relay && !strings.HasSuffix(dom, "."+relay)
 }
 
diff --git a/internal/relay/server_test.go b/internal/relay/server_test.go
index c798c02a..3add3f9e 100644
--- a/internal/relay/server_test.go
+++ b/internal/relay/server_test.go
@@ -149,9 +149,7 @@ func TestExtractThreadInfoConversationID(t *testing.T) {
 }
 
 func TestEnvelopeFromTrusted(t *testing.T) {
-	s := &session{
-		relay: &Server{outboundFromDomain: "send.e2a.dev"},
-	}
+	srv := &Server{outboundFromDomain: "send.e2a.dev"}
 
 	cases := []struct {
 		envelope string
@@ -167,8 +165,7 @@ func TestEnvelopeFromTrusted(t *testing.T) {
 		{"agent@evilsend.e2a.dev", false},              // suffix-match attack (no dot before)
 	}
 	for _, c := range cases {
-		s.from = c.envelope
-		if got := s.envelopeFromTrusted(); got != c.want {
+		if got := srv.envelopeFromTrusted(c.envelope); got != c.want {
 			t.Errorf("envelopeFromTrusted(from=%q) = %v, want %v", c.envelope, got, c.want)
 		}
 	}
@@ -176,11 +173,8 @@ func TestEnvelopeFromTrusted(t *testing.T) {
 
 func TestEnvelopeFromTrustedUnconfigured(t *testing.T) {
 	// If outboundFromDomain isn't configured, trust nothing — fail closed.
-	s := &session{
-		relay: &Server{outboundFromDomain: ""},
-		from:  "agent@send.e2a.dev",
-	}
-	if s.envelopeFromTrusted() {
+	srv := &Server{outboundFromDomain: ""}
+	if srv.envelopeFromTrusted("agent@send.e2a.dev") {
 		t.Error("should not trust any envelope when outboundFromDomain is empty")
 	}
 }
@@ -188,9 +182,7 @@ func TestEnvelopeFromTrustedUnconfigured(t *testing.T) {
 func TestSenderResolvable(t *testing.T) {
 	// #299: the shared "via e2a" relay sender (agent@<outboundFromDomain>) carries
 	// no per-agent identity, so the inbound gate must treat it as unresolvable.
-	s := &session{
-		relay: &Server{outboundFromDomain: "send.e2a.dev"},
-	}
+	srv := &Server{outboundFromDomain: "send.e2a.dev"}
 	cases := []struct {
 		sender string
 		want   bool
@@ -204,7 +196,7 @@ func TestSenderResolvable(t *testing.T) {
 		{"nodomain", true},                     // garbage; let normal matching/flagging handle it
 	}
 	for _, c := range cases {
-		if got := s.senderResolvable(c.sender); got != c.want {
+		if got := srv.senderResolvable(c.sender); got != c.want {
 			t.Errorf("senderResolvable(%q) = %v, want %v", c.sender, got, c.want)
 		}
 	}
@@ -213,8 +205,8 @@ func TestSenderResolvable(t *testing.T) {
 func TestSenderResolvableUnconfigured(t *testing.T) {
 	// With no outboundFromDomain there is no shared-relay address to recognize, so
 	// every sender is treated as resolvable (no spurious flagging).
-	s := &session{relay: &Server{outboundFromDomain: ""}}
-	if !s.senderResolvable("agent@send.e2a.dev") {
+	srv := &Server{outboundFromDomain: ""}
+	if !srv.senderResolvable("agent@send.e2a.dev") {
 		t.Error("unconfigured relay should treat all senders as resolvable")
 	}
 }
@@ -287,12 +279,10 @@ func TestExtractThreadInfoDecodesEncodedSubject(t *testing.T) {
 	}
 }
 
-func TestSessionResetClearsThreadInfo(t *testing.T) {
+func TestSessionReset(t *testing.T) {
 	s := &session{
-		from:           "alice@example.com",
-		recipients:     []string{"bot@agent.example.com"},
-		inboundMsgID:   "<abc@gmail.com>",
-		inboundSubject: "Hello",
+		from:       "alice@example.com",
+		recipients: []string{"bot@agent.example.com"},
 	}
 
 	s.Reset()
@@ -303,10 +293,4 @@ func TestSessionResetClearsThreadInfo(t *testing.T) {
 	if s.recipients != nil {
 		t.Errorf("recipients should be nil after Reset, got %v", s.recipients)
 	}
-	if s.inboundMsgID != "" {
-		t.Errorf("inboundMsgID should be empty after Reset, got %q", s.inboundMsgID)
-	}
-	if s.inboundSubject != "" {
-		t.Errorf("inboundSubject should be empty after Reset, got %q", s.inboundSubject)
-	}
 }
diff --git a/migrations/056_inbound_intake.sql b/migrations/056_inbound_intake.sql
new file mode 100644
index 00000000..060c7db2
--- /dev/null
+++ b/migrations/056_inbound_intake.sql
@@ -0,0 +1,54 @@
+-- 056_inbound_intake.sql
+--
+-- Queue-first inbound pipeline (docs/design/inbound-message-pipeline-river.md).
+-- inbound_intake is Layer 1: the durable landing pad for a received message. The
+-- SMTP session writes the raw MIME + envelope + connecting IP here (the ONLY work
+-- that gates 250) and enqueues a River job (QueueInbound) referencing the row in the
+-- same transaction; the internal/inboundprocess worker then parses, screens,
+-- persists the messages row, and delivers — off the SMTP critical path.
+--
+-- This is NOT a hand-rolled claim queue (no SKIP-LOCKED lease here): River owns
+-- claim/retry/rescue via the job. The table is just the durable record the job reads
+-- (raw is up to 10MB — too large for a job arg) plus the dedup key.
+--
+-- remote_ip is captured because SPF (RFC 7208) needs the connecting IP, available
+-- only in-session — the async worker cannot recompute it.
+--
+-- Additive + idempotent. inbound_intake is a fresh table (no prod-size rewrite risk).
+
+CREATE TABLE IF NOT EXISTS inbound_intake (
+    id             TEXT PRIMARY KEY,                         -- intk_<rand>
+    recipient      TEXT NOT NULL,                            -- RCPT TO (the agent)
+    envelope_from  TEXT NOT NULL DEFAULT '',                 -- MAIL FROM (for SPF/DMARC)
+    remote_ip      TEXT NOT NULL DEFAULT '',                 -- connecting IP (for SPF)
+    raw_message    BYTEA NOT NULL,                           -- the raw MIME
+    message_id     TEXT NOT NULL DEFAULT '',                 -- sender's RFC 5322 Message-ID (dedup)
+    content_hash   TEXT NOT NULL,                            -- sha256 of raw (dedup)
+    status         TEXT NOT NULL DEFAULT 'accepted'
+        CHECK (status IN ('accepted', 'processed', 'failed')),
+    process_job_id BIGINT,                                   -- River QueueInbound job id
+    message_fk     TEXT,                                     -- resulting messages.id once processed
+    detail         TEXT,                                     -- failure detail (status='failed')
+    created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),       -- accept time
+    processed_at   TIMESTAMPTZ                               -- terminal (processed/failed) time
+);
+
+-- Dedup: an MTA retry after a lost 250 re-sends the same (recipient, Message-ID,
+-- body). ON CONFLICT DO NOTHING against this index makes the accept idempotent — a
+-- duplicate re-uses the row and enqueues no second job. When Message-ID is absent
+-- ('') the key degrades to (recipient, content_hash), still collapsing identical
+-- resends.
+CREATE UNIQUE INDEX IF NOT EXISTS idx_inbound_intake_dedup
+    ON inbound_intake (recipient, message_id, content_hash);
+
+-- Startup/periodic reconciler: accepted rows that never got a job (crash between
+-- insert and enqueue, or the mode-flip moment). Partial index keeps it cheap.
+CREATE INDEX IF NOT EXISTS idx_inbound_intake_unenqueued
+    ON inbound_intake (created_at)
+    WHERE status = 'accepted' AND process_job_id IS NULL;
+
+-- Retention sweep: prune processed rows older than the window (raw also lives in
+-- messages.raw_message once processed).
+CREATE INDEX IF NOT EXISTS idx_inbound_intake_processed_at
+    ON inbound_intake (processed_at)
+    WHERE status = 'processed';