diff --git a/backend/migrations/0035_search_fts.sql b/backend/migrations/0035_search_fts.sql
new file mode 100644
index 0000000..519c213
--- /dev/null
+++ b/backend/migrations/0035_search_fts.sql
@@ -0,0 +1,55 @@
+-- Weighted full-text search column (search_fts) + version stamp, kept fresh by
+-- a BEFORE trigger. Fast, metadata-only DDL only (README D1): the nullable
+-- columns force no table rewrite on PG16, and the function/trigger are instant.
+-- Pre-existing rows are populated by the resumable drainer (ftsBackfill.js);
+-- the GIN index is built CONCURRENTLY in the separate no-transaction migration
+-- 0037 (a $$-quoted plpgsql body cannot survive the no-transaction ; splitter,
+-- so the trigger and the CONCURRENTLY index MUST live in different files).
+--
+-- The setweight(...) expression MUST stay identical to
+-- lexicalRepo.searchFtsExpr('NEW'); fts_version = 1 matches lexicalRepo.FTS_VERSION.
+-- Idempotent (IF NOT EXISTS / CREATE OR REPLACE / DROP IF EXISTS) because a
+-- crash before the schema_migrations INSERT retries this migration.
+
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS search_fts tsvector;
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS fts_version int;
+
+CREATE OR REPLACE FUNCTION messages_search_fts_refresh() RETURNS trigger AS $$
+BEGIN
+  -- Skip recompute on an UPDATE that changes none of the indexed source
+  -- columns (read/star flag flips, snippet-only writes), so the trigger does
+  -- not tax hot sync UPSERTs; this also lets the backfill's explicit SET win.
+  IF TG_OP = 'UPDATE'
+     AND NEW.subject      IS NOT DISTINCT FROM OLD.subject
+     AND NEW.from_name    IS NOT DISTINCT FROM OLD.from_name
+     AND NEW.from_email   IS NOT DISTINCT FROM OLD.from_email
+     AND NEW.to_addresses IS NOT DISTINCT FROM OLD.to_addresses
+     AND NEW.cc_addresses IS NOT DISTINCT FROM OLD.cc_addresses
+     AND NEW.body_text    IS NOT DISTINCT FROM OLD.body_text
+  THEN
+     RETURN NEW;
+  END IF;
+
+  BEGIN
+    NEW.search_fts := setweight(to_tsvector('english', coalesce(NEW.subject,'')), 'A') ||
+    setweight(to_tsvector('english', coalesce(NEW.from_name,'') || ' ' || coalesce(NEW.from_email,'')), 'B') ||
+    setweight(to_tsvector('english', coalesce(NEW.to_addresses::text,'') || ' ' || coalesce(NEW.cc_addresses::text,'')), 'C') ||
+    setweight(to_tsvector('english', LEFT(coalesce(NEW.body_text,''), 600000)), 'D');
+    NEW.fts_version := 1;
+  EXCEPTION WHEN program_limit_exceeded THEN
+    -- Even with the 600k LEFT cap, a pathologically dense/multibyte body can
+    -- exceed Postgres's ~1MB tsvector limit (SQLSTATE 54000). Never fail the
+    -- row write: leave search_fts NULL so the message still persists and stays
+    -- findable via the ILIKE fallback; the backfill's row-by-row skip stamps it.
+    NEW.search_fts := NULL;
+    NEW.fts_version := NULL;
+  END;
+
+  RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS trg_messages_search_fts ON messages;
+CREATE TRIGGER trg_messages_search_fts
+  BEFORE INSERT OR UPDATE ON messages
+  FOR EACH ROW EXECUTE FUNCTION messages_search_fts_refresh();
diff --git a/backend/migrations/0036_background_jobs.sql b/backend/migrations/0036_background_jobs.sql
new file mode 100644
index 0000000..d189933
--- /dev/null
+++ b/backend/migrations/0036_background_jobs.sql
@@ -0,0 +1,19 @@
+-- Generic progress substrate for observable background drainers (first consumer:
+-- the FTS backfill). Plain table, fast DDL. One row per (kind, account); global
+-- jobs use a NULL account_id, COALESCE'd to '' in the unique index so the upsert
+-- has a single conflict target for both cases.
+
+CREATE TABLE IF NOT EXISTS background_jobs (
+  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  kind        VARCHAR(64) NOT NULL,
+  account_id  UUID REFERENCES email_accounts(id) ON DELETE CASCADE,
+  state       VARCHAR(20) NOT NULL DEFAULT 'idle',
+  processed   BIGINT NOT NULL DEFAULT 0,
+  total       BIGINT NOT NULL DEFAULT 0,
+  last_error  TEXT,
+  started_at  TIMESTAMPTZ,
+  updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS ux_background_jobs_kind_account
+  ON background_jobs (kind, COALESCE(account_id::text, ''));
diff --git a/backend/migrations/0037_search_fts_index.sql b/backend/migrations/0037_search_fts_index.sql
new file mode 100644
index 0000000..f82b7db
--- /dev/null
+++ b/backend/migrations/0037_search_fts_index.sql
@@ -0,0 +1,24 @@
+-- no-transaction
+--
+-- GIN index that serves `search_fts @@ tsquery`, plus a partial btree that lets
+-- the backfill drainer (and any "needs backfill" probe) find not-yet-stamped
+-- rows without a full seq scan; it self-prunes as fts_version = 1 fills in.
+-- CONCURRENTLY must run outside a transaction, so these live apart from 0035's
+-- trigger. Idempotent via IF NOT EXISTS (retried if a crash precedes the
+-- schema_migrations INSERT).
+--
+-- DROP ... IF EXISTS before each CREATE: a cancelled or crashed CREATE INDEX
+-- CONCURRENTLY leaves an INVALID index under the target name. On retry, plain
+-- IF NOT EXISTS sees that name and silently skips the create, recording the
+-- migration as done while the scan stays unindexed forever. This file only re-runs
+-- after such a failure — in which case the index is either absent (drop is a no-op)
+-- or invalid (drop removes the dead stub) — so dropping first is safe and cheap.
+
+DROP INDEX CONCURRENTLY IF EXISTS idx_messages_search_fts;
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_messages_search_fts
+  ON messages USING GIN (search_fts);
+
+DROP INDEX CONCURRENTLY IF EXISTS idx_messages_fts_stale_v1;
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_messages_fts_stale_v1
+  ON messages (date DESC)
+  WHERE fts_version IS DISTINCT FROM 1;
diff --git a/backend/src/index.js b/backend/src/index.js
index 483dcfa..70b0a85 100644
--- a/backend/src/index.js
+++ b/backend/src/index.js
@@ -18,6 +18,7 @@ import accountRoutes from './routes/accounts.js';
 import mailRoutes from './routes/mail.js';
 import searchRoutes from './routes/search.js';
 import adminRoutes from './routes/admin.js';
+import indexingRoutes from './routes/indexing.js';
 import totpRoutes from './routes/totp.js';
 import oidcApiRouter, { oidcBrowserRouter } from './routes/oidc.js';
 import rulesRoutes from './routes/rules.js';
@@ -30,6 +31,7 @@ import gtdRoutes from './routes/gtd.js';
 import carddavRouter from './routes/carddav.js';
 import carddavAccountRouter from './routes/carddavAccount.js';
 import { startCardavScheduler } from './services/carddavSync.js';
+import { scheduleFtsBackfill } from './services/search/ftsBackfill.js';
 import { encryptExistingCredentials, query } from './services/db.js';
 import { runMigrations } from './services/migrations.js';
 import { parseVCard } from './utils/vcard.js';
@@ -167,6 +169,7 @@ app.use('/api/mail', sendRoutes);
 app.use('/api/mail', draftRoutes);
 app.use('/api/search', searchRoutes);
 app.use('/api/admin', adminRoutes);
+app.use('/api/admin/indexing', indexingRoutes);
 app.use('/api/totp', totpRoutes);
 app.use('/api/rules', rulesRoutes);
 app.use('/api/block-list', blockListRoutes);
@@ -244,6 +247,9 @@ imapManager.startSnoozeWatcher();
 // Schedule periodic CardDAV contact sync for any connected accounts.
 startCardavScheduler();
 
+// Postgres-only drainer: populate search_fts for pre-existing rows (no IMAP).
+scheduleFtsBackfill();
+
 // Re-connect all enabled IMAP accounts on startup with bounded concurrency so a
 // large user base doesn't hammer IMAP servers and the DB connection pool at once.
 try {
diff --git a/backend/src/routes/accounts.js b/backend/src/routes/accounts.js
index 3ef08d7..d947ead 100644
--- a/backend/src/routes/accounts.js
+++ b/backend/src/routes/accounts.js
@@ -9,6 +9,9 @@ import { getConnectionPolicy } from '../services/connectionPolicy.js';
 import { invalidateGtdConfigCache, sanitizeGtdFoldersDetailed, findGtdFolderCollisions, DEFAULT_GTD_FOLDERS } from '../services/gtdConfig.js';
 import { invalidateOwnerAddressesCache } from '../services/gtdTransitions.js';
 import { createKeyedSerializer } from '../utils/keyedSerializer.js';
+import { startAccountBodyBackfill } from '../services/bodyBackfill.js';
+import { providerProfile } from '../services/imapManager.js';
+import { upsertJob } from '../services/backgroundJobs.js';
 
 // Serialize an account's reconnect triggers so a rapid settings change (e.g. a
 // gtd_enabled double-toggle) can't fire two overlapping disconnect→connect chains —
@@ -421,7 +424,18 @@ router.post('/:id/reindex', async (req, res) => {
         console.error(`Manual reindex error for ${account.email_address}:`, err.message)
       );
     }
-    res.json({ ok: true, alreadyRunning });
+
+    // Also kick a body-materialization pass for allowlisted providers. Fire-and-forget: the
+    // drainer self-guards against concurrent runs and caps each session, and Gmail/PurelyMail/
+    // Microsoft are gated off inside startBodyBackfill. Progress lands in background_jobs.
+    const bodyBackfillEnabled = providerProfile(account).bodyBackfill;
+    if (bodyBackfillEnabled) {
+      startAccountBodyBackfill(account, imapManager, upsertJob).catch(err =>
+        console.error(`Body backfill error for ${account.email_address}:`, err.message)
+      );
+    }
+
+    res.json({ ok: true, alreadyRunning, bodyBackfillEnabled });
   } catch (err) {
     console.error('POST /accounts/:id/reindex error:', err.message);
     res.status(500).json({ error: 'Failed to start reindex' });
diff --git a/backend/src/routes/indexing.js b/backend/src/routes/indexing.js
new file mode 100644
index 0000000..d8966d2
--- /dev/null
+++ b/backend/src/routes/indexing.js
@@ -0,0 +1,13 @@
+import { Router } from 'express';
+import { requireAdmin } from '../middleware/auth.js';
+import { listJobs } from '../services/backgroundJobs.js';
+
+const router = Router();
+
+// GET /api/admin/indexing/status — admin-gated (same guard as routes/ai.js).
+// Surfaces every background drainer's progress, e.g. "FTS backfill: N/M".
+router.get('/status', requireAdmin, async (_req, res) => {
+  res.json({ jobs: await listJobs() });
+});
+
+export default router;
diff --git a/backend/src/routes/search.js b/backend/src/routes/search.js
index dbd4379..cb3c8a6 100644
--- a/backend/src/routes/search.js
+++ b/backend/src/routes/search.js
@@ -1,6 +1,8 @@
 import { Router } from 'express';
 import { query } from '../services/db.js';
 import { requireAuth } from '../middleware/auth.js';
+import { parseQuery } from '../services/search/queryParser.js';
+import { search } from '../services/search/searchService.js';
 
 const router = Router();
 router.use(requireAuth);
@@ -30,230 +32,24 @@ function searchLimiter(req, res, next) {
   next();
 }
 
-// Parses a raw search string into structured operator filters and free-text
-// terms. Supports from: to: subject: has: is: after: before: in:, quoted values
-// (from:"John Smith"), and a leading '-' that negates either an operator
-// (-from:smith) or a bare word (-invoice). Filters are a list (not a map) so
-// repeated/negated operators like `from:a -from:b` are all preserved.
-export function parseSearchQuery(raw) {
-  const filters = [];
-  const terms = [];
-
-  // The leading (-?) captures optional negation. \b sits between an optional '-'
-  // and the operator name, so both `from:` and `-from:` match.
-  const opPattern = /(-?)\b(from|to|subject|has|is|after|before|in):("([^"]*)"|([\S]+))/gi;
-  const remaining = raw.replace(opPattern, (_, neg, key, _v, quoted, unquoted) => {
-    const k = key.toLowerCase();
-    const v = (quoted !== undefined ? quoted : (unquoted || '')).toLowerCase().trim();
-    if (v) filters.push({ key: k, value: v, negate: neg === '-' });
-    return ' ';
-  }).trim();
-
-  for (const word of remaining.split(/\s+/)) {
-    let w = word.trim();
-    if (!w || w === '-') continue; // skip blanks and a lone '-' (nothing to negate)
-    let negate = false;
-    if (w[0] === '-' && w.length > 1) { negate = true; w = w.slice(1); }
-    terms.push({ value: w, negate });
-  }
-
-  return { filters, terms };
-}
-
-// Wraps a positive condition so that when negated it also matches rows where the
-// underlying columns are NULL (COALESCE(..., false) treats NULL as "not a match",
-// which NOT then flips to a match — the intuitive meaning of exclusion).
-function negateCond(sql) {
-  return `NOT COALESCE((${sql}), false)`;
-}
-
-export function resolveSearchFolderScope(filters, folderParam = '') {
-  let folderScope;
-  let folderFuzzy = false; // in:<name> matches loosely; the folder param is exact
-
-  for (const f of filters) {
-    if (f.key !== 'in') continue;
-    if (f.value === 'all') { folderScope = null; }
-    else { folderScope = f.value; folderFuzzy = true; }
-  }
-
-  if (folderScope === undefined) {
-    folderScope = (folderParam || '').trim() || null;
-    folderFuzzy = false;
-  }
-
-  return { folderScope, folderFuzzy };
-}
-
-export function shouldExcludeTrashFromSearch(folderScope) {
-  return folderScope === null;
-}
-
-export function trashFolderExclusionCondition() {
-  return `NOT EXISTS (
-        SELECT 1
-        FROM folders f
-        WHERE f.account_id = m.account_id
-          AND f.path = m.folder
-          AND (f.special_use = '\\Trash'
-               OR lower(f.name) LIKE '%trash%'
-               OR lower(f.name) LIKE '%deleted%')
-      )`;
-}
-
-// Postgres refuses to build a tsvector larger than ~1MB of packed lexemes
-// (SQLSTATE 54000), so an oversized body would 500 the whole search. Cap the
-// text fed to to_tsvector at 600k chars — matching msgvault's maxFTSBodyChars
-// (internal/store/dialect_pg.go) — so one huge email can't crash the query.
-// Exported because slice 02's search_fts trigger caps the same way.
-export const FTS_BODY_CHAR_CAP = 600000;
-
-// Builds the per-term free-text OR-condition: a term matches if it appears in
-// the sender, the subject, the stored search_vector, or the length-capped body.
-// Extracted so the body cap is a single, testable source of truth.
-export function freeTextTermCondition(likeIdx, ftsIdx) {
-  return `(
-        m.from_name ILIKE $${likeIdx}
-        OR m.from_email ILIKE $${likeIdx}
-        OR m.subject ILIKE $${likeIdx}
-        OR m.search_vector @@ plainto_tsquery('english', $${ftsIdx})
-        OR to_tsvector('english', LEFT(coalesce(m.body_text,''), ${FTS_BODY_CHAR_CAP})) @@ plainto_tsquery('english', $${ftsIdx})
-      )`;
-}
-
 router.get('/', searchLimiter, async (req, res) => {
   const { q, accountId, limit = 50, offset = 0 } = req.query;
   const trimmed = (q || '').trim();
   if (!trimmed) return res.json({ messages: [] });
   if (trimmed.length > 500) return res.status(400).json({ error: 'Search query too long' });
 
-  const accountsResult = await query(
-    'SELECT id FROM email_accounts WHERE user_id = $1 AND enabled = true',
-    [req.session.userId]
-  );
-  const userAccountIds = accountsResult.rows.map(r => r.id);
-  if (!userAccountIds.length) return res.json({ messages: [] });
-
-  const targetIds = accountId && userAccountIds.includes(accountId)
-    ? [accountId] : userAccountIds;
-
-  const cap = Math.max(1, Math.min(parseInt(limit) || 50, 200));
-  const { filters, terms } = parseSearchQuery(trimmed);
-
-  const conditions = [];
-  const params = [targetIds];
-  let p = 2;
-
-  // Folder scope. `in:` in the query wins; otherwise the client-supplied `folder`
-  // param (the folder the user is currently viewing) applies. `undefined` means
-  // no in: operator was given, so we fall back to the param below.
-  //   folderScope === null   → search all folders
-  //   folderScope === string → restrict to that folder
-
-  // ── Operator filters ──────────────────────────────────────────────────────
-
-  for (const f of filters) {
-    // in: controls scope rather than adding a row condition; negation is
-    // meaningless here so it's ignored.
-    if (f.key === 'in') {
-      continue;
-    }
-
-    let cond = null;
-
-    if (f.key === 'from') {
-      params.push(`%${f.value}%`);
-      cond = `(m.from_email ILIKE $${p} OR m.from_name ILIKE $${p})`;
-      p++;
-    } else if (f.key === 'subject') {
-      params.push(`%${f.value}%`);
-      cond = `m.subject ILIKE $${p++}`;
-    } else if (f.key === 'to') {
-      // to: searches the to/cc address JSON — cast to text covers name and email
-      params.push(`%${f.value}%`);
-      cond = `(m.to_addresses::text ILIKE $${p} OR m.cc_addresses::text ILIKE $${p})`;
-      p++;
-    } else if (f.key === 'has') {
-      if (f.value === 'attachment' || f.value === 'attachments') cond = `m.has_attachments = true`;
-    } else if (f.key === 'is') {
-      if (f.value === 'unread')  cond = `m.is_read = false`;
-      else if (f.value === 'read')    cond = `m.is_read = true`;
-      else if (f.value === 'starred') cond = `m.is_starred = true`;
-    } else if (f.key === 'after') {
-      const d = new Date(f.value);
-      if (!isNaN(d)) { params.push(d.toISOString()); cond = `m.date >= $${p++}`; }
-    } else if (f.key === 'before') {
-      const d = new Date(f.value);
-      if (!isNaN(d)) { params.push(d.toISOString()); cond = `m.date < $${p++}`; }
-    }
-
-    if (cond) conditions.push(f.negate ? negateCond(cond) : cond);
-  }
-
-  // ── Free-text terms ───────────────────────────────────────────────────────
-  // Each term must match at least one of: from, subject (ILIKE — good for names
-  // and partial words), or body content (FTS — good for large text with stemming).
-  // AND between all terms: every word must appear somewhere in the email.
-  // A negated term (-word) must appear nowhere.
-
-  for (const term of terms.slice(0, 10)) {
-    if (term.value.length < 2) continue; // single-char terms are too broad and expensive
-    params.push(`%${term.value}%`); // ILIKE pattern
-    const likeIdx = p++;
-
-    params.push(term.value); // raw term for plainto_tsquery
-    const ftsIdx = p++;
-
-    const cond = freeTextTermCondition(likeIdx, ftsIdx);
-    conditions.push(term.negate ? negateCond(cond) : cond);
-  }
-
-  // Require at least one real search condition before applying folder scope, so a
-  // bare `in:inbox` (or a lone folder param) never dumps an entire folder.
-  if (!conditions.length) return res.json({ messages: [], query: q });
-
-  // Resolve folder scope: in: operator wins; otherwise use the param.
-  const { folderScope, folderFuzzy } = resolveSearchFolderScope(filters, req.query.folder || '');
-  if (folderScope) {
-    if (folderFuzzy) {
-      // in:<name> — case-insensitive match on a folder named exactly that, or a
-      // nested folder whose path ends in it (in:sent → "Sent" or "Personal/Sent").
-      // A multi-word leaf like "[Gmail]/Sent Mail" needs the quoted form in:"sent mail".
-      params.push(folderScope);
-      params.push(`%/${folderScope}`);
-      conditions.push(`(m.folder ILIKE $${p} OR m.folder ILIKE $${p + 1})`);
-      p += 2;
-    } else {
-      params.push(folderScope);
-      conditions.push(`m.folder = $${p++}`);
-    }
-  } else if (shouldExcludeTrashFromSearch(folderScope)) {
-    // Deleting moves mail into Trash, where it remains searchable by explicit
-    // folder queries like in:trash. Keep ordinary all-folder searches from
-    // resurfacing freshly-deleted messages after the optimistic UI guard expires.
-    conditions.push(trashFolderExclusionCondition());
-  }
-
-  const off = Math.max(0, parseInt(offset) || 0);
-  params.push(cap);
-  params.push(off);
-
+  const parsed = parseQuery(trimmed);
   try {
-    const result = await query(`
-      SELECT
-        m.id, m.uid, m.folder, m.subject, m.from_name, m.from_email,
-        m.date, m.snippet, m.is_read, m.is_starred, m.has_attachments, m.account_id,
-        a.name as account_name, a.email_address as account_email, a.color as account_color
-      FROM messages m
-      JOIN email_accounts a ON m.account_id = a.id
-      WHERE m.account_id = ANY($1)
-        AND m.is_deleted = false
-        AND ${conditions.join('\n        AND ')}
-      ORDER BY m.date DESC
-      LIMIT $${p} OFFSET $${p + 1}
-    `, params);
-
-    res.json({ messages: result.rows, query: q });
+    const result = await search({
+      userId: req.session.userId,
+      accountId,
+      parsed,
+      folderParam: req.query.folder || '',
+      limit,
+      offset,
+    });
+    // Response is a strict superset of the historical { messages, query }.
+    res.json({ ...result, query: q, unsupported: parsed.unsupported, errors: parsed.errors });
   } catch (err) {
     console.error('Search error:', err);
     res.status(500).json({ error: 'Search failed' });
diff --git a/backend/src/routes/search.test.js b/backend/src/routes/search.test.js
index 3ef54c7..9b305e1 100644
--- a/backend/src/routes/search.test.js
+++ b/backend/src/routes/search.test.js
@@ -1,157 +1,61 @@
-import { describe, it, expect, vi } from 'vitest';
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import express from 'express';
 
-// search.js opens a DB handle and registers auth middleware at import time;
-// neither is exercised by the pure parser under test, so stub them out.
+// The route must not hold SQL: it parses, calls the service, and JSON-frames.
+// Mock inline + import the mocked binding (avoids the vi.mock hoisting TDZ trap).
+vi.mock('../services/search/searchService.js', () => ({ search: vi.fn() }));
+vi.mock('../middleware/auth.js', () => ({ requireAuth: (req, _res, next) => { req.session = { userId: 'u1' }; next(); } }));
 vi.mock('../services/db.js', () => ({ query: vi.fn() }));
-vi.mock('../middleware/auth.js', () => ({ requireAuth: vi.fn() }));
 
-import {
-  parseSearchQuery,
-  resolveSearchFolderScope,
-  shouldExcludeTrashFromSearch,
-  trashFolderExclusionCondition,
-  freeTextTermCondition,
-  FTS_BODY_CHAR_CAP,
-} from './search.js';
-
-describe('parseSearchQuery', () => {
-  it('treats bare words as free-text terms', () => {
-    const { filters, terms } = parseSearchQuery('hello world');
-    expect(filters).toEqual([]);
-    expect(terms).toEqual([
-      { value: 'hello', negate: false },
-      { value: 'world', negate: false },
-    ]);
-  });
-
-  it('extracts positive operators and lowercases their values', () => {
-    const { filters, terms } = parseSearchQuery('from:Amazon subject:Invoice hello');
-    expect(filters).toEqual([
-      { key: 'from', value: 'amazon', negate: false },
-      { key: 'subject', value: 'invoice', negate: false },
-    ]);
-    expect(terms).toEqual([{ value: 'hello', negate: false }]);
-  });
-
-  it('supports quoted operator values with spaces', () => {
-    const { filters, terms } = parseSearchQuery('from:"John Smith" report');
-    expect(filters).toEqual([{ key: 'from', value: 'john smith', negate: false }]);
-    expect(terms).toEqual([{ value: 'report', negate: false }]);
-  });
-
-  it('negates an operator when prefixed with -', () => {
-    const { filters, terms } = parseSearchQuery('-from:Smith report');
-    expect(filters).toEqual([{ key: 'from', value: 'smith', negate: true }]);
-    expect(terms).toEqual([{ value: 'report', negate: false }]);
-  });
-
-  it('negates a free-text term when prefixed with -', () => {
-    const { filters, terms } = parseSearchQuery('report -invoice');
-    expect(filters).toEqual([]);
-    expect(terms).toEqual([
-      { value: 'report', negate: false },
-      { value: 'invoice', negate: true },
-    ]);
-  });
-
-  it('parses the in: scope operator (in:all and named folders)', () => {
-    expect(parseSearchQuery('in:all invoice').filters).toEqual([
-      { key: 'in', value: 'all', negate: false },
-    ]);
-    expect(parseSearchQuery('in:Sent proposal').filters).toEqual([
-      { key: 'in', value: 'sent', negate: false },
-    ]);
-  });
-
-  it('preserves repeated and mixed positive/negative operators', () => {
-    const { filters } = parseSearchQuery('from:alice -from:bob is:unread');
-    expect(filters).toEqual([
-      { key: 'from', value: 'alice', negate: false },
-      { key: 'from', value: 'bob', negate: true },
-      { key: 'is', value: 'unread', negate: false },
-    ]);
-  });
-
-  it('ignores a lone - so it is not treated as a negated empty term', () => {
-    const { filters, terms } = parseSearchQuery('report - draft');
-    expect(filters).toEqual([]);
-    expect(terms).toEqual([
-      { value: 'report', negate: false },
-      { value: 'draft', negate: false },
-    ]);
-  });
-
-  it('returns empty structures for blank input', () => {
-    expect(parseSearchQuery('')).toEqual({ filters: [], terms: [] });
-    expect(parseSearchQuery('    ')).toEqual({ filters: [], terms: [] });
-  });
-
-  it('does not treat unknown prefixes as operators', () => {
-    const { filters, terms } = parseSearchQuery('label:work');
-    expect(filters).toEqual([]);
-    expect(terms).toEqual([{ value: 'label:work', negate: false }]);
-  });
-
-  it('handles all supported operator keys', () => {
-    const raw = 'from:a to:b subject:c has:attachment is:starred after:2024-01-01 before:2024-12-31 in:archive';
-    const keys = parseSearchQuery(raw).filters.map(f => f.key);
-    expect(keys).toEqual(['from', 'to', 'subject', 'has', 'is', 'after', 'before', 'in']);
-  });
-
-  it('scopes search to the client folder param when no in: operator is present', () => {
-    const { filters } = parseSearchQuery('subject:newsletter');
-    expect(resolveSearchFolderScope(filters, 'INBOX')).toEqual({
-      folderScope: 'INBOX',
-      folderFuzzy: false,
+import { search } from '../services/search/searchService.js';
+import searchRouter from './search.js';
+
+function makeApp() {
+  const app = express();
+  app.use((req, _res, next) => { req.session = { userId: 'u1' }; next(); });
+  app.use('/api/search', searchRouter);
+  return app;
+}
+
+async function get(app, path) {
+  const { default: request } = await import('node:http');
+  return new Promise((resolve) => {
+    const server = app.listen(0, () => {
+      const port = server.address().port;
+      request.get(`http://127.0.0.1:${port}${path}`, (res) => {
+        let body = '';
+        res.on('data', c => (body += c));
+        res.on('end', () => { server.close(); resolve({ status: res.statusCode, json: JSON.parse(body || '{}') }); });
+      });
     });
   });
+}
 
-  it('lets in: override the client folder param', () => {
-    const { filters } = parseSearchQuery('in:trash subject:newsletter');
-    expect(resolveSearchFolderScope(filters, 'INBOX')).toEqual({
-      folderScope: 'trash',
-      folderFuzzy: true,
-    });
-  });
-
-  it('excludes trash-like folders from ordinary all-folder searches', () => {
-    const { filters } = parseSearchQuery('subject:newsletter');
-    const { folderScope } = resolveSearchFolderScope(filters);
-    expect(folderScope).toBeNull();
-    expect(shouldExcludeTrashFromSearch(folderScope)).toBe(true);
-    expect(trashFolderExclusionCondition()).toContain('NOT EXISTS');
-    expect(trashFolderExclusionCondition()).toContain('%trash%');
-    expect(trashFolderExclusionCondition()).toContain('%deleted%');
-  });
-
-  it('keeps explicit folder searches eligible to find trash messages', () => {
-    const { filters } = parseSearchQuery('in:trash subject:newsletter');
-    const { folderScope } = resolveSearchFolderScope(filters);
-    expect(shouldExcludeTrashFromSearch(folderScope)).toBe(false);
-  });
-});
+beforeEach(() => search.mockReset());
 
-describe('freeTextTermCondition (oversized-body crash hotfix)', () => {
-  it('caps the body tsvector so a multi-megabyte email cannot exceed the 1MB tsvector limit', () => {
-    const cond = freeTextTermCondition(3, 4);
-    // The body text is fed to to_tsvector through LEFT(..., FTS_BODY_CHAR_CAP).
-    // Without this cap a single oversized body raises SQLSTATE 54000 and 500s search.
-    expect(cond).toContain(
-      `to_tsvector('english', LEFT(coalesce(m.body_text,''), ${FTS_BODY_CHAR_CAP}))`
-    );
-    // Guard against a regression back to the uncapped expression.
-    expect(cond).not.toContain("to_tsvector('english', coalesce(m.body_text,'')) @@");
+describe('GET /api/search', () => {
+  it('short-circuits a blank query with { messages: [] } and never calls the service', async () => {
+    const res = await get(makeApp(), '/api/search?q=');
+    expect(res.status).toBe(200);
+    expect(res.json).toEqual({ messages: [] });
+    expect(search).not.toHaveBeenCalled();
   });
 
-  it('pins the cap at 600000 chars (msgvault maxFTSBodyChars parity)', () => {
-    expect(FTS_BODY_CHAR_CAP).toBe(600000);
+  it('rejects an over-500-char query with 400', async () => {
+    const res = await get(makeApp(), `/api/search?q=${'a'.repeat(501)}`);
+    expect(res.status).toBe(400);
   });
 
-  it('still matches sender, subject, and the stored search_vector at the given param positions', () => {
-    const cond = freeTextTermCondition(3, 4);
-    expect(cond).toContain('m.from_name ILIKE $3');
-    expect(cond).toContain('m.from_email ILIKE $3');
-    expect(cond).toContain('m.subject ILIKE $3');
-    expect(cond).toContain("m.search_vector @@ plainto_tsquery('english', $4)");
+  it('returns a superset response: messages + query + mode + page + unsupported + errors', async () => {
+    search.mockResolvedValue({ messages: [{ id: 'm1' }], mode: 'lexical', page: { offset: 0, limit: 50, hasMore: false } });
+    const res = await get(makeApp(), '/api/search?q=larger:5M%20invoice');
+    expect(res.status).toBe(200);
+    expect(res.json.messages).toEqual([{ id: 'm1' }]);
+    expect(res.json.query).toBe('larger:5M invoice');
+    expect(res.json.mode).toBe('lexical');
+    expect(res.json.page).toEqual({ offset: 0, limit: 50, hasMore: false });
+    // larger: is recognized but unserviceable — surfaced, not silently dropped.
+    expect(res.json.unsupported).toEqual([{ key: 'larger', token: 'larger:5m' }]);
+    expect(res.json.errors).toEqual([]);
   });
 });
diff --git a/backend/src/services/backgroundJobs.js b/backend/src/services/backgroundJobs.js
new file mode 100644
index 0000000..799b05e
--- /dev/null
+++ b/backend/src/services/backgroundJobs.js
@@ -0,0 +1,26 @@
+import { query } from './db.js';
+
+// Upsert a drainer's progress row. One row per (kind, account); global jobs
+// pass accountId = null. started_at is set once and preserved across updates.
+export async function upsertJob({ kind, accountId = null, state, processed = 0, total = 0, lastError = null }) {
+  await query(`
+    INSERT INTO background_jobs (kind, account_id, state, processed, total, last_error, started_at, updated_at)
+    VALUES ($1, $2, $3, $4, $5, $6, NOW(), NOW())
+    ON CONFLICT (kind, COALESCE(account_id::text, '')) DO UPDATE SET
+      state = EXCLUDED.state,
+      processed = EXCLUDED.processed,
+      total = EXCLUDED.total,
+      last_error = EXCLUDED.last_error,
+      started_at = COALESCE(background_jobs.started_at, EXCLUDED.started_at),
+      updated_at = NOW()
+  `, [kind, accountId, state, processed, total, lastError]);
+}
+
+export async function listJobs() {
+  const { rows } = await query(
+    `SELECT kind, account_id, state, processed, total, last_error, started_at, updated_at
+       FROM background_jobs
+      ORDER BY updated_at DESC`
+  );
+  return rows;
+}
diff --git a/backend/src/services/backgroundJobs.test.js b/backend/src/services/backgroundJobs.test.js
new file mode 100644
index 0000000..f3ecfd2
--- /dev/null
+++ b/backend/src/services/backgroundJobs.test.js
@@ -0,0 +1,25 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+vi.mock('./db.js', () => ({ query: vi.fn() }));
+import { query } from './db.js';
+import { upsertJob, listJobs } from './backgroundJobs.js';
+
+beforeEach(() => query.mockReset());
+
+describe('backgroundJobs', () => {
+  it('upserts one row per (kind, account) via the COALESCE unique index', async () => {
+    query.mockResolvedValue({ rows: [] });
+    await upsertJob({ kind: 'fts_backfill', state: 'running', processed: 10, total: 100 });
+    const [sql, params] = query.mock.calls[0];
+    expect(sql).toContain('INSERT INTO background_jobs');
+    expect(sql).toContain("ON CONFLICT (kind, COALESCE(account_id::text, ''))");
+    expect(sql).toContain('started_at = COALESCE(background_jobs.started_at, EXCLUDED.started_at)');
+    expect(params).toEqual(['fts_backfill', null, 'running', 10, 100, null]);
+  });
+
+  it('lists jobs newest-first', async () => {
+    query.mockResolvedValue({ rows: [{ kind: 'fts_backfill' }] });
+    const jobs = await listJobs();
+    expect(jobs).toEqual([{ kind: 'fts_backfill' }]);
+    expect(query.mock.calls[0][0]).toContain('ORDER BY updated_at DESC');
+  });
+});
diff --git a/backend/src/services/bodyBackfill.js b/backend/src/services/bodyBackfill.js
new file mode 100644
index 0000000..ccb855b
--- /dev/null
+++ b/backend/src/services/bodyBackfill.js
@@ -0,0 +1,181 @@
+import { query } from './db.js';
+import { providerProfile } from './imapManager.js';
+
+// Body-materialization drainer. Fills messages.body_text/body_html over IMAP so weight-D
+// lexical search has material to work with, WITHOUT growing the imapManager
+// singleton (README invariant) and WITHOUT touching throttle-hostile providers. ALL policy —
+// the scan predicate, batching, pacing, provider gating, session cap, quiet-window
+// backpressure, and the per-account circuit breaker — lives here. The singleton exposes only
+// the narrow fetchBodiesForMessages() entry point this module calls.
+//
+// Shape is copied from imapManager.startSnippetIndexer, but this module is standalone: its
+// per-account run guard and circuit-breaker state are module-level (not instance fields), and
+// its IMAP/quiet-window/progress/clock dependencies are injected so it unit-tests against a
+// fake fetchBodies with no real IMAP.
+
+const BATCH_SIZE = 50;              // messages fetched per batch
+const MIN_BATCH_DELAY_MS = 2000;    // floor between batches (provider batchDelay may raise it)
+const MAX_BATCHES_PER_RUN = 200;    // session cap: 10,000 messages per run, then resume next kick
+const QUIET_WINDOW_MS = 8000;       // pause extra when the user opened a message this recently
+const MAX_CONSECUTIVE_ERRORS = 3;   // abort the run after this many failing batches in a row
+const BACKOFF_BASE_MS = 10 * 60 * 1000;    // first circuit-breaker back-off
+const BACKOFF_MAX_MS = 2 * 60 * 60 * 1000; // cap
+
+// Per-account run guard and circuit breaker. Module-level so a single process runs at most one
+// body drainer per account (the snippet indexer uses instance fields for the same purpose).
+const running = new Set();          // accountId
+const backoff = new Map();          // accountId -> { failures, until }
+
+// Test seam: reset module state between unit tests.
+export function resetBodyBackfillState() {
+  running.clear();
+  backoff.clear();
+}
+
+const realSleep = (ms) => new Promise((r) => setTimeout(r, ms));
+
+// Core drainer. Dependencies are injected so tests supply a fake fetchBodies:
+//   fetchBodies(accountId, ids)  -> Promise<{ fetched: number }>  (writes bodies; throws on a
+//                                   connection-level failure so this loop can back off)
+//   getLastActivityMs(accountId) -> number   (ms timestamp of the user's last live body open)
+//   isSnippetIndexerRunning(accountId) -> boolean  (the snippet indexer does its own sustained
+//                                   BODY[]-ish fetches over an independent connection; running
+//                                   both loops for the same account at once double-books IMAP
+//                                   connections against the same provider limit)
+//   upsertJobProgress({ accountId, kind, processed, total, state }) -> Promise  (this is
+//                                   backgroundJobs.upsertJob's own shape — callers inject it
+//                                   directly; the drainer speaks its vocabulary, no adapter)
+//   sleep(ms) -> Promise    (injectable for fast, deterministic tests)
+//   now() -> number         (injectable clock)
+export async function startBodyBackfill(account, deps) {
+  const {
+    fetchBodies,
+    getLastActivityMs = () => 0,
+    isSnippetIndexerRunning = () => false,
+    upsertJobProgress = async () => {},
+    sleep = realSleep,
+    now = Date.now,
+  } = deps;
+
+  const cfg = providerProfile(account);
+  if (!cfg.bodyBackfill) return;                    // provider gate (Gmail/PurelyMail/etc.)
+  if (running.has(account.id)) return;              // one drainer per account
+  const bo = backoff.get(account.id);
+  if (bo && now() < bo.until) return;               // circuit breaker open
+  running.add(account.id);
+
+  const batchDelay = Math.max(cfg.batchDelay || 0, MIN_BATCH_DELAY_MS);
+  const kind = 'body_backfill';
+  let batchCount = 0;
+  let failed = false;
+
+  try {
+    // Denominator + starting coverage for progress. total is the whole eligible mailbox so the
+    // progress bar reflects real coverage, not just this run.
+    const countRes = await query(
+      `SELECT count(*)::int AS total,
+              count(*) FILTER (WHERE body_text IS NOT NULL OR body_html IS NOT NULL)::int AS have_body
+       FROM messages WHERE account_id = $1 AND is_deleted = false`,
+      [account.id]
+    );
+    const total = countRes.rows[0].total;
+    let haveBody = countRes.rows[0].have_body;
+    if (total - haveBody <= 0) {
+      await upsertJobProgress({ accountId: account.id, kind, processed: haveBody, total, state: 'done' });
+      return;
+    }
+
+    // Keyset by id ascending: id is a non-null, unique UUID, so the cursor advances past every
+    // row exactly once per run — including rows whose body cannot be fetched (they stay NULL but
+    // are not re-selected this run, so the run always terminates). Recency ordering is
+    // intentionally not used: body coverage is a background quality lever.
+    let cursorId = null;
+    let consecutiveErrors = 0;
+
+    while (true) {
+      // Stop if the account was deleted mid-run.
+      const alive = await query('SELECT id FROM email_accounts WHERE id = $1', [account.id]);
+      if (!alive.rows.length) return;
+
+      // Defer to the snippet indexer rather than double-book BODY[] IMAP connections for the
+      // same account — checked every iteration (not just at kick time) so a session already in
+      // progress backs off cleanly the moment the indexer starts, instead of racing it.
+      if (isSnippetIndexerRunning(account.id)) {
+        await upsertJobProgress({ accountId: account.id, kind, processed: haveBody, total, state: 'deferred' });
+        return;
+      }
+
+      if (batchCount >= MAX_BATCHES_PER_RUN) {
+        await upsertJobProgress({ accountId: account.id, kind, processed: haveBody, total, state: 'paused' });
+        return;
+      }
+
+      const batchRes = await query(
+        `SELECT id FROM messages
+         WHERE account_id = $1 AND body_text IS NULL AND body_html IS NULL AND is_deleted = false
+           AND ($2::uuid IS NULL OR id > $2)
+         ORDER BY id ASC
+         LIMIT $3`,
+        [account.id, cursorId, BATCH_SIZE]
+      );
+      if (!batchRes.rows.length) break; // drained
+      const ids = batchRes.rows.map((r) => r.id);
+
+      try {
+        const { fetched } = await fetchBodies(account.id, ids);
+        haveBody += fetched;
+        cursorId = ids[ids.length - 1]; // advance only on success
+        batchCount++;
+        consecutiveErrors = 0;
+      } catch {
+        consecutiveErrors++;
+        await sleep(cfg.errorDelay || batchDelay);
+        if (consecutiveErrors >= MAX_CONSECUTIVE_ERRORS) {
+          failed = true;
+          // Write a terminal 'error' state (ftsBackfill.js convention) so background_jobs
+          // doesn't keep showing a stale 'running' row for an account that's actually stalled.
+          await upsertJobProgress({ accountId: account.id, kind, processed: haveBody, total, state: 'error' });
+          return; // finally trips the circuit breaker
+        }
+        continue; // retry the same batch (cursor not advanced); fetchBodies reopens IMAP
+      }
+
+      await upsertJobProgress({ accountId: account.id, kind, processed: haveBody, total, state: 'running' });
+
+      // Quiet-window backpressure: pause longer when the user is actively opening messages so
+      // background BODY[] traffic doesn't compete with click-time fetches.
+      const quietFor = now() - getLastActivityMs(account.id);
+      const extraDelay = quietFor < QUIET_WINDOW_MS ? QUIET_WINDOW_MS - quietFor : 0;
+      await sleep(batchDelay + extraDelay);
+    }
+
+    await upsertJobProgress({ accountId: account.id, kind, processed: haveBody, total, state: 'done' });
+  } catch {
+    failed = true;
+  } finally {
+    running.delete(account.id);
+    // Circuit breaker: a run that failed without draining a single batch (e.g. the provider
+    // refuses the extra connection) backs off exponentially so we stop reopening competing IMAP
+    // connections. Any progress — or a clean finish — clears the backoff.
+    if (failed && batchCount === 0) {
+      const failures = (backoff.get(account.id)?.failures || 0) + 1;
+      const delay = Math.min(BACKOFF_BASE_MS * 2 ** (failures - 1), BACKOFF_MAX_MS);
+      backoff.set(account.id, { failures, until: now() + delay });
+    } else {
+      backoff.delete(account.id);
+    }
+  }
+}
+
+// Convenience wrapper for the reindex route: builds the injected deps from the imapManager
+// singleton and the injected progress sink (backgroundJobs.upsertJob — the drainer already
+// emits its shape), then runs the core drainer. Kept here (not in the route) so the wiring is
+// unit-testable and bodyBackfill.js never imports backgroundJobs.js directly.
+export function startAccountBodyBackfill(account, imapManager, upsertJobProgress) {
+  return startBodyBackfill(account, {
+    fetchBodies: (accountId, ids) => imapManager.fetchBodiesForMessages(accountId, ids),
+    getLastActivityMs: (accountId) => imapManager.lastUserActivity.get(accountId) || 0,
+    isSnippetIndexerRunning: (accountId) => imapManager.snippetIndexerRunning.has(accountId),
+    upsertJobProgress,
+  });
+}
diff --git a/backend/src/services/bodyBackfill.test.js b/backend/src/services/bodyBackfill.test.js
new file mode 100644
index 0000000..3b3da49
--- /dev/null
+++ b/backend/src/services/bodyBackfill.test.js
@@ -0,0 +1,290 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+
+vi.mock('./db.js', () => ({ query: vi.fn() }));
+vi.mock('./imapManager.js', () => ({ providerProfile: vi.fn() }));
+
+import { query } from './db.js';
+import { providerProfile } from './imapManager.js';
+import { startBodyBackfill, startAccountBodyBackfill, resetBodyBackfillState } from './bodyBackfill.js';
+
+const account = { id: 'acct-1', imap_host: 'imap.icloud.com' };
+const enabledProfile = { bodyBackfill: true, batchDelay: 2000, errorDelay: 100 };
+
+// Instant, assertable deps so the drainer runs with no wall-clock waits.
+function baseDeps(overrides = {}) {
+  return {
+    fetchBodies: vi.fn().mockResolvedValue({ fetched: 0 }),
+    getLastActivityMs: () => 0,
+    isSnippetIndexerRunning: () => false,
+    upsertJobProgress: vi.fn().mockResolvedValue(),
+    sleep: vi.fn().mockResolvedValue(),
+    now: () => 1_000_000,
+    ...overrides,
+  };
+}
+
+// Route the mocked query by SQL shape. `batches` is an array of id-arrays returned in order;
+// once exhausted it returns an empty batch (drained).
+function mockQuery({ total = 0, haveBody = 0, batches = [], alive = true } = {}) {
+  let i = 0;
+  query.mockImplementation((sql) => {
+    if (/AS total/.test(sql)) return Promise.resolve({ rows: [{ total, have_body: haveBody }] });
+    if (/FROM email_accounts/.test(sql)) return Promise.resolve({ rows: alive ? [{ id: 'acct-1' }] : [] });
+    if (/body_text IS NULL/.test(sql)) {
+      const rows = (batches[i++] || []).map((id) => ({ id }));
+      return Promise.resolve({ rows });
+    }
+    return Promise.resolve({ rows: [] });
+  });
+}
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  resetBodyBackfillState();
+});
+
+describe('startBodyBackfill — provider gate', () => {
+  it('no-ops for a provider with bodyBackfill:false and never queries', async () => {
+    providerProfile.mockReturnValue({ bodyBackfill: false });
+    const deps = baseDeps();
+    await startBodyBackfill(account, deps);
+    expect(deps.fetchBodies).not.toHaveBeenCalled();
+    expect(query).not.toHaveBeenCalled();
+  });
+});
+
+describe('startBodyBackfill — nothing to do', () => {
+  it('reports complete and fetches nothing when coverage is already full', async () => {
+    providerProfile.mockReturnValue(enabledProfile);
+    mockQuery({ total: 100, haveBody: 100 });
+    const deps = baseDeps();
+    await startBodyBackfill(account, deps);
+    expect(deps.fetchBodies).not.toHaveBeenCalled();
+    expect(deps.upsertJobProgress).toHaveBeenCalledWith(
+      expect.objectContaining({ kind: 'body_backfill', state: 'done', processed: 100, total: 100 })
+    );
+  });
+});
+
+describe('startBodyBackfill — batching + keyset + progress', () => {
+  it('drains 120 messages in batches of 50/50/20 and advances the cursor', async () => {
+    providerProfile.mockReturnValue(enabledProfile);
+    const b1 = Array.from({ length: 50 }, (_, k) => `a${k}`);
+    const b2 = Array.from({ length: 50 }, (_, k) => `b${k}`);
+    const b3 = Array.from({ length: 20 }, (_, k) => `c${k}`);
+    mockQuery({ total: 120, haveBody: 0, batches: [b1, b2, b3] });
+    const deps = baseDeps({ fetchBodies: vi.fn().mockImplementation((_id, ids) => Promise.resolve({ fetched: ids.length })) });
+
+    await startBodyBackfill(account, deps);
+
+    expect(deps.fetchBodies).toHaveBeenCalledTimes(3);
+    expect(deps.fetchBodies).toHaveBeenNthCalledWith(1, 'acct-1', b1);
+    expect(deps.fetchBodies).toHaveBeenNthCalledWith(2, 'acct-1', b2);
+    expect(deps.fetchBodies).toHaveBeenNthCalledWith(3, 'acct-1', b3);
+    // Cursor advances to the last id of the previous batch (keyset param $2).
+    const scanCalls = query.mock.calls.filter(([sql]) => /body_text IS NULL/.test(sql));
+    expect(scanCalls[0][1][1]).toBeNull();      // first scan: cursor null
+    expect(scanCalls[1][1][1]).toBe('a49');     // second scan: last id of batch 1
+    expect(scanCalls[2][1][1]).toBe('b49');     // third scan: last id of batch 2
+    expect(deps.upsertJobProgress).toHaveBeenLastCalledWith(
+      expect.objectContaining({ state: 'done', processed: 120, total: 120 })
+    );
+  });
+});
+
+describe('startBodyBackfill — session cap + resume', () => {
+  it('pauses after MAX_BATCHES_PER_RUN batches, then resumes on the next call', async () => {
+    providerProfile.mockReturnValue(enabledProfile);
+    // Never drains: every scan returns 50 fresh ids so the run hits the batch cap.
+    let n = 0;
+    query.mockImplementation((sql) => {
+      if (/AS total/.test(sql)) return Promise.resolve({ rows: [{ total: 100000, have_body: 0 }] });
+      if (/FROM email_accounts/.test(sql)) return Promise.resolve({ rows: [{ id: 'acct-1' }] });
+      if (/body_text IS NULL/.test(sql)) return Promise.resolve({ rows: Array.from({ length: 50 }, () => ({ id: `id-${n++}` })) });
+      return Promise.resolve({ rows: [] });
+    });
+    const deps = baseDeps({ fetchBodies: vi.fn().mockResolvedValue({ fetched: 50 }) });
+
+    await startBodyBackfill(account, deps);
+    expect(deps.fetchBodies).toHaveBeenCalledTimes(200); // MAX_BATCHES_PER_RUN
+    expect(deps.upsertJobProgress).toHaveBeenLastCalledWith(
+      expect.objectContaining({ state: 'paused' })
+    );
+
+    // Resume: the run guard was released in finally, so a second call proceeds (not gated).
+    await startBodyBackfill(account, deps);
+    expect(deps.fetchBodies).toHaveBeenCalledTimes(400);
+  });
+});
+
+describe('startBodyBackfill — single drainer per account', () => {
+  it('a second concurrent call for the same account is a no-op', async () => {
+    providerProfile.mockReturnValue(enabledProfile);
+    mockQuery({ total: 50, haveBody: 0, batches: [['x1']] });
+    const deps = baseDeps({ fetchBodies: vi.fn().mockResolvedValue({ fetched: 1 }) });
+
+    // Both are invoked synchronously in the same tick: the first adds the account to the
+    // run guard before its first await, so the second sees it and returns immediately.
+    const p1 = startBodyBackfill(account, deps);
+    const p2 = startBodyBackfill(account, deps);
+    await Promise.all([p1, p2]);
+
+    expect(deps.fetchBodies).toHaveBeenCalledTimes(1); // only p1 did work
+  });
+});
+
+describe('startBodyBackfill — circuit breaker trip + recovery', () => {
+  it('trips after 3 consecutive batch errors, blocks retries in the window, recovers after it', async () => {
+    providerProfile.mockReturnValue(enabledProfile);
+    mockQuery({ total: 100, haveBody: 0, batches: [['x1'], ['x1'], ['x1']] });
+    let clock = 1_000_000;
+    const failing = baseDeps({
+      fetchBodies: vi.fn().mockRejectedValue(new Error('Command failed')),
+      now: () => clock,
+    });
+
+    // Trip: 3 consecutive rejections abort the run and open the breaker.
+    await startBodyBackfill(account, failing);
+    expect(failing.fetchBodies).toHaveBeenCalledTimes(3);
+    // A terminal 'error' progress row is written so background_jobs doesn't keep showing a
+    // stale 'running' state for an account that's actually stalled.
+    expect(failing.upsertJobProgress).toHaveBeenLastCalledWith(
+      expect.objectContaining({ kind: 'body_backfill', state: 'error' })
+    );
+
+    // Blocked: a call inside the back-off window no-ops (breaker open).
+    const blocked = baseDeps({ fetchBodies: vi.fn(), now: () => clock });
+    await startBodyBackfill(account, blocked);
+    expect(blocked.fetchBodies).not.toHaveBeenCalled();
+
+    // Recover: advance past the back-off window; a fresh call runs again.
+    clock += 3 * 60 * 60 * 1000; // beyond BACKOFF_MAX_MS (2h)
+    mockQuery({ total: 100, haveBody: 0, batches: [['y1']] });
+    const recovered = baseDeps({ fetchBodies: vi.fn().mockResolvedValue({ fetched: 1 }), now: () => clock });
+    await startBodyBackfill(account, recovered);
+    expect(recovered.fetchBodies).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe('startBodyBackfill — quiet-window backpressure', () => {
+  it('adds the remaining quiet window to the post-batch delay when the user is active', async () => {
+    providerProfile.mockReturnValue(enabledProfile);
+    mockQuery({ total: 1, haveBody: 0, batches: [['m1']] });
+    const NOW = 5_000_000;
+    const deps = baseDeps({
+      fetchBodies: vi.fn().mockResolvedValue({ fetched: 1 }),
+      now: () => NOW,
+      getLastActivityMs: () => NOW - 2000, // user opened a message 2s ago
+    });
+
+    await startBodyBackfill(account, deps);
+
+    // batchDelay = max(2000, 2000) = 2000; quiet window 8000 - 2000 elapsed = 6000 extra.
+    expect(deps.sleep).toHaveBeenCalledWith(8000);
+  });
+});
+
+describe('startBodyBackfill — defers to the snippet indexer', () => {
+  it('gates the kick: never fetches when the indexer is already running', async () => {
+    providerProfile.mockReturnValue(enabledProfile);
+    mockQuery({ total: 100, haveBody: 0, batches: [['x1']] });
+    const deps = baseDeps({ isSnippetIndexerRunning: () => true });
+
+    await startBodyBackfill(account, deps);
+
+    expect(deps.fetchBodies).not.toHaveBeenCalled();
+    expect(deps.upsertJobProgress).toHaveBeenCalledWith(
+      expect.objectContaining({ kind: 'body_backfill', state: 'deferred' })
+    );
+  });
+
+  it('re-checks every iteration: defers mid-session once the indexer starts, without tripping the breaker', async () => {
+    providerProfile.mockReturnValue(enabledProfile);
+    const b1 = ['a1'];
+    const b2 = ['b1'];
+    mockQuery({ total: 100, haveBody: 0, batches: [b1, b2] });
+    let indexerRunning = false;
+    const deps = baseDeps({
+      fetchBodies: vi.fn().mockImplementation(() => {
+        indexerRunning = true; // simulate the snippet indexer kicking in right after batch 1
+        return Promise.resolve({ fetched: 1 });
+      }),
+      isSnippetIndexerRunning: () => indexerRunning,
+    });
+
+    await startBodyBackfill(account, deps);
+
+    // Batch 1 runs (indexer not yet running); batch 2's scan never happens because the
+    // re-check at the top of the next iteration sees the indexer is now running and bails.
+    expect(deps.fetchBodies).toHaveBeenCalledTimes(1);
+    expect(deps.upsertJobProgress).toHaveBeenLastCalledWith(
+      expect.objectContaining({ state: 'deferred' })
+    );
+
+    // Deferring is not a failure: the circuit breaker must not be armed by it.
+    const resumed = baseDeps({ fetchBodies: vi.fn().mockResolvedValue({ fetched: 1 }) });
+    mockQuery({ total: 100, haveBody: 1, batches: [['c1']] });
+    await startBodyBackfill(account, resumed);
+    expect(resumed.fetchBodies).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe('startBodyBackfill — forward progress', () => {
+  it('drains to completion even when every batch reports {fetched:0} (resolved, not rejected)', async () => {
+    providerProfile.mockReturnValue(enabledProfile);
+    const b1 = Array.from({ length: 50 }, (_, k) => `a${k}`);
+    const b2 = Array.from({ length: 10 }, (_, k) => `b${k}`);
+    mockQuery({ total: 60, haveBody: 0, batches: [b1, b2] });
+    const deps = baseDeps({ fetchBodies: vi.fn().mockResolvedValue({ fetched: 0 }) });
+
+    await startBodyBackfill(account, deps);
+
+    // The cursor still advances on a resolved (non-throwing) call even when it wrote nothing —
+    // e.g. every message in the batch legitimately has no fetchable body — so the run still
+    // reaches completion instead of looping forever or being mistaken for a failure.
+    expect(deps.fetchBodies).toHaveBeenCalledTimes(2);
+    expect(deps.upsertJobProgress).toHaveBeenLastCalledWith(
+      expect.objectContaining({ state: 'done', processed: 0, total: 60 })
+    );
+  });
+});
+
+describe('startAccountBodyBackfill — adapter wiring', () => {
+  it('wires imapManager.fetchBodiesForMessages and lastUserActivity into the core drainer', async () => {
+    vi.useFakeTimers();
+    providerProfile.mockReturnValue(enabledProfile);
+    mockQuery({ total: 1, haveBody: 0, batches: [['m1']] });
+    const fakeImap = {
+      fetchBodiesForMessages: vi.fn().mockResolvedValue({ fetched: 1 }),
+      lastUserActivity: new Map([['acct-1', Date.now()]]),
+      snippetIndexerRunning: new Set(),
+    };
+    const upsert = vi.fn().mockResolvedValue();
+
+    const p = startAccountBodyBackfill(account, fakeImap, upsert);
+    await vi.runAllTimersAsync(); // flush the real setTimeout used by the default sleep
+    await p;
+
+    expect(fakeImap.fetchBodiesForMessages).toHaveBeenCalledWith('acct-1', ['m1']);
+    expect(upsert).toHaveBeenCalledWith(
+      expect.objectContaining({ kind: 'body_backfill', accountId: 'acct-1' })
+    );
+    vi.useRealTimers();
+  });
+
+  it('wires imapManager.snippetIndexerRunning so an already-running indexer defers the kick', async () => {
+    providerProfile.mockReturnValue(enabledProfile);
+    mockQuery({ total: 1, haveBody: 0, batches: [['m1']] });
+    const fakeImap = {
+      fetchBodiesForMessages: vi.fn().mockResolvedValue({ fetched: 1 }),
+      lastUserActivity: new Map(),
+      snippetIndexerRunning: new Set(['acct-1']),
+    };
+    const upsert = vi.fn().mockResolvedValue();
+
+    await startAccountBodyBackfill(account, fakeImap, upsert);
+
+    expect(fakeImap.fetchBodiesForMessages).not.toHaveBeenCalled();
+    expect(upsert).toHaveBeenCalledWith(expect.objectContaining({ state: 'deferred' }));
+  });
+});
diff --git a/backend/src/services/imapManager.js b/backend/src/services/imapManager.js
index ed83496..f621942 100644
--- a/backend/src/services/imapManager.js
+++ b/backend/src/services/imapManager.js
@@ -418,6 +418,10 @@ function safeDate(d) {
 // flagPollEveryTicks:  for non-push flag providers, poll flags every N successful sync ticks.
 // snippetIndex:        run the background snippet indexer after backfill.
 //                      Disabled for providers that throttle body fetches too aggressively.
+// bodyBackfill:        run the background body-materialization drainer
+//                      (services/bodyBackfill.js) for this provider. Enabled only for
+//                      providers that tolerate sustained background BODY[] fetches;
+//                      Gmail/PurelyMail/Microsoft stay off in v1 (unproven / throttle-hostile).
 // skipFolderPatterns:  folder path substrings to skip during backfill (label-view dedup).
 // skipFolderNames:     exact folder paths to skip (non-selectable namespace containers).
 // batchSize/Delay/errorDelay/batchesPerConn: backfill rate-limit tuning.
@@ -432,6 +436,7 @@ const PROVIDERS = {
     fetchBody: false,
     pushesFlags: false,
     snippetIndex: false,
+    bodyBackfill: false,
     speculativeFetch: false,
     skipFolderPatterns: ['all mail', '[gmail]/starred', '[gmail]/important'],
     // [Gmail] is a namespace container — not a selectable mailbox. It must be
@@ -443,6 +448,7 @@ const PROVIDERS = {
     fetchBody: false,
     pushesFlags: true,
     snippetIndex: true,
+    bodyBackfill: true,
     speculativeFetch: false,
     skipFolderPatterns: [],
     skipFolderNames: [],
@@ -453,6 +459,7 @@ const PROVIDERS = {
     fetchBody: false,
     pushesFlags: true,
     snippetIndex: true,
+    bodyBackfill: true,
     speculativeFetch: true,
     skipFolderPatterns: [],
     skipFolderNames: [],
@@ -462,6 +469,7 @@ const PROVIDERS = {
     fetchBody: false,
     pushesFlags: true,
     snippetIndex: true,
+    bodyBackfill: false,
     speculativeFetch: true,
     skipFolderPatterns: [],
     skipFolderNames: [],
@@ -489,6 +497,7 @@ const PROVIDERS = {
     usesIdle: false,
     pushesFlags: false,
     snippetIndex: false,
+    bodyBackfill: false,
     speculativeFetch: false,
     preferFreshBodyFetch: true,
     freshInboxSync: true,
@@ -507,6 +516,7 @@ const PROVIDERS = {
     fetchBody: false,
     pushesFlags: true,
     snippetIndex: true,
+    bodyBackfill: true,
     speculativeFetch: true,
     skipFolderPatterns: [],
     skipFolderNames: [],
@@ -3723,6 +3733,30 @@ export class ImapManager {
   // Called in the background (via setImmediate) so it doesn't block the sync path.
   // By the time the user clicks the email (typically 2–10s later), the body is already
   // in the DB and the click returns instantly without a live IMAP round-trip.
+  // Shared fetch+sanitize+snippet+UPDATE step behind every body-materialization path
+  // (opportunistic new-mail prefetch, on-view folder prefetch, and the body-backfill
+  // drainer). Kept as one helper so the UPDATE shape that fires the search_fts
+  // triggers never drifts between callers. Returns true if a body was actually written.
+  // Private (underscore-prefixed, matching this file's convention for internal helpers like
+  // _enqueueFlagPush) — not new public surface, so it doesn't count against the
+  // one-narrow-method invariant that governs fetchBodiesForMessages.
+  async _fetchAndStoreBody(account, msg) {
+    const { html, text, attachments } = await this.fetchMessageBody(
+      account, msg.uid, msg.folder || 'INBOX'
+    );
+    const safeHtml = html ? sanitizeEmail(html) : null;
+    if (!safeHtml && !text) return false;
+    const snip = snippetFromBody(text, safeHtml || html);
+    await query(
+      `UPDATE messages
+       SET body_html = $1, body_text = $2, attachments = $3,
+           snippet = CASE WHEN $5 != '' THEN $5 ELSE snippet END
+       WHERE id = $4`,
+      [sanitizeStr(safeHtml), sanitizeStr(text), JSON.stringify(attachments || []), msg.id, sanitizeStr(snip)]
+    );
+    return true;
+  }
+
   async prefetchNewMessageBodies(account, messages) {
     for (const msg of messages) {
       try {
@@ -3733,26 +3767,58 @@ export class ImapManager {
         );
         if (existing.rows.length) continue;
 
-        const { html, text, attachments } = await this.fetchMessageBody(
-          account, msg.uid, msg.folder || 'INBOX'
-        );
-        const safeHtml = html ? sanitizeEmail(html) : null;
-        if (safeHtml || text) {
-          const snip = snippetFromBody(text, safeHtml || html);
-          await query(
-            `UPDATE messages
-             SET body_html = $1, body_text = $2, attachments = $3,
-                 snippet = CASE WHEN $5 != '' THEN $5 ELSE snippet END
-             WHERE id = $4`,
-            [sanitizeStr(safeHtml), sanitizeStr(text), JSON.stringify(attachments || []), msg.id, sanitizeStr(snip)]
-          );
-        }
+        await this._fetchAndStoreBody(account, msg);
       } catch (err) {
         console.warn(`Body prefetch failed for uid ${msg.uid}:`, err.message);
       }
     }
   }
 
+  // Narrow entry point for the body-materialization drainer (services/bodyBackfill.js).
+  // Given a set of message IDs, fetch each body over IMAP and persist it with the same
+  // UPDATE the prefetch path uses. This is the ONLY method the drainer calls into the
+  // singleton — all batching, pacing, provider gating, and progress live in bodyBackfill.js
+  // (imapManager invariant, specs/search-overhaul/README.md). Returns the number of bodies
+  // actually written.
+  //
+  // A single poison message (deleted/renamed folder, a per-message server-side rejection)
+  // must not wedge the whole batch — bodyBackfill.js's keyset cursor only advances past ids
+  // this method actually processed, and its circuit breaker only trips on a batch that made
+  // zero progress. So each message's fetch+store is individually caught and skipped here;
+  // only when the WHOLE batch fails (zero successes and at least one error — almost always a
+  // dead connection, not a bad message) do we rethrow, so the drainer's breaker can still back
+  // a genuinely broken account off.
+  async fetchBodiesForMessages(accountId, messageIds) {
+    if (!messageIds.length) return { fetched: 0 };
+
+    const accountResult = await query('SELECT * FROM email_accounts WHERE id = $1', [accountId]);
+    if (!accountResult.rows.length) return { fetched: 0 };
+    const account = accountResult.rows[0];
+
+    // Only touch rows that still lack a body — a concurrent open/prefetch may have filled
+    // some since the drainer selected them.
+    const rowsResult = await query(
+      `SELECT id, uid, folder FROM messages
+       WHERE id = ANY($1::uuid[]) AND body_html IS NULL AND body_text IS NULL`,
+      [messageIds]
+    );
+
+    let fetched = 0;
+    let errors = 0;
+    let lastError = null;
+    for (const msg of rowsResult.rows) {
+      try {
+        if (await this._fetchAndStoreBody(account, msg)) fetched++;
+      } catch (err) {
+        errors++;
+        lastError = err;
+        console.warn(`Body backfill: skipped message ${msg.id} (uid ${msg.uid}):`, err.message);
+      }
+    }
+    if (fetched === 0 && errors > 0) throw lastError;
+    return { fetched };
+  }
+
   // Background body prefetch for messages currently visible in a folder.
   // Called after GET /messages responds so the user gets a fast first impression
   // without waiting for this work. Respects the quiet window — pauses between
@@ -3786,18 +3852,7 @@ export class ImapManager {
         );
         if (existing.rows.length) continue;
 
-        const { html, text, attachments } = await this.fetchMessageBody(account, msg.uid, msg.folder);
-        const safeHtml = html ? sanitizeEmail(html) : null;
-        if (safeHtml || text) {
-          const snip = snippetFromBody(text, safeHtml || html);
-          await query(
-            `UPDATE messages
-             SET body_html = $1, body_text = $2, attachments = $3,
-                 snippet = CASE WHEN $5 != '' THEN $5 ELSE snippet END
-             WHERE id = $4`,
-            [sanitizeStr(safeHtml), sanitizeStr(text), JSON.stringify(attachments || []), msg.id, sanitizeStr(snip)]
-          );
-        }
+        await this._fetchAndStoreBody(account, msg);
       } catch (err) {
         console.warn(`Folder body prefetch failed for uid ${msg.uid}:`, err.message);
       }
diff --git a/backend/src/services/imapManager.test.js b/backend/src/services/imapManager.test.js
index 1540298..5120364 100644
--- a/backend/src/services/imapManager.test.js
+++ b/backend/src/services/imapManager.test.js
@@ -15,7 +15,8 @@ import { ImapManager, providerProfile, makeClientCfg, gtdRelocateGuard, insertCo
 import { query } from './db.js';
 import { invalidateGtdConfigCache } from './gtdConfig.js';
 import { runGtdTransitions, threadKeysInFolders } from './gtdTransitions.js';
-import { parseMessage } from './messageParser.js';
+import { sanitizeEmail } from './emailSanitizer.js';
+import { parseMessage, snippetFromBody } from './messageParser.js';
 
 const account = (imap_host, oauth_provider = null) => ({ imap_host, oauth_provider });
 
@@ -104,6 +105,22 @@ describe('providerProfile — host detection', () => {
   });
 });
 
+// ── providerProfile — bodyBackfill allowlist ─────────────────────────────────
+
+describe('providerProfile — bodyBackfill allowlist', () => {
+  it('enables body backfill for well-behaved providers (apple, yahoo, generic)', () => {
+    expect(providerProfile(account('imap.icloud.com')).bodyBackfill).toBe(true);
+    expect(providerProfile(account('imap.mail.yahoo.com')).bodyBackfill).toBe(true);
+    expect(providerProfile(account('imap.fastmail.com')).bodyBackfill).toBe(true);
+  });
+
+  it('excludes throttle-hostile / unproven providers (google, purelymail, microsoft)', () => {
+    expect(providerProfile(account('imap.gmail.com')).bodyBackfill).toBe(false);
+    expect(providerProfile(account('imap.purelymail.com')).bodyBackfill).toBe(false);
+    expect(providerProfile(account('outlook.office365.com')).bodyBackfill).toBe(false);
+  });
+});
+
 // ── providerProfile — oauth_provider detection ────────────────────────────────
 
 describe('providerProfile — oauth_provider fallback', () => {
@@ -1154,3 +1171,117 @@ describe('syncMessages — empty local cache vs nonempty server (wiring)', () =>
     expect(query.mock.calls.some(([sql]) => sql.includes('UPDATE folders SET highest_modseq'))).toBe(false);
   });
 });
+
+describe('ImapManager.fetchBodiesForMessages', () => {
+  // The constructor starts four setInterval timers; clear them so tests leave no open handles.
+  function makeManager() {
+    const mgr = new ImapManager({});
+    clearInterval(mgr._healthCheckTimer);
+    clearInterval(mgr._snippetSchedulerTimer);
+    clearInterval(mgr._stalenessCheckTimer);
+    clearInterval(mgr._flagPushReconcilerTimer);
+    return mgr;
+  }
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it('returns {fetched:0} without touching IMAP when given no ids', async () => {
+    const mgr = makeManager();
+    mgr.fetchMessageBody = vi.fn();
+    const result = await mgr.fetchBodiesForMessages('acct-1', []);
+    expect(result).toEqual({ fetched: 0 });
+    expect(mgr.fetchMessageBody).not.toHaveBeenCalled();
+  });
+
+  it('fetches each still-empty message and writes its body via UPDATE', async () => {
+    const mgr = makeManager();
+    mgr.fetchMessageBody = vi.fn().mockResolvedValue({ html: '<p>hi</p>', text: 'hi', attachments: [] });
+    sanitizeEmail.mockReturnValue('<p>hi</p>');
+    snippetFromBody.mockReturnValue('hi');
+    query.mockImplementation((sql) => {
+      if (/FROM email_accounts/.test(sql)) return Promise.resolve({ rows: [{ id: 'acct-1', email_address: 'x@y' }] });
+      if (/SELECT id, uid, folder FROM messages/.test(sql)) {
+        return Promise.resolve({ rows: [{ id: 'm1', uid: 5, folder: 'INBOX' }, { id: 'm2', uid: 6, folder: 'Sent' }] });
+      }
+      return Promise.resolve({ rows: [] }); // UPDATE
+    });
+
+    const result = await mgr.fetchBodiesForMessages('acct-1', ['m1', 'm2']);
+
+    expect(result).toEqual({ fetched: 2 });
+    expect(mgr.fetchMessageBody).toHaveBeenCalledWith(expect.objectContaining({ id: 'acct-1' }), 5, 'INBOX');
+    expect(mgr.fetchMessageBody).toHaveBeenCalledWith(expect.objectContaining({ id: 'acct-1' }), 6, 'Sent');
+    const updateCalls = query.mock.calls.filter(([sql]) => /UPDATE messages/.test(sql));
+    expect(updateCalls).toHaveLength(2);
+  });
+
+  it('skips a message whose body comes back empty (no html and no text)', async () => {
+    const mgr = makeManager();
+    mgr.fetchMessageBody = vi.fn().mockResolvedValue({ html: null, text: null, attachments: [] });
+    query.mockImplementation((sql) => {
+      if (/FROM email_accounts/.test(sql)) return Promise.resolve({ rows: [{ id: 'acct-1' }] });
+      if (/SELECT id, uid, folder FROM messages/.test(sql)) return Promise.resolve({ rows: [{ id: 'm1', uid: 5, folder: 'INBOX' }] });
+      return Promise.resolve({ rows: [] });
+    });
+
+    const result = await mgr.fetchBodiesForMessages('acct-1', ['m1']);
+
+    expect(result).toEqual({ fetched: 0 });
+    const updateCalls = query.mock.calls.filter(([sql]) => /UPDATE messages/.test(sql));
+    expect(updateCalls).toHaveLength(0);
+  });
+
+  it('propagates a fetchMessageBody failure so the drainer can back off', async () => {
+    const mgr = makeManager();
+    mgr.fetchMessageBody = vi.fn().mockRejectedValue(new Error('Command failed'));
+    query.mockImplementation((sql) => {
+      if (/FROM email_accounts/.test(sql)) return Promise.resolve({ rows: [{ id: 'acct-1' }] });
+      if (/SELECT id, uid, folder FROM messages/.test(sql)) return Promise.resolve({ rows: [{ id: 'm1', uid: 5, folder: 'INBOX' }] });
+      return Promise.resolve({ rows: [] });
+    });
+
+    await expect(mgr.fetchBodiesForMessages('acct-1', ['m1'])).rejects.toThrow('Command failed');
+  });
+
+  it('skips a poison message that throws mid-batch and still fetches the rest (no wedge)', async () => {
+    const mgr = makeManager();
+    mgr.fetchMessageBody = vi.fn()
+      .mockRejectedValueOnce(new Error("Mailbox doesn't exist"))
+      .mockResolvedValueOnce({ html: '<p>hi</p>', text: 'hi', attachments: [] });
+    sanitizeEmail.mockReturnValue('<p>hi</p>');
+    snippetFromBody.mockReturnValue('hi');
+    query.mockImplementation((sql) => {
+      if (/FROM email_accounts/.test(sql)) return Promise.resolve({ rows: [{ id: 'acct-1' }] });
+      if (/SELECT id, uid, folder FROM messages/.test(sql)) {
+        return Promise.resolve({ rows: [{ id: 'm1', uid: 5, folder: 'Stale' }, { id: 'm2', uid: 6, folder: 'INBOX' }] });
+      }
+      return Promise.resolve({ rows: [] });
+    });
+
+    const result = await mgr.fetchBodiesForMessages('acct-1', ['m1', 'm2']);
+
+    // The poison message (m1) is skipped, not thrown — the loop keeps going so a batch
+    // containing it still makes forward progress instead of wedging the account forever.
+    expect(result).toEqual({ fetched: 1 });
+    expect(mgr.fetchMessageBody).toHaveBeenCalledTimes(2);
+    const updateCalls = query.mock.calls.filter(([sql]) => /UPDATE messages/.test(sql));
+    expect(updateCalls).toHaveLength(1);
+  });
+
+  it('rethrows when every message in the batch fails (genuine outage, not one bad message)', async () => {
+    const mgr = makeManager();
+    mgr.fetchMessageBody = vi.fn().mockRejectedValue(new Error('Connection closed'));
+    query.mockImplementation((sql) => {
+      if (/FROM email_accounts/.test(sql)) return Promise.resolve({ rows: [{ id: 'acct-1' }] });
+      if (/SELECT id, uid, folder FROM messages/.test(sql)) {
+        return Promise.resolve({ rows: [{ id: 'm1', uid: 5, folder: 'INBOX' }, { id: 'm2', uid: 6, folder: 'INBOX' }] });
+      }
+      return Promise.resolve({ rows: [] });
+    });
+
+    await expect(mgr.fetchBodiesForMessages('acct-1', ['m1', 'm2'])).rejects.toThrow('Connection closed');
+    expect(mgr.fetchMessageBody).toHaveBeenCalledTimes(2); // both attempted, neither skipped silently
+  });
+});
diff --git a/backend/src/services/search/ftsBackfill.js b/backend/src/services/search/ftsBackfill.js
new file mode 100644
index 0000000..bacef25
--- /dev/null
+++ b/backend/src/services/search/ftsBackfill.js
@@ -0,0 +1,108 @@
+import { query } from '../db.js';
+import { searchFtsExpr, FTS_VERSION } from './lexicalRepo.js';
+import { upsertJob } from '../backgroundJobs.js';
+
+const KIND = 'fts_backfill';
+const BATCH = 3000;                 // 2–5k rows/batch
+const IDLE_MS = 5 * 60 * 1000;
+const defaultSleep = () => new Promise((r) => setTimeout(r, 250));
+
+let running = false;
+
+// FTS_VERSION is inlined as a literal (not a bind param) so the predicate
+// matches the partial index idx_messages_fts_stale_v1 exactly and the batch
+// scan is index-served rather than a growing seq scan as the tail drains.
+const STALE_PRED = `fts_version IS DISTINCT FROM ${FTS_VERSION}`;
+
+const batchSql = `
+  WITH batch AS (
+    SELECT id FROM messages
+     WHERE ${STALE_PRED}
+     ORDER BY date DESC NULLS LAST
+     LIMIT $1
+  )
+  UPDATE messages m
+     SET search_fts = ${searchFtsExpr('m')},
+         fts_version = ${FTS_VERSION}
+    FROM batch
+   WHERE m.id = batch.id
+`;
+
+async function processBatch() {
+  try {
+    const res = await query(batchSql, [BATCH]);
+    return res.rowCount;
+  } catch (err) {
+    if (err.code !== '54000') throw err; // program_limit_exceeded
+    return processBatchRowByRow();
+  }
+}
+
+// A single pathological body tripped SQLSTATE 54000 for the whole batch.
+// Re-process row by row so one bad row can't wedge the drainer.
+async function processBatchRowByRow() {
+  const { rows } = await query(
+    `SELECT id FROM messages WHERE ${STALE_PRED} ORDER BY date DESC NULLS LAST LIMIT $1`,
+    [BATCH]
+  );
+  let done = 0;
+  for (const { id } of rows) {
+    try {
+      await query(
+        `UPDATE messages m SET search_fts = ${searchFtsExpr('m')}, fts_version = ${FTS_VERSION} WHERE m.id = $1`,
+        [id]
+      );
+    } catch (err) {
+      if (err.code !== '54000') throw err;
+      // Stamp the version so it is never retried; leave search_fts NULL (still
+      // served by the ILIKE fallback in lexicalRepo).
+      await query(`UPDATE messages SET fts_version = ${FTS_VERSION} WHERE id = $1 AND search_fts IS NULL`, [id]);
+      console.warn(`FTS backfill: skipped oversized message ${id} (tsvector too large)`);
+    }
+    done++;
+  }
+  return done;
+}
+
+export async function runFtsBackfill({ sleep = defaultSleep } = {}) {
+  if (running) return; // single-flight
+  running = true;
+  try {
+    const { rows: [{ remaining }] } = await query(
+      `SELECT count(*)::int AS remaining FROM messages WHERE ${STALE_PRED}`
+    );
+    if (remaining === 0) {
+      await upsertJob({ kind: KIND, state: 'done', processed: 0, total: 0 });
+      return;
+    }
+    const total = remaining;
+    let processed = 0;
+    await upsertJob({ kind: KIND, state: 'running', processed, total });
+
+    while (true) {
+      let n;
+      try {
+        n = await processBatch();
+      } catch (err) {
+        await upsertJob({ kind: KIND, state: 'error', processed, total, lastError: err.message });
+        throw err;
+      }
+      if (n === 0) break;
+      processed += n;
+      await upsertJob({ kind: KIND, state: 'running', processed: Math.min(processed, total), total });
+      await sleep();
+    }
+    await upsertJob({ kind: KIND, state: 'done', processed: total, total });
+  } finally {
+    running = false;
+  }
+}
+
+export function scheduleFtsBackfill() {
+  runFtsBackfill().catch((err) => console.error('FTS backfill error:', err.message));
+  const timer = setInterval(() => {
+    runFtsBackfill().catch((err) => console.error('FTS backfill error:', err.message));
+  }, IDLE_MS);
+  timer.unref?.();
+  return timer;
+}
diff --git a/backend/src/services/search/ftsBackfill.test.js b/backend/src/services/search/ftsBackfill.test.js
new file mode 100644
index 0000000..3539739
--- /dev/null
+++ b/backend/src/services/search/ftsBackfill.test.js
@@ -0,0 +1,45 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+// Mock inline + import the mocked bindings (avoids the vi.mock hoisting TDZ trap).
+vi.mock('../db.js', () => ({ query: vi.fn() }));
+vi.mock('../backgroundJobs.js', () => ({ upsertJob: vi.fn() }));
+import { query } from '../db.js';
+import { upsertJob } from '../backgroundJobs.js';
+import { runFtsBackfill } from './ftsBackfill.js';
+
+const noSleep = () => Promise.resolve();
+beforeEach(() => { query.mockReset(); upsertJob.mockReset(); });
+
+describe('runFtsBackfill', () => {
+  it('marks the job done and does no work when nothing is stale', async () => {
+    query.mockResolvedValueOnce({ rows: [{ remaining: 0 }] }); // count
+    await runFtsBackfill({ sleep: noSleep });
+    expect(query).toHaveBeenCalledTimes(1); // count only, no batch UPDATE
+    expect(upsertJob).toHaveBeenCalledWith(expect.objectContaining({ kind: 'fts_backfill', state: 'done' }));
+  });
+
+  it('drains in batches until an UPDATE affects zero rows, reporting progress', async () => {
+    query
+      .mockResolvedValueOnce({ rows: [{ remaining: 5000 }] }) // count
+      .mockResolvedValueOnce({ rowCount: 3000 })              // batch 1
+      .mockResolvedValueOnce({ rowCount: 2000 })              // batch 2
+      .mockResolvedValueOnce({ rowCount: 0 });                // batch 3 → stop
+    await runFtsBackfill({ sleep: noSleep });
+    const batchSql = query.mock.calls[1][0];
+    expect(batchSql).toContain('fts_version IS DISTINCT FROM 1');
+    expect(batchSql).toContain("setweight(to_tsvector('english', coalesce(m.subject,'')), 'A')");
+    expect(upsertJob).toHaveBeenLastCalledWith(
+      expect.objectContaining({ state: 'done', processed: 5000, total: 5000 })
+    );
+  });
+
+  it('is single-flight: a second concurrent call returns immediately', async () => {
+    let release;
+    const gate = new Promise((r) => (release = r));
+    query.mockImplementationOnce(async () => { await gate; return { rows: [{ remaining: 0 }] }; });
+    const first = runFtsBackfill({ sleep: noSleep });
+    await runFtsBackfill({ sleep: noSleep }); // returns immediately (guarded)
+    release();
+    await first;
+    expect(query).toHaveBeenCalledTimes(1);
+  });
+});
diff --git a/backend/src/services/search/lexicalRepo.js b/backend/src/services/search/lexicalRepo.js
new file mode 100644
index 0000000..4d346fc
--- /dev/null
+++ b/backend/src/services/search/lexicalRepo.js
@@ -0,0 +1,267 @@
+import { shouldExcludeTrashFromSearch } from './queryParser.js';
+
+// Postgres refuses to build a tsvector larger than ~1MB of packed lexemes
+// (SQLSTATE 54000). Cap the text fed to to_tsvector at 600k chars — matching
+// msgvault's maxFTSBodyChars (internal/store/dialect_pg.go). Reused by the
+// slice-02 search_fts trigger and backfill via searchFtsExpr().
+export const FTS_BODY_CHAR_CAP = 600000;
+
+// Single source of truth for the stored-tsvector layout version. The trigger
+// and backfill both stamp fts_version = FTS_VERSION; a layout/dictionary
+// change bumps this and re-runs the backfill.
+export const FTS_VERSION = 1;
+
+// Wraps a positive condition so that when negated it also matches rows where
+// the underlying columns are NULL.
+export function negateCond(sql) {
+  return `NOT COALESCE((${sql}), false)`;
+}
+
+export function trashFolderExclusionCondition() {
+  return `NOT EXISTS (
+        SELECT 1
+        FROM folders f
+        WHERE f.account_id = m.account_id
+          AND f.path = m.folder
+          AND (f.special_use = '\\Trash'
+               OR lower(f.name) LIKE '%trash%'
+               OR lower(f.name) LIKE '%deleted%')
+      )`;
+}
+
+// One free-text term matches if it appears in the sender, subject, the stored
+// search_vector, or the length-capped body. Extracted so the body cap is a
+// single, testable source of truth (the ranked, search_fts-first variant takes
+// over once rows are backfilled).
+export function freeTextTermCondition(likeIdx, ftsIdx) {
+  return `(
+        m.from_name ILIKE $${likeIdx}
+        OR m.from_email ILIKE $${likeIdx}
+        OR m.subject ILIKE $${likeIdx}
+        OR m.search_vector @@ plainto_tsquery('english', $${ftsIdx})
+        OR to_tsvector('english', LEFT(coalesce(m.body_text,''), ${FTS_BODY_CHAR_CAP})) @@ plainto_tsquery('english', $${ftsIdx})
+      )`;
+}
+
+// BM25-style lexical rank (ts_rank_cd; class weights D,C,B,A → 10:4:1
+// subject:from:rest; length normalization 32).
+export const LEXICAL_RANK_SQL = (vectorExpr, queryExpr) =>
+  `ts_rank_cd(ARRAY[0.1, 0.1, 0.4, 1.0]::real[], ${vectorExpr}, ${queryExpr}, 32)`;
+
+// A term counts as searchable text only if it has at least one letter or
+// digit; msgvault's hasFTSToken drops punctuation-only tokens ("!!!", "***")
+// the same way, since they'd normalize to zero lexemes and can't usefully
+// match or rank anything.
+export function hasSearchableToken(term) {
+  return /[\p{L}\p{N}]/u.test(term);
+}
+
+// A bare free-text word is a single search token; a term containing whitespace
+// only arises from a quoted multi-word phrase (queryParser doesn't emit these
+// today, but the builders below stay correct if it ever does). Phrases keep
+// ordinary non-prefix matching — prefix-expanding a multi-word phrase isn't a
+// single lexeme operation and would change its meaning.
+function isPhraseTerm(term) {
+  return /\s/.test(term);
+}
+
+// msgvault's BuildFTSArg prefix-matches every bare word (typing "amaz" still
+// finds "amazon") — plainto_tsquery only matches whole stemmed words, which
+// regressed recall vs the old ILIKE substring behavior once a row is
+// backfilled onto search_fts. quote_literal($N) is a plain SQL bind param
+// reference (no JS-side string building), so the raw term travels through as
+// an ordinary parameter; quote_literal() at query time safely wraps it as a
+// single tsquery lexeme literal — neutralizing any &, |, !, (, ), ', or :
+// the term might contain — before ':*' marks it for prefix matching.
+export function ftsTermQueryArg(ftsIdx, term) {
+  return isPhraseTerm(term)
+    ? `plainto_tsquery('english', $${ftsIdx})`
+    : `to_tsquery('english', quote_literal($${ftsIdx}) || ':*')`;
+}
+
+function ftsMatchExpr(vectorExpr, ftsIdx, term) {
+  return `${vectorExpr} @@ ${ftsTermQueryArg(ftsIdx, term)}`;
+}
+
+// Ranked variant used once rows carry search_fts: backfilled rows match via the
+// GIN-indexed tsvector; rows not yet backfilled (search_fts IS NULL) fall back
+// to the legacy ILIKE/search_vector/body branch so recall never regresses
+// mid-backfill. The fallback matches nothing once search_fts is populated,
+// letting the planner use the GIN index exclusively.
+export function freeTextTermConditionRanked(likeIdx, ftsIdx, term) {
+  return `(
+        ${ftsMatchExpr('m.search_fts', ftsIdx, term)}
+        OR (m.search_fts IS NULL AND ${freeTextTermCondition(likeIdx, ftsIdx)})
+      )`;
+}
+
+// English stopwords ("the", "for", "you", …) normalize to an EMPTY tsquery
+// (numnode = 0), and `tsvector @@ <empty>` is FALSE for every row — so one
+// stopword in an AND'd term chain silently nuked ALL results once rows were
+// backfilled onto search_fts ("waiting for invoice" → 0 hits, because of
+// "for"; the ILIKE fallback arm is dead once search_fts is populated).
+// A term whose tsquery normalizes empty must contribute NOTHING instead:
+// this wraps the term's already-polarity-applied condition so it is
+// vacuously TRUE — for positive AND negated terms alike — whenever the
+// term's tsquery is empty. The numnode() probe runs on the SAME bind (and
+// the SAME prefix-or-phrase construction, via ftsTermQueryArg) the match
+// uses, so guard and match can never disagree; a query of ONLY stopwords
+// degrades to a filter-only, date-ordered search. The relevance rank needs
+// no guard: `&&` drops an empty tsquery operand and ts_rank_cd over a fully
+// empty tsquery is 0.
+export function stopwordSafeCondition(ftsIdx, term, condition) {
+  return `(numnode(${ftsTermQueryArg(ftsIdx, term)}) = 0 OR ${condition})`;
+}
+
+// The one owner of a free-text term's full predicate: ranked FTS match,
+// un-backfilled ILIKE fallback, polarity, and stopword vacuity. `bind(value) →
+// '$n'` pushes onto the caller's params; the raw ordinals expected by
+// freeTextTermConditionRanked are recovered from the placeholders.
+export function freeTextTermClause(term, negate, bind) {
+  const likeIdx = Number(bind(`%${term}%`).slice(1));
+  const ftsIdx = Number(bind(term).slice(1));
+  const cond = freeTextTermConditionRanked(likeIdx, ftsIdx, term);
+  return stopwordSafeCondition(ftsIdx, term, negate ? negateCond(cond) : cond);
+}
+
+// The weighted tsvector written into messages.search_fts. `ref` is the row
+// reference: 'm' for the backfill UPDATE, 'NEW' for the BEFORE trigger. The
+// class weights map to ts_rank_cd's D,C,B,A array (10:4:1 subject:from:rest).
+// MUST stay identical to migration 0035's trigger body.
+export function searchFtsExpr(ref) {
+  return `setweight(to_tsvector('english', coalesce(${ref}.subject,'')), 'A') ||
+    setweight(to_tsvector('english', coalesce(${ref}.from_name,'') || ' ' || coalesce(${ref}.from_email,'')), 'B') ||
+    setweight(to_tsvector('english', coalesce(${ref}.to_addresses::text,'') || ' ' || coalesce(${ref}.cc_addresses::text,'')), 'C') ||
+    setweight(to_tsvector('english', LEFT(coalesce(${ref}.body_text,''), ${FTS_BODY_CHAR_CAP})), 'D')`;
+}
+
+// Structured operator predicates (from/to/cc/subject/has/is/after/before),
+// negation-aware. `bind(value) → '$n'` pushes value onto the caller's param
+// list and returns its placeholder — the caller owns `params`/the running
+// index, so searchLexical remains the single owner of these predicates.
+// Excludes free-text terms and folder scope (`in:`), which are not
+// structured row predicates. Extracted from searchLexical byte-identically —
+// same branches, same ILIKE-arm reuse, same skip rules.
+export function buildOperatorClauses(filters, bind) {
+  const conditions = [];
+  for (const f of filters) {
+    if (f.key === 'in') continue; // controls scope, not a row condition
+    let cond = null;
+
+    if (f.key === 'from') {
+      const idx = bind(`%${f.value}%`);
+      cond = `(m.from_email ILIKE ${idx} OR m.from_name ILIKE ${idx})`;
+    } else if (f.key === 'subject') {
+      cond = `m.subject ILIKE ${bind(`%${f.value}%`)}`;
+    } else if (f.key === 'to') {
+      const idx = bind(`%${f.value}%`);
+      cond = `(m.to_addresses::text ILIKE ${idx} OR m.cc_addresses::text ILIKE ${idx})`;
+    } else if (f.key === 'cc') {
+      cond = `m.cc_addresses::text ILIKE ${bind(`%${f.value}%`)}`;
+    } else if (f.key === 'has') {
+      if (f.value === 'attachment' || f.value === 'attachments') cond = `m.has_attachments = true`;
+    } else if (f.key === 'is') {
+      if (f.value === 'unread') cond = `m.is_read = false`;
+      else if (f.value === 'read') cond = `m.is_read = true`;
+      else if (f.value === 'starred') cond = `m.is_starred = true`;
+    } else if (f.key === 'after') {
+      const d = new Date(f.value);
+      if (!isNaN(d)) cond = `m.date >= ${bind(d.toISOString())}`;
+    } else if (f.key === 'before') {
+      const d = new Date(f.value);
+      if (!isNaN(d)) cond = `m.date < ${bind(d.toISOString())}`;
+    }
+
+    if (cond) conditions.push(f.negate ? negateCond(cond) : cond);
+  }
+  return conditions;
+}
+
+// Folder-scope predicate(s) for the messages table. `folderScope`/`folderFuzzy` come from
+// resolveSearchFolderScope: a truthy fuzzy scope (in:<name>) matches the bare
+// name OR any .../<name> path; an exact scope (REST ?folder=) matches the full
+// path; a null scope carries no explicit folder and excludes trash-like folders
+// (the default search scope). `bind(value) → '$n'` — the caller owns params.
+export function buildFolderScopeClauses(folderScope, folderFuzzy, bind) {
+  const conditions = [];
+  if (folderScope) {
+    if (folderFuzzy) {
+      const name = bind(folderScope);
+      const path = bind(`%/${folderScope}`);
+      conditions.push(`(m.folder ILIKE ${name} OR m.folder ILIKE ${path})`);
+    } else {
+      conditions.push(`m.folder = ${bind(folderScope)}`);
+    }
+  } else if (shouldExcludeTrashFromSearch(folderScope)) {
+    conditions.push(trashFolderExclusionCondition());
+  }
+  return conditions;
+}
+
+export async function searchLexical(client, { parsed, accountIds, folderScope, folderFuzzy, ordering, limit, offset }) {
+  const { filters, terms } = parsed;
+  const params = [accountIds];
+  let p = 2;
+  const bind = (v) => { params.push(v); return `$${p++}`; };
+  const conditions = buildOperatorClauses(filters, bind);
+
+  for (const term of terms.slice(0, 10)) {
+    // Single-char terms are too broad/expensive; a punctuation-only term
+    // (e.g. "!!!") has no letters/digits to search on — msgvault's
+    // hasFTSToken drops it the same way rather than handing Postgres a
+    // token that would normalize to zero lexemes.
+    if (term.value.length < 2 || !hasSearchableToken(term.value)) continue;
+    conditions.push(freeTextTermClause(term.value, term.negate, bind));
+  }
+
+  // A bare in:inbox (or lone folder param) must never dump a whole folder.
+  if (!conditions.length) return { rows: [], hasCondition: false };
+
+  for (const cond of buildFolderScopeClauses(folderScope, folderFuzzy, bind)) conditions.push(cond);
+
+  // D5: rank a free-text search by ts_rank_cd over the combined positive terms
+  // (date/id tiebreak); a filter-only search stays date-ordered. The rank arg
+  // must be bound before LIMIT/OFFSET.
+  let orderBy = 'ORDER BY m.date DESC';
+  const positiveTerms = terms.slice(0, 10)
+    .filter(t => !t.negate && t.value.length >= 2 && hasSearchableToken(t.value))
+    .map(t => t.value);
+  if (ordering === 'relevance' && positiveTerms.length) {
+    // Rank with the SAME prefix-aware tsquery the MATCH predicate uses
+    // (freeTextTermConditionRanked → ftsTermQueryArg): one bind per term,
+    // combined with && (ts_rank_cd takes a single tsquery). Reusing the one
+    // term→tsquery-arg builder is what makes
+    // predicate and rank impossible to diverge — a prefix-only hit ("invo"
+    // matching "invoice") then ranks by ts_rank_cd instead of getting rank 0
+    // (which COALESCE(...,0) would collapse to date order). Phrases keep
+    // non-prefix matching via ftsTermQueryArg, exactly as the predicate does.
+    const rankArgs = positiveTerms.map((term) => { params.push(term); return p++; });
+    const rankQuery = positiveTerms.map((term, i) => ftsTermQueryArg(rankArgs[i], term)).join(' && ');
+    const rankExpr = LEXICAL_RANK_SQL('m.search_fts', rankQuery);
+    // ts_rank_cd(..., NULL::tsvector, ...) returns NULL for un-backfilled rows
+    // (search_fts IS NULL), and Postgres sorts NULLs FIRST in DESC order — so
+    // without COALESCE every un-backfilled row would outrank every properly
+    // ranked hit during the backfill window. COALESCE(...,0) keeps them
+    // sorting by the date/id tiebreak instead, below any real rank score.
+    orderBy = `ORDER BY COALESCE(${rankExpr}, 0) DESC, m.date DESC, m.id DESC`;
+  }
+
+  params.push(limit);
+  params.push(offset);
+
+  const result = await client(`
+      SELECT
+        m.id, m.uid, m.folder, m.subject, m.from_name, m.from_email,
+        m.date, m.snippet, m.is_read, m.is_starred, m.has_attachments, m.account_id,
+        a.name as account_name, a.email_address as account_email, a.color as account_color
+      FROM messages m
+      JOIN email_accounts a ON m.account_id = a.id
+      WHERE m.account_id = ANY($1)
+        AND m.is_deleted = false
+        AND ${conditions.join('\n        AND ')}
+      ${orderBy}
+      LIMIT $${p} OFFSET $${p + 1}
+    `, params);
+
+  return { rows: result.rows, hasCondition: true };
+}
diff --git a/backend/src/services/search/lexicalRepo.test.js b/backend/src/services/search/lexicalRepo.test.js
new file mode 100644
index 0000000..0144e31
--- /dev/null
+++ b/backend/src/services/search/lexicalRepo.test.js
@@ -0,0 +1,385 @@
+import { describe, it, expect, vi } from 'vitest';
+import {
+  FTS_BODY_CHAR_CAP,
+  FTS_VERSION,
+  negateCond,
+  trashFolderExclusionCondition,
+  freeTextTermCondition,
+  searchFtsExpr,
+  searchLexical,
+  LEXICAL_RANK_SQL,
+  freeTextTermConditionRanked,
+  freeTextTermClause,
+  stopwordSafeCondition,
+  buildOperatorClauses,
+  buildFolderScopeClauses,
+} from './lexicalRepo.js';
+
+describe('SQL fragment builders', () => {
+  it('pins the body cap at 600000 and FTS version at 1 (msgvault parity)', () => {
+    expect(FTS_BODY_CHAR_CAP).toBe(600000);
+    expect(FTS_VERSION).toBe(1);
+  });
+
+  it('caps the body tsvector inside the free-text condition', () => {
+    const cond = freeTextTermCondition(3, 4);
+    expect(cond).toContain(
+      `to_tsvector('english', LEFT(coalesce(m.body_text,''), ${FTS_BODY_CHAR_CAP}))`
+    );
+    expect(cond).toContain('m.from_name ILIKE $3');
+    expect(cond).toContain("m.search_vector @@ plainto_tsquery('english', $4)");
+    expect(cond).not.toContain("to_tsvector('english', coalesce(m.body_text,'')) @@");
+  });
+
+  it('negateCond wraps a positive condition to also match NULL columns', () => {
+    expect(negateCond('m.is_read = true')).toBe('NOT COALESCE((m.is_read = true), false)');
+  });
+
+  it('trashFolderExclusionCondition targets trash-like folders', () => {
+    const sql = trashFolderExclusionCondition();
+    expect(sql).toContain('NOT EXISTS');
+    expect(sql).toContain('%trash%');
+    expect(sql).toContain('%deleted%');
+  });
+
+  it('searchFtsExpr builds a weighted A/B/C/D tsvector for the given row alias', () => {
+    const expr = searchFtsExpr('m');
+    expect(expr).toContain("setweight(to_tsvector('english', coalesce(m.subject,'')), 'A')");
+    expect(expr).toContain("coalesce(m.from_name,'') || ' ' || coalesce(m.from_email,'')");
+    expect(expr).toContain("LEFT(coalesce(m.body_text,''), 600000)), 'D')");
+    // NEW.-form for the trigger must be produced by the same builder.
+    expect(searchFtsExpr('NEW')).toContain("coalesce(NEW.subject,'')");
+  });
+});
+
+describe('searchLexical', () => {
+  function mockClient() {
+    const calls = [];
+    const fn = vi.fn(async (text, params) => { calls.push({ text, params }); return { rows: [{ id: 'x' }] }; });
+    return { fn, calls };
+  }
+
+  it('returns hasCondition:false and never queries when there is no real condition', async () => {
+    const { fn } = mockClient();
+    const res = await searchLexical(fn, {
+      parsed: { filters: [{ key: 'in', value: 'inbox', negate: false }], terms: [] },
+      accountIds: ['a1'], folderScope: 'inbox', folderFuzzy: true, ordering: 'date', limit: 50, offset: 0,
+    });
+    expect(res).toEqual({ rows: [], hasCondition: false });
+    expect(fn).not.toHaveBeenCalled();
+  });
+
+  it('emits the pre-existing SQL shape and bind params for a from:+term all-folder query', async () => {
+    const { fn, calls } = mockClient();
+    const res = await searchLexical(fn, {
+      parsed: {
+        filters: [{ key: 'from', value: 'amazon', negate: false }],
+        terms: [{ value: 'invoice', negate: false }],
+      },
+      accountIds: ['a1', 'a2'], folderScope: null, folderFuzzy: false, ordering: 'date', limit: 50, offset: 0,
+    });
+    expect(res.hasCondition).toBe(true);
+    expect(res.rows).toEqual([{ id: 'x' }]);
+    const { text, params } = calls[0];
+    // Params: [accountIds, from-like, term-like, term-fts, limit, offset]
+    expect(params).toEqual([['a1', 'a2'], '%amazon%', '%invoice%', 'invoice', 50, 0]);
+    // Load-bearing SQL shape (byte-identical gate: the human-runnable old-vs-new diff).
+    expect(text).toContain('WHERE m.account_id = ANY($1)');
+    expect(text).toContain('AND m.is_deleted = false');
+    expect(text).toContain('(m.from_email ILIKE $2 OR m.from_name ILIKE $2)');
+    expect(text).toContain('NOT EXISTS'); // trash exclusion on all-folder search
+    expect(text).toContain('ORDER BY m.date DESC');
+    expect(text).toContain('LIMIT $5 OFFSET $6');
+  });
+
+  it('adds a cc_addresses predicate for the new cc: operator', async () => {
+    const { fn, calls } = mockClient();
+    await searchLexical(fn, {
+      parsed: { filters: [{ key: 'cc', value: 'boss', negate: false }], terms: [] },
+      accountIds: ['a1'], folderScope: 'INBOX', folderFuzzy: false, ordering: 'date', limit: 50, offset: 0,
+    });
+    expect(calls[0].text).toContain('m.cc_addresses::text ILIKE $2');
+    expect(calls[0].params).toEqual([['a1'], '%boss%', 'INBOX', 50, 0]);
+  });
+
+});
+
+describe('ranked lexical query (slice 02)', () => {
+  function mockClient() {
+    const calls = [];
+    const fn = vi.fn(async (text, params) => { calls.push({ text, params }); return { rows: [] }; });
+    return { fn, calls };
+  }
+
+  it('LEXICAL_RANK_SQL emits ts_rank_cd with the D,C,B,A weight array and normalization 32', () => {
+    const expr = LEXICAL_RANK_SQL('m.search_fts', "plainto_tsquery('english', $7)");
+    expect(expr).toBe("ts_rank_cd(ARRAY[0.1, 0.1, 0.4, 1.0]::real[], m.search_fts, plainto_tsquery('english', $7), 32)");
+  });
+
+  it('freeTextTermConditionRanked prefix-matches a single-word term against search_fts, falling back only when it IS NULL', () => {
+    const cond = freeTextTermConditionRanked(3, 4, 'invoice');
+    expect(cond).toContain("m.search_fts @@ to_tsquery('english', quote_literal($4) || ':*')");
+    expect(cond).not.toContain("m.search_fts @@ plainto_tsquery");
+    expect(cond).toContain('m.search_fts IS NULL AND');
+    // The IS-NULL fallback branch is untouched legacy SQL (freeTextTermCondition):
+    // still plainto_tsquery + ILIKE, since it only serves un-backfilled rows.
+    expect(cond).toContain(`LEFT(coalesce(m.body_text,''), ${FTS_BODY_CHAR_CAP})`);
+    expect(cond).toContain("m.search_vector @@ plainto_tsquery('english', $4)");
+  });
+
+  it('freeTextTermConditionRanked keeps non-prefix phrase matching for a quoted multi-word term', () => {
+    const cond = freeTextTermConditionRanked(3, 4, 'weekly report');
+    expect(cond).toContain("m.search_fts @@ plainto_tsquery('english', $4)");
+    expect(cond).not.toContain("m.search_fts @@ to_tsquery");
+  });
+
+  it('prefix-matching passes the raw term through as an ordinary bind param — quoting/escaping happens in SQL via quote_literal, never in JS', async () => {
+    const { fn, calls } = mockClient();
+    const weird = `d'Angelo:*(evil)`;
+    await searchLexical(fn, {
+      parsed: { filters: [], terms: [{ value: weird, negate: false }] },
+      accountIds: ['a1'], folderScope: 'INBOX', folderFuzzy: false, ordering: 'date', limit: 50, offset: 0,
+    });
+    const { text, params } = calls[0];
+    // The term is bound VERBATIM (no JS-side escaping) — quote_literal() at
+    // query time is what neutralizes '/:/* /()/etc. so it can't act as tsquery
+    // syntax or break out of the quoted lexeme.
+    expect(params).toContain(weird);
+    expect(text).toContain("to_tsquery('english', quote_literal($3) || ':*')");
+  });
+
+  it('drops a punctuation-only term (msgvault hasFTSToken parity) instead of emitting an invalid tsquery', async () => {
+    const { fn } = mockClient();
+    const res = await searchLexical(fn, {
+      parsed: { filters: [], terms: [{ value: '!!!', negate: false }] },
+      accountIds: ['a1'], folderScope: 'INBOX', folderFuzzy: false, ordering: 'date', limit: 50, offset: 0,
+    });
+    // A punctuation-only term contributes no condition at all — same
+    // treatment as an under-length term — so a bare "!!!" search never
+    // dumps the whole folder and never reaches Postgres with a term that
+    // would normalize to zero lexemes.
+    expect(res).toEqual({ rows: [], hasCondition: false });
+    expect(fn).not.toHaveBeenCalled();
+  });
+
+  it('drops a punctuation-only term while keeping a real term alongside it', async () => {
+    const { fn, calls } = mockClient();
+    await searchLexical(fn, {
+      parsed: { filters: [], terms: [{ value: 'invoice', negate: false }, { value: '***', negate: false }] },
+      accountIds: ['a1'], folderScope: 'INBOX', folderFuzzy: false, ordering: 'date', limit: 50, offset: 0,
+    });
+    // Only the real term's params were pushed: [accountIds, like, fts, folder, limit, offset].
+    expect(calls[0].params).toEqual([['a1'], '%invoice%', 'invoice', 'INBOX', 50, 0]);
+  });
+
+  it('orders by ts_rank_cd then date/id for a relevance query, binding the rank args before LIMIT/OFFSET', async () => {
+    const { fn, calls } = mockClient();
+    await searchLexical(fn, {
+      parsed: { filters: [], terms: [{ value: 'invoice', negate: false }, { value: 'urgent', negate: false }] },
+      accountIds: ['a1'], folderScope: null, folderFuzzy: false, ordering: 'relevance', limit: 50, offset: 0,
+    });
+    const { text, params } = calls[0];
+    // NULL-safe: ts_rank_cd(..., NULL::tsvector, ...) returns NULL, and Postgres
+    // sorts NULLs FIRST in DESC order — without COALESCE, every un-backfilled
+    // (search_fts IS NULL) row would outrank every properly ranked hit during
+    // the backfill window. COALESCE(..., 0) keeps un-backfilled rows sorting
+    // by date/id like today, below any real (non-negative) rank score.
+    expect(text).toContain('ORDER BY COALESCE(ts_rank_cd');
+    expect(text).toContain('32), 0) DESC, m.date DESC, m.id DESC');
+    // Fix 5: rank reuses the SAME prefix-aware per-term tsquery as the MATCH
+    // predicate (ftsTermQueryArg), one bind per term combined with && — NOT a
+    // single plainto_tsquery over the joined string — so a prefix-only hit ranks
+    // by ts_rank_cd instead of collapsing to COALESCE(0) date order.
+    expect(text).toContain(
+      "ts_rank_cd(ARRAY[0.1, 0.1, 0.4, 1.0]::real[], m.search_fts, to_tsquery('english', quote_literal($6) || ':*') && to_tsquery('english', quote_literal($7) || ':*'), 32)"
+    );
+    // Params: [accountIds, t1-like, t1-fts, t2-like, t2-fts, t1-rank, t2-rank, limit, offset]
+    expect(params).toEqual([['a1'], '%invoice%', 'invoice', '%urgent%', 'urgent', 'invoice', 'urgent', 50, 0]);
+  });
+
+  it('ranks a prefix-only term with the SAME builder output as the MATCH predicate (Fix 5 — never rank 0)', async () => {
+    const { fn, calls } = mockClient();
+    await searchLexical(fn, {
+      parsed: { filters: [], terms: [{ value: 'invo', negate: false }] },
+      accountIds: ['a1'], folderScope: 'INBOX', folderFuzzy: false, ordering: 'relevance', limit: 50, offset: 0,
+    });
+    const { text, params } = calls[0];
+    // The MATCH predicate prefix-matches via ftsTermQueryArg; the rank arg is the
+    // SAME per-term construction (different bind ordinal only), so predicate and
+    // rank can never diverge on prefix-vs-plainto.
+    expect(text).toContain("m.search_fts @@ to_tsquery('english', quote_literal($3) || ':*')"); // predicate
+    expect(text).toContain(
+      "ts_rank_cd(ARRAY[0.1, 0.1, 0.4, 1.0]::real[], m.search_fts, to_tsquery('english', quote_literal($5) || ':*'), 32)" // rank
+    );
+    // Params: [accountIds, term-like, term-fts(predicate), folder, term-fts(rank), limit, offset]
+    expect(params).toEqual([['a1'], '%invo%', 'invo', 'INBOX', 'invo', 50, 0]);
+  });
+
+  it('keeps non-prefix phrase matching in the rank for a quoted multi-word positive term (mirrors the predicate)', async () => {
+    const { fn, calls } = mockClient();
+    await searchLexical(fn, {
+      parsed: { filters: [], terms: [{ value: 'weekly report', negate: false }] },
+      accountIds: ['a1'], folderScope: 'INBOX', folderFuzzy: false, ordering: 'relevance', limit: 50, offset: 0,
+    });
+    const { text } = calls[0];
+    // A phrase term ranks (and matches) via plainto_tsquery — no prefix ':*'.
+    expect(text).toContain("ts_rank_cd(ARRAY[0.1, 0.1, 0.4, 1.0]::real[], m.search_fts, plainto_tsquery('english', $5), 32)");
+    expect(text).not.toContain("quote_literal($5) || ':*'");
+  });
+
+  it('keeps date ordering (no ts_rank_cd) for a filter-only query', async () => {
+    const { fn, calls } = mockClient();
+    await searchLexical(fn, {
+      parsed: { filters: [{ key: 'is', value: 'unread', negate: false }], terms: [] },
+      accountIds: ['a1'], folderScope: 'INBOX', folderFuzzy: false, ordering: 'date', limit: 50, offset: 0,
+    });
+    expect(calls[0].text).toContain('ORDER BY m.date DESC');
+    expect(calls[0].text).not.toContain('ts_rank_cd');
+  });
+});
+
+describe('stopword-safe free-text predicates (Wave D Fix 1)', () => {
+  // Root cause:
+  //   to_tsquery('english', quote_literal('for') || ':*') normalizes to an
+  //   EMPTY tsquery (numnode = 0), and `tsvector @@ <empty>` is FALSE for
+  //   every row — so once rows were backfilled onto search_fts, one english
+  //   stopword in an AND'd term chain ("waiting for invoice") nuked ALL
+  //   results. The guard makes such a term vacuously TRUE instead.
+  function mockClient() {
+    const calls = [];
+    const fn = vi.fn(async (text, params) => { calls.push({ text, params }); return { rows: [] }; });
+    return { fn, calls };
+  }
+
+  it('stopwordSafeCondition wraps a clause with a numnode()=0 escape on the SAME bind as the match', () => {
+    expect(stopwordSafeCondition(4, 'invoice', 'X')).toBe(
+      "(numnode(to_tsquery('english', quote_literal($4) || ':*')) = 0 OR X)"
+    );
+    // Phrase terms probe emptiness through the SAME plainto construction the
+    // match uses ("the on" normalizes empty exactly like a stopword word).
+    expect(stopwordSafeCondition(4, 'weekly report', 'X')).toBe(
+      "(numnode(plainto_tsquery('english', $4)) = 0 OR X)"
+    );
+  });
+
+  it('wraps every positive free-text term condition, binding NO extra params', async () => {
+    const { fn, calls } = mockClient();
+    await searchLexical(fn, {
+      parsed: { filters: [], terms: [
+        { value: 'waiting', negate: false },
+        { value: 'for', negate: false },
+        { value: 'invoice', negate: false },
+      ] },
+      accountIds: ['a1'], folderScope: 'INBOX', folderFuzzy: false, ordering: 'date', limit: 50, offset: 0,
+    });
+    const { text, params } = calls[0];
+    // One guard per term, reusing that term's fts bind ordinal ($3, $5, $7).
+    for (const ftsIdx of [3, 5, 7]) {
+      expect(text).toContain(`(numnode(to_tsquery('english', quote_literal($${ftsIdx}) || ':*')) = 0 OR (`);
+    }
+    // Guard adds no binds: [accountIds, like1, fts1, like2, fts2, like3, fts3, folder, limit, offset]
+    expect(params).toEqual([['a1'], '%waiting%', 'waiting', '%for%', 'for', '%invoice%', 'invoice', 'INBOX', 50, 0]);
+  });
+
+  it('applies the guard OUTSIDE the negation so a negated stopword also contributes nothing', async () => {
+    const { fn, calls } = mockClient();
+    await searchLexical(fn, {
+      parsed: { filters: [], terms: [{ value: 'the', negate: true }] },
+      accountIds: ['a1'], folderScope: 'INBOX', folderFuzzy: false, ordering: 'date', limit: 50, offset: 0,
+    });
+    const { text } = calls[0];
+    // (empty OR NOT COALESCE(match)) — vacuously TRUE for a stopword in BOTH
+    // polarities; a plain NOT-wrap of the guarded condition would instead be
+    // FALSE and exclude everything.
+    expect(text).toContain("(numnode(to_tsquery('english', quote_literal($3) || ':*')) = 0 OR NOT COALESCE(");
+  });
+
+  it('freeTextTermClause owns the complete ranked lexical predicate', () => {
+    const params = [];
+    let p = 2;
+    const bind = (v) => { params.push(v); return `$${p++}`; };
+    const clause = freeTextTermClause('invoice', false, bind);
+    expect(params).toEqual(['%invoice%', 'invoice']);
+    expect(clause.startsWith("(numnode(to_tsquery('english', quote_literal($3) || ':*')) = 0 OR (")).toBe(true);
+    expect(clause).toContain("m.search_fts @@ to_tsquery('english', quote_literal($3) || ':*')");
+    expect(clause).toContain('m.search_fts IS NULL AND');
+  });
+
+});
+
+describe('buildOperatorClauses (Phase 4 Task 2a — extracted from searchLexical)', () => {
+  function bindHarness(start = 2) {
+    const params = [];
+    let p = start;
+    const bind = (v) => { params.push(v); return `$${p++}`; };
+    return { bind, params };
+  }
+
+  it('builds a from: predicate reusing one bind for both ILIKE arms', () => {
+    const { bind, params } = bindHarness();
+    const conds = buildOperatorClauses([{ key: 'from', value: 'amazon', negate: false }], bind);
+    expect(conds).toEqual(['(m.from_email ILIKE $2 OR m.from_name ILIKE $2)']);
+    expect(params).toEqual(['%amazon%']);
+  });
+
+  it('builds a cc: predicate', () => {
+    const { bind, params } = bindHarness();
+    const conds = buildOperatorClauses([{ key: 'cc', value: 'boss', negate: false }], bind);
+    expect(conds).toEqual(['m.cc_addresses::text ILIKE $2']);
+    expect(params).toEqual(['%boss%']);
+  });
+
+  it('negates a structured operator via negateCond', () => {
+    const { bind } = bindHarness();
+    const conds = buildOperatorClauses([{ key: 'subject', value: 'invoice', negate: true }], bind);
+    expect(conds).toEqual(['NOT COALESCE((m.subject ILIKE $2), false)']);
+  });
+
+  it('skips in: (scope control, not a row condition) and malformed after/before', () => {
+    const { bind, params } = bindHarness();
+    const conds = buildOperatorClauses([
+      { key: 'in', value: 'inbox', negate: false },
+      { key: 'after', value: 'not-a-date', negate: false },
+    ], bind);
+    expect(conds).toEqual([]);
+    expect(params).toEqual([]);
+  });
+
+  it('does not bind free-text terms or folder scope (structured operators only)', () => {
+    const { bind } = bindHarness();
+    const conds = buildOperatorClauses([{ key: 'is', value: 'unread', negate: false }], bind);
+    expect(conds).toEqual(['m.is_read = false']);
+  });
+});
+
+describe('buildFolderScopeClauses', () => {
+  function bindHarness(start = 2) {
+    const params = [];
+    let p = start;
+    const bind = (v) => { params.push(v); return `$${p++}`; };
+    return { bind, params };
+  }
+
+  it('fuzzy in:<name> matches the bare name or any .../<name> path', () => {
+    const { bind, params } = bindHarness();
+    const conds = buildFolderScopeClauses('sent', true, bind);
+    expect(conds).toEqual(['(m.folder ILIKE $2 OR m.folder ILIKE $3)']);
+    expect(params).toEqual(['sent', '%/sent']);
+  });
+
+  it('an exact folderScope (REST ?folder=) matches the full path', () => {
+    const { bind, params } = bindHarness();
+    const conds = buildFolderScopeClauses('INBOX', false, bind);
+    expect(conds).toEqual(['m.folder = $2']);
+    expect(params).toEqual(['INBOX']);
+  });
+
+  it('a null folderScope excludes trash-like folders (default search scope), binding nothing', () => {
+    const { bind, params } = bindHarness();
+    const conds = buildFolderScopeClauses(null, false, bind);
+    expect(conds).toHaveLength(1);
+    expect(conds[0]).toContain('NOT EXISTS');
+    expect(conds[0]).toContain('%trash%');
+    expect(params).toEqual([]);
+  });
+});
diff --git a/backend/src/services/search/migrations.test.js b/backend/src/services/search/migrations.test.js
new file mode 100644
index 0000000..53f2a8c
--- /dev/null
+++ b/backend/src/services/search/migrations.test.js
@@ -0,0 +1,72 @@
+import { describe, it, expect } from 'vitest';
+import { readFileSync } from 'node:fs';
+import { fileURLToPath } from 'node:url';
+import { dirname, join } from 'node:path';
+import { searchFtsExpr, FTS_VERSION } from './lexicalRepo.js';
+
+const MIGRATIONS = join(dirname(fileURLToPath(import.meta.url)), '../../../migrations');
+const read = (f) => readFileSync(join(MIGRATIONS, f), 'utf8');
+const norm = (s) => s.replace(/\s+/g, ' ').trim();
+
+describe('0035_search_fts.sql', () => {
+  const sql = read('0035_search_fts.sql');
+
+  it('is transactional (no no-transaction header — the $$ body cannot survive the ; splitter)', () => {
+    expect(/^--\s*no-transaction/im.test(sql)).toBe(false);
+  });
+
+  it('adds nullable search_fts + fts_version with IF NOT EXISTS (fast metadata DDL, D1)', () => {
+    expect(norm(sql)).toContain('ADD COLUMN IF NOT EXISTS search_fts tsvector');
+    expect(norm(sql)).toContain('ADD COLUMN IF NOT EXISTS fts_version int');
+    expect(sql).not.toContain('GENERATED ALWAYS AS'); // D1: not a generated column
+  });
+
+  it('installs a BEFORE INSERT OR UPDATE trigger whose body equals searchFtsExpr(NEW)', () => {
+    expect(norm(sql)).toContain('BEFORE INSERT OR UPDATE ON messages');
+    expect(norm(sql)).toContain(norm(searchFtsExpr('NEW')));
+    expect(norm(sql)).toContain(`NEW.fts_version := ${FTS_VERSION}`);
+  });
+
+  it('handles the oversized-tsvector case gracefully (never fails the row write)', () => {
+    expect(sql).toContain('EXCEPTION WHEN program_limit_exceeded');
+  });
+
+  it('skips recompute only on columns that actually feed searchFtsExpr — snippet is not one of them', () => {
+    // snippet isn't part of the weighted A/B/C/D expression (subject, from,
+    // to/cc, body_text), so guarding on it made a snippet-only UPDATE (e.g.
+    // read/star-adjacent metadata writes that also touch snippet) recompute
+    // an identical search_fts for nothing.
+    expect(sql).toContain('NEW.subject      IS NOT DISTINCT FROM OLD.subject');
+    expect(sql).toContain('NEW.from_name    IS NOT DISTINCT FROM OLD.from_name');
+    expect(sql).toContain('NEW.from_email   IS NOT DISTINCT FROM OLD.from_email');
+    expect(sql).toContain('NEW.to_addresses IS NOT DISTINCT FROM OLD.to_addresses');
+    expect(sql).toContain('NEW.cc_addresses IS NOT DISTINCT FROM OLD.cc_addresses');
+    expect(sql).toContain('NEW.body_text    IS NOT DISTINCT FROM OLD.body_text');
+    expect(sql).not.toContain('OLD.snippet');
+    expect(sql).not.toContain('NEW.snippet');
+  });
+});
+
+describe('0037_search_fts_index.sql', () => {
+  const sql = read('0037_search_fts_index.sql');
+
+  it('runs outside a transaction and builds both indexes CONCURRENTLY', () => {
+    expect(/^--\s*no-transaction/im.test(sql)).toBe(true);
+    expect(sql).toContain('CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_messages_search_fts');
+    expect(sql).toContain('USING GIN (search_fts)');
+    expect(sql).toContain('idx_messages_fts_stale_v1');
+    expect(sql).toContain(`WHERE fts_version IS DISTINCT FROM ${FTS_VERSION}`);
+    expect(sql).not.toContain('$$'); // no function bodies — safe for the ; splitter
+  });
+
+  // A cancelled/crashed CREATE INDEX CONCURRENTLY leaves an INVALID index under the
+  // target name; retrying with IF NOT EXISTS silently skips it, so the scan stays
+  // unindexed forever. Drop-if-exists before each create makes the retry crash-idempotent.
+  it('drops each index CONCURRENTLY before creating it (invalid-index retry hazard)', () => {
+    for (const name of ['idx_messages_search_fts', 'idx_messages_fts_stale_v1']) {
+      expect(sql).toContain(`DROP INDEX CONCURRENTLY IF EXISTS ${name}`);
+      expect(sql.indexOf(`DROP INDEX CONCURRENTLY IF EXISTS ${name}`))
+        .toBeLessThan(sql.indexOf(`CREATE INDEX CONCURRENTLY IF NOT EXISTS ${name}`));
+    }
+  });
+});
diff --git a/backend/src/services/search/queryParser.js b/backend/src/services/search/queryParser.js
new file mode 100644
index 0000000..bdba1e3
--- /dev/null
+++ b/backend/src/services/search/queryParser.js
@@ -0,0 +1,185 @@
+// Pure query grammar for lexical search. Ports msgvault's operator set
+// (internal/search/parser.go); operators Mailflow's schema cannot serve
+// (larger:/smaller: — no size column, bcc: — not stored, label:/l: — no
+// labels) are RECORDED as unsupported, never silently widened into a match.
+
+// Multi-char keys precede single-char `l` so the alternation resolves
+// `label:` before `l:`. The leading (-?) captures optional negation; \b sits
+// between an optional '-' and the key so both `from:` and `-from:` match.
+const OP_KEYS = 'from|to|cc|bcc|subject|has|is|after|before|in|older_than|newer_than|larger|smaller|label|l';
+const OP_PATTERN = new RegExp(`(-?)\\b(${OP_KEYS}):("([^"]*)"|([\\S]+))`, 'gi');
+
+// Bare quoted phrases (msgvault tokenize parity, parser.go:395-464): a double-
+// OR single-quoted span that STARTS a token (preceded by start/whitespace,
+// optionally negated with '-') becomes ONE phrase term. A quote glued to the
+// tail of a token — from:"John Smith", d'Angelo — never starts a phrase:
+// operator values belong to OP_PATTERN and mid-word apostrophes are text.
+// Backslash escapes keep a quote char from terminating the span.
+const PHRASE_PATTERN = /(^|\s)(-?)(["'])((?:\\.|(?!\3)[^\\])*)\3/g;
+
+// Port of msgvault's unescapeQuotedValue: `\\` and an escaped quote collapse
+// to the bare char; any other `\x` keeps its backslash literally.
+function unescapePhrase(s) {
+  let out = '';
+  let escaped = false;
+  for (const ch of s) {
+    if (escaped) {
+      out += (ch === '\\' || ch === '"' || ch === "'") ? ch : '\\' + ch;
+      escaped = false;
+    } else if (ch === '\\') {
+      escaped = true;
+    } else {
+      out += ch;
+    }
+  }
+  if (escaped) out += '\\';
+  return out;
+}
+
+// Sizes: 5M / 100K / 1G → bytes. Longer suffixes checked first. Returns null
+// on anything unparseable (port of parser.go parseSize).
+function parseSize(value) {
+  const v = value.trim().toUpperCase();
+  const mult = { KB: 1024, MB: 1048576, GB: 1073741824, K: 1024, M: 1048576, G: 1073741824 };
+  for (const suffix of ['KB', 'MB', 'GB', 'K', 'M', 'G']) {
+    if (v.endsWith(suffix)) {
+      const num = parseFloat(v.slice(0, -suffix.length));
+      if (Number.isNaN(num)) return null;
+      return Math.floor(num * mult[suffix]);
+    }
+  }
+  return /^\d+$/.test(v) ? parseInt(v, 10) : null;
+}
+
+// Relative ages: 7d / 2w / 1m / 1y → an absolute ISO timestamp relative to
+// `now` (port of parser.go parseRelativeDate). Returns null if unparseable.
+function relativeAgeToISO(value, now) {
+  const m = /^(\d+)([dwmy])$/.exec(value.trim().toLowerCase());
+  if (!m) return null;
+  const n = parseInt(m[1], 10);
+  const d = new Date(now.getTime());
+  switch (m[2]) {
+    case 'd': d.setUTCDate(d.getUTCDate() - n); break;
+    case 'w': d.setUTCDate(d.getUTCDate() - n * 7); break;
+    case 'm': d.setUTCMonth(d.getUTCMonth() - n); break;
+    case 'y': d.setUTCFullYear(d.getUTCFullYear() - n); break;
+    default: return null;
+  }
+  return d.toISOString();
+}
+
+function applyOperator(key, value, negate, ctx) {
+  const { filters, unsupported, errors, now } = ctx;
+  switch (key) {
+    case 'from': case 'to': case 'cc':
+    case 'subject': case 'has': case 'is':
+    case 'in':
+      if (value) filters.push({ key, value, negate });
+      return;
+    case 'after':
+    case 'before': {
+      if (!value) return;
+      // A bad date must be RECORDED, not silently dropped downstream
+      // (buildOperatorClauses skips unparseable dates via isNaN, which would
+      // silently WIDEN the results). Same uniform message parser.go's
+      // operatorValueError emits for a bad before:/after: value.
+      if (isNaN(new Date(value))) {
+        errors.push(`invalid value "${value}" for ${key}: — expected a date like YYYY-MM-DD`);
+        return;
+      }
+      filters.push({ key, value, negate });
+      return;
+    }
+    case 'newer_than':
+    case 'older_than': {
+      const iso = relativeAgeToISO(value, now);
+      if (!iso) {
+        errors.push(`invalid value "${value}" for ${key}: — expected a relative age like 7d, 2w, 1m, or 1y`);
+        return;
+      }
+      filters.push({ key: key === 'newer_than' ? 'after' : 'before', value: iso, negate });
+      return;
+    }
+    case 'larger':
+    case 'smaller': {
+      if (parseSize(value) === null) {
+        errors.push(`invalid value "${value}" for ${key}: — expected a size like 5M, 100K, or 1G`);
+        return;
+      }
+      // Recognized and well-formed, but messages has no size column.
+      unsupported.push({ key, token: `${key}:${value}` });
+      return;
+    }
+    case 'bcc':
+    case 'label':
+    case 'l':
+      // No bcc column / no labels concept in Mailflow.
+      if (value) unsupported.push({ key: key === 'l' ? 'label' : key, token: `${key}:${value}` });
+      return;
+    default:
+      return;
+  }
+}
+
+export function parseQuery(raw, { now = new Date() } = {}) {
+  const filters = [];
+  const terms = [];
+  const unsupported = [];
+  const errors = [];
+  const ctx = { filters, unsupported, errors, now };
+
+  // Phase 1: lift bare quoted phrases out BEFORE the operator grammar runs, so
+  // a colon inside a phrase ("subject:not an operator") can't be parsed as an
+  // operator. Each phrase leaves a U+E000-delimited placeholder in the string
+  // (a Private Use Area char) — restored in term order below — so phrases and
+  // bare words keep their relative positions. U+E000 is stripped from the raw
+  // input first, so user text can never forge a placeholder.
+  const phrases = [];
+  const withPhrases = (raw || '').replace(/\uE000/g, '').replace(PHRASE_PATTERN, (_, pre, neg, _q, body) => {
+    const value = unescapePhrase(body);
+    if (!value.trim()) return pre; // empty phrase ("") contributes nothing
+    phrases.push({ value, negate: neg === '-' });
+    return `${pre}\uE000${phrases.length - 1}\uE000`;
+  });
+
+  const remaining = withPhrases.replace(OP_PATTERN, (_, neg, key, _v, quoted, unquoted) => {
+    const k = key.toLowerCase();
+    const value = (quoted !== undefined ? quoted : (unquoted || '')).toLowerCase().trim();
+    applyOperator(k, value, neg === '-', ctx);
+    return ' ';
+  }).trim();
+
+  for (const word of remaining.split(/\s+/)) {
+    let w = word.trim();
+    if (!w || w === '-') continue; // skip blanks and a lone '-' (nothing to negate)
+    const ph = /^\uE000(\d+)\uE000$/.exec(w);
+    if (ph) { terms.push(phrases[Number(ph[1])]); continue; }
+    let negate = false;
+    if (w[0] === '-' && w.length > 1) { negate = true; w = w.slice(1); }
+    terms.push({ value: w, negate });
+  }
+
+  return { filters, terms, unsupported, errors };
+}
+
+export function resolveSearchFolderScope(filters, folderParam = '') {
+  let folderScope;
+  let folderFuzzy = false; // in:<name> matches loosely; the folder param is exact
+
+  for (const f of filters) {
+    if (f.key !== 'in') continue;
+    if (f.value === 'all') { folderScope = null; }
+    else { folderScope = f.value; folderFuzzy = true; }
+  }
+
+  if (folderScope === undefined) {
+    folderScope = (folderParam || '').trim() || null;
+    folderFuzzy = false;
+  }
+
+  return { folderScope, folderFuzzy };
+}
+
+export function shouldExcludeTrashFromSearch(folderScope) {
+  return folderScope === null;
+}
diff --git a/backend/src/services/search/queryParser.test.js b/backend/src/services/search/queryParser.test.js
new file mode 100644
index 0000000..58485aa
--- /dev/null
+++ b/backend/src/services/search/queryParser.test.js
@@ -0,0 +1,225 @@
+import { describe, it, expect } from 'vitest';
+import {
+  parseQuery,
+  resolveSearchFolderScope,
+  shouldExcludeTrashFromSearch,
+} from './queryParser.js';
+
+const NOW = new Date('2026-07-16T00:00:00.000Z');
+
+describe('parseQuery — preserved grammar', () => {
+  it('treats bare words as free-text terms', () => {
+    const { filters, terms } = parseQuery('hello world');
+    expect(filters).toEqual([]);
+    expect(terms).toEqual([
+      { value: 'hello', negate: false },
+      { value: 'world', negate: false },
+    ]);
+  });
+
+  it('extracts positive operators and lowercases their values', () => {
+    const { filters, terms } = parseQuery('from:Amazon subject:Invoice hello');
+    expect(filters).toEqual([
+      { key: 'from', value: 'amazon', negate: false },
+      { key: 'subject', value: 'invoice', negate: false },
+    ]);
+    expect(terms).toEqual([{ value: 'hello', negate: false }]);
+  });
+
+  it('supports quoted operator values with spaces', () => {
+    const { filters } = parseQuery('from:"John Smith" report');
+    expect(filters).toEqual([{ key: 'from', value: 'john smith', negate: false }]);
+  });
+
+  it('negates an operator when prefixed with -', () => {
+    const { filters } = parseQuery('-from:Smith report');
+    expect(filters).toEqual([{ key: 'from', value: 'smith', negate: true }]);
+  });
+
+  it('negates a free-text term when prefixed with -', () => {
+    const { terms } = parseQuery('report -invoice');
+    expect(terms).toEqual([
+      { value: 'report', negate: false },
+      { value: 'invoice', negate: true },
+    ]);
+  });
+
+  it('preserves repeated and mixed positive/negative operators', () => {
+    const { filters } = parseQuery('from:alice -from:bob is:unread');
+    expect(filters).toEqual([
+      { key: 'from', value: 'alice', negate: false },
+      { key: 'from', value: 'bob', negate: true },
+      { key: 'is', value: 'unread', negate: false },
+    ]);
+  });
+
+  it('ignores a lone - so it is not treated as a negated empty term', () => {
+    const { terms } = parseQuery('report - draft');
+    expect(terms).toEqual([
+      { value: 'report', negate: false },
+      { value: 'draft', negate: false },
+    ]);
+  });
+
+  it('returns empty structures for blank input', () => {
+    expect(parseQuery('')).toEqual({ filters: [], terms: [], unsupported: [], errors: [] });
+    expect(parseQuery('    ')).toEqual({ filters: [], terms: [], unsupported: [], errors: [] });
+  });
+
+  it('parses in:all and named folders', () => {
+    expect(parseQuery('in:all invoice').filters).toEqual([
+      { key: 'in', value: 'all', negate: false },
+    ]);
+    expect(parseQuery('in:Sent proposal').filters).toEqual([
+      { key: 'in', value: 'sent', negate: false },
+    ]);
+  });
+});
+
+describe('parseQuery — new msgvault operators', () => {
+  it('adds cc: as a real filter', () => {
+    const { filters } = parseQuery('cc:boss@corp.com report');
+    expect(filters).toEqual([{ key: 'cc', value: 'boss@corp.com', negate: false }]);
+  });
+
+  it('maps newer_than: to an after: date and older_than: to a before: date', () => {
+    const newer = parseQuery('newer_than:2w', { now: NOW }).filters;
+    expect(newer[0].key).toBe('after');
+    expect(newer[0].value.startsWith('2026-07-02')).toBe(true);
+
+    const older = parseQuery('older_than:7d', { now: NOW }).filters;
+    expect(older[0].key).toBe('before');
+    expect(older[0].value.startsWith('2026-07-09')).toBe(true);
+  });
+
+  it('records larger:/smaller:/bcc:/label:/l: as recognized-but-unsupported (never widened)', () => {
+    const { filters, unsupported } = parseQuery('larger:5M bcc:x@y.com label:work l:home smaller:100K');
+    expect(filters).toEqual([]); // none of these silently become predicates
+    expect(unsupported).toEqual([
+      { key: 'larger', token: 'larger:5m' },
+      { key: 'bcc', token: 'bcc:x@y.com' },
+      { key: 'label', token: 'label:work' },
+      { key: 'label', token: 'l:home' },
+      { key: 'smaller', token: 'smaller:100k' },
+    ]);
+  });
+
+  it('records a malformed typed value as an error, not a filter', () => {
+    const { errors, unsupported } = parseQuery('larger:5X older_than:3q');
+    expect(unsupported).toEqual([]);
+    expect(errors).toHaveLength(2);
+    expect(errors[0]).toContain('larger');
+    expect(errors[1]).toContain('older_than');
+  });
+});
+
+describe('parseQuery — bare quoted phrases (Wave D Fix 4, msgvault tokenize parity)', () => {
+  it('treats a double-quoted phrase as ONE term without the quote chars (port "quoted phrase")', () => {
+    expect(parseQuery('"hello world"').terms).toEqual([{ value: 'hello world', negate: false }]);
+  });
+
+  it('mixes phrases with operators and bare words in order (port "mixed operators and text")', () => {
+    const { filters, terms } = parseQuery('from:alice@example.com "meeting notes" urgent');
+    expect(filters).toEqual([{ key: 'from', value: 'alice@example.com', negate: false }]);
+    expect(terms).toEqual([
+      { value: 'meeting notes', negate: false },
+      { value: 'urgent', negate: false },
+    ]);
+  });
+
+  it('keeps colons inside a phrase from becoming operators (port QuotedPhrasesWithColons)', () => {
+    expect(parseQuery('"foo:bar"').terms).toEqual([{ value: 'foo:bar', negate: false }]);
+    expect(parseQuery('"meeting at 10:30"').terms).toEqual([{ value: 'meeting at 10:30', negate: false }]);
+    expect(parseQuery('"check http://example.com"').terms).toEqual([{ value: 'check http://example.com', negate: false }]);
+    expect(parseQuery('"a:b:c:d"').terms).toEqual([{ value: 'a:b:c:d', negate: false }]);
+  });
+
+  it('parses a colon phrase alongside a real operator (port "quoted colon phrase mixed with real operator")', () => {
+    const { filters, terms } = parseQuery('from:alice@example.com "subject:not an operator"');
+    expect(filters).toEqual([{ key: 'from', value: 'alice@example.com', negate: false }]);
+    expect(terms).toEqual([{ value: 'subject:not an operator', negate: false }]);
+  });
+
+  it('parses a leading phrase before an operator (port "operator followed by quoted colon phrase")', () => {
+    const { filters, terms } = parseQuery('"re: meeting notes" from:bob@example.com');
+    expect(filters).toEqual([{ key: 'from', value: 'bob@example.com', negate: false }]);
+    expect(terms).toEqual([{ value: 're: meeting notes', negate: false }]);
+  });
+
+  it('accepts single-quoted phrases (msgvault tokenize takes both quote chars)', () => {
+    expect(parseQuery("'hello world'").terms).toEqual([{ value: 'hello world', negate: false }]);
+  });
+
+  it('never starts a phrase at an apostrophe INSIDE a word', () => {
+    expect(parseQuery("d'Angelo report").terms).toEqual([
+      { value: "d'Angelo", negate: false },
+      { value: 'report', negate: false },
+    ]);
+  });
+
+  it('unescapes backslash-escaped quotes inside a phrase (msgvault unescapeQuotedValue)', () => {
+    expect(parseQuery('"say \\"hi\\" now"').terms).toEqual([{ value: 'say "hi" now', negate: false }]);
+    expect(parseQuery('"a\\\\b"').terms).toEqual([{ value: 'a\\b', negate: false }]);
+  });
+
+  it('negates a phrase with a leading -', () => {
+    expect(parseQuery('report -"weekly digest"').terms).toEqual([
+      { value: 'report', negate: false },
+      { value: 'weekly digest', negate: true },
+    ]);
+  });
+
+  it('drops an empty phrase and leaves op:"value" quoting to the operator grammar', () => {
+    expect(parseQuery('""').terms).toEqual([]);
+    expect(parseQuery('from:"John Smith"').filters).toEqual([{ key: 'from', value: 'john smith', negate: false }]);
+    expect(parseQuery('from:"John Smith"').terms).toEqual([]);
+  });
+});
+
+describe('parseQuery — before:/after: validation (Wave D Fix 5)', () => {
+  it('records an invalid before:/after: value as an error, never a filter (no silent widening)', () => {
+    const { filters, errors } = parseQuery('before:notadate after:2025-99-99 invoice');
+    expect(filters).toEqual([]);
+    expect(errors).toEqual([
+      'invalid value "notadate" for before: — expected a date like YYYY-MM-DD',
+      'invalid value "2025-99-99" for after: — expected a date like YYYY-MM-DD',
+    ]);
+  });
+
+  it('keeps valid dates as filters (msgvault parseDate accepts several forms)', () => {
+    const { filters, errors } = parseQuery('after:2025-01-02 before:2025/03/04');
+    expect(errors).toEqual([]);
+    expect(filters.map((f) => f.key)).toEqual(['after', 'before']);
+  });
+});
+
+describe('resolveSearchFolderScope / shouldExcludeTrashFromSearch', () => {
+  it('scopes to the client folder param when no in: operator is present', () => {
+    const { filters } = parseQuery('subject:newsletter');
+    expect(resolveSearchFolderScope(filters, 'INBOX')).toEqual({
+      folderScope: 'INBOX',
+      folderFuzzy: false,
+    });
+  });
+
+  it('lets in: override the client folder param', () => {
+    const { filters } = parseQuery('in:trash subject:newsletter');
+    expect(resolveSearchFolderScope(filters, 'INBOX')).toEqual({
+      folderScope: 'trash',
+      folderFuzzy: true,
+    });
+  });
+
+  it('flags all-folder searches for trash exclusion', () => {
+    const { filters } = parseQuery('subject:newsletter');
+    const { folderScope } = resolveSearchFolderScope(filters);
+    expect(folderScope).toBeNull();
+    expect(shouldExcludeTrashFromSearch(folderScope)).toBe(true);
+  });
+
+  it('keeps explicit folder searches eligible to find trash', () => {
+    const { filters } = parseQuery('in:trash subject:newsletter');
+    const { folderScope } = resolveSearchFolderScope(filters);
+    expect(shouldExcludeTrashFromSearch(folderScope)).toBe(false);
+  });
+});
diff --git a/backend/src/services/search/searchService.js b/backend/src/services/search/searchService.js
new file mode 100644
index 0000000..49e6114
--- /dev/null
+++ b/backend/src/services/search/searchService.js
@@ -0,0 +1,43 @@
+import { query } from '../db.js';
+import { searchLexical } from './lexicalRepo.js';
+import { resolveSearchFolderScope } from './queryParser.js';
+
+function clampLimit(limit) {
+  return Math.max(1, Math.min(parseInt(limit) || 50, 200));
+}
+
+export async function search(request) {
+  const { userId, accountId, parsed, folderParam = '', limit = 50, offset = 0 } = request;
+
+  const cap = clampLimit(limit);
+  const off = Math.max(0, parseInt(offset) || 0);
+  const emptyPage = { offset: off, limit: cap, hasMore: false };
+
+  const accountsResult = await query(
+    'SELECT id FROM email_accounts WHERE user_id = $1 AND enabled = true',
+    [userId]
+  );
+  let accountIds = accountsResult.rows.map(r => r.id);
+  if (!accountIds.length) return { messages: [], mode: 'lexical', page: emptyPage };
+
+  // Optional single-account narrowing, only within the authenticated user's scope.
+  if (accountId && accountIds.includes(accountId)) accountIds = [accountId];
+
+  const { folderScope, folderFuzzy } = resolveSearchFolderScope(parsed.filters, folderParam);
+
+  // D5: a free-text search (≥1 positive, non-trivial term) ranks by relevance;
+  // a filter-only search stays date-ordered.
+  const hasPositiveText = parsed.terms.some(t => !t.negate && t.value.length >= 2);
+  const ordering = hasPositiveText ? 'relevance' : 'date';
+
+  const { rows, hasCondition } = await searchLexical(query, {
+    parsed, accountIds, folderScope, folderFuzzy, ordering, limit: cap, offset: off,
+  });
+  if (!hasCondition) return { messages: [], mode: 'lexical', page: emptyPage };
+
+  return {
+    messages: rows,
+    mode: 'lexical',
+    page: { offset: off, limit: cap, hasMore: rows.length === cap },
+  };
+}
diff --git a/backend/src/services/search/searchService.test.js b/backend/src/services/search/searchService.test.js
new file mode 100644
index 0000000..4735be6
--- /dev/null
+++ b/backend/src/services/search/searchService.test.js
@@ -0,0 +1,67 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+
+vi.mock('../db.js', () => ({ query: vi.fn() }));
+vi.mock('./lexicalRepo.js', async (importOriginal) => {
+  const actual = await importOriginal();
+  return { ...actual, searchLexical: vi.fn() };
+});
+
+import { query } from '../db.js';
+import { searchLexical } from './lexicalRepo.js';
+import { search } from './searchService.js';
+
+beforeEach(() => {
+  query.mockReset();
+  searchLexical.mockReset();
+});
+
+function withAccounts(ids) {
+  query.mockResolvedValueOnce({ rows: ids.map(id => ({ id })) });
+}
+
+describe('searchService.search', () => {
+  it('returns an empty shaped result when the user has no enabled accounts', async () => {
+    withAccounts([]);
+    const res = await search({ userId: 'u1', parsed: { filters: [], terms: [{ value: 'hi', negate: false }] } });
+    expect(res).toEqual({ messages: [], mode: 'lexical', page: { offset: 0, limit: 50, hasMore: false } });
+    expect(searchLexical).not.toHaveBeenCalled();
+  });
+
+  it('scopes to a single account only when it belongs to the user', async () => {
+    withAccounts(['a1', 'a2']);
+    searchLexical.mockResolvedValue({ rows: [], hasCondition: true });
+    await search({ userId: 'u1', accountId: 'a2', parsed: { filters: [], terms: [{ value: 'hi', negate: false }] } });
+    expect(searchLexical.mock.calls[0][1].accountIds).toEqual(['a2']);
+  });
+
+  it('chooses relevance ordering for free text and date ordering for filter-only queries', async () => {
+    withAccounts(['a1']);
+    searchLexical.mockResolvedValue({ rows: [], hasCondition: true });
+    await search({ userId: 'u1', parsed: { filters: [], terms: [{ value: 'invoice', negate: false }] } });
+    expect(searchLexical.mock.calls[0][1].ordering).toBe('relevance');
+
+    query.mockReset(); searchLexical.mockReset();
+    withAccounts(['a1']);
+    searchLexical.mockResolvedValue({ rows: [], hasCondition: true });
+    await search({ userId: 'u1', parsed: { filters: [{ key: 'is', value: 'unread', negate: false }], terms: [] } });
+    expect(searchLexical.mock.calls[0][1].ordering).toBe('date');
+  });
+
+  it('shapes rows into messages + page, clamping the limit to 200 and flagging hasMore on a full page', async () => {
+    withAccounts(['a1']);
+    searchLexical.mockResolvedValue({ rows: new Array(200).fill({ id: 'm' }), hasCondition: true });
+    const res = await search({ userId: 'u1', limit: 9999, offset: 40, parsed: { filters: [], terms: [{ value: 'x', negate: false }] } });
+    expect(res.messages).toHaveLength(200);
+    expect(res.mode).toBe('lexical');
+    expect(res.page).toEqual({ offset: 40, limit: 200, hasMore: true });
+    expect(searchLexical.mock.calls[0][1].limit).toBe(200);
+  });
+
+  it('returns empty when there is no real search condition', async () => {
+    withAccounts(['a1']);
+    searchLexical.mockResolvedValue({ rows: [], hasCondition: false });
+    const res = await search({ userId: 'u1', parsed: { filters: [{ key: 'in', value: 'inbox', negate: false }], terms: [] } });
+    expect(res.messages).toEqual([]);
+    expect(res.page.hasMore).toBe(false);
+  });
+});
diff --git a/backend/src/testSupport/testAccount.js b/backend/src/testSupport/testAccount.js
new file mode 100644
index 0000000..88b0afc
--- /dev/null
+++ b/backend/src/testSupport/testAccount.js
@@ -0,0 +1,19 @@
+// Shared IT/dev-script fixture: seed and tear down a throwaway user + email account
+// against a real DB. `db` is anything with .query(text, params) — a pg.Client, a
+// pooled client, or the app pool. Deleting the user cascades to email_accounts →
+// messages (ON DELETE CASCADE), so cleanupAccount removes everything in one step.
+import { randomUUID } from 'crypto';
+
+export async function seedAccount(db, label = 'it') {
+  const u = await db.query('INSERT INTO users (username) VALUES ($1) RETURNING id', [`${label}-${randomUUID()}`]);
+  const userId = u.rows[0].id;
+  const a = await db.query(
+    'INSERT INTO email_accounts (user_id, name, email_address) VALUES ($1, $2, $3) RETURNING id',
+    [userId, label, `${label}-${randomUUID()}@example.com`],
+  );
+  return { userId, accountId: a.rows[0].id };
+}
+
+export async function cleanupAccount(db, userId) {
+  if (userId) await db.query('DELETE FROM users WHERE id = $1', [userId]);
+}
diff --git a/frontend/src/components/MessageList.jsx b/frontend/src/components/MessageList.jsx
index efb16df..f466518 100644
--- a/frontend/src/components/MessageList.jsx
+++ b/frontend/src/components/MessageList.jsx
@@ -2171,7 +2171,9 @@ export default function MessageList() {
   const showInboxIcon = !isUnified && selectedFolder === 'INBOX' && !searchQuery.trim();
 
   const label = searchQuery.trim()
-    ? `Search: "${searchQuery}"`
+    // No quotes around the query: when the header truncates with a CSS ellipsis
+    // the closing quote would be lost, stranding an unbalanced opening one.
+    ? `Search: ${searchQuery}`
     : isUnified ? t('sidebar.allInboxes') : selectedFolder;
 
   // Non-INBOX folders omitted: byAccount is account-total, not folder-specific, so it would mislead.
@@ -2246,7 +2248,7 @@ export default function MessageList() {
                   <polyline points="22 12 16 12 14 15 10 15 8 12 2 12"/>
                   <path d="M5.45 5.11L2 12v6a2 2 0 002 2h16a2 2 0 002-2v-6l-3.45-6.89A2 2 0 0016.76 4H7.24a2 2 0 00-1.79 1.11z"/>
                 </svg>
-              ) : label}
+              ) : <span style={{ overflow: 'hidden', textOverflow: 'ellipsis', whiteSpace: 'nowrap', minWidth: 0 }}>{label}</span>}
             </h2>
             {headerUnread > 0 && !searchQuery.trim() && (
               <span style={{
@@ -2378,7 +2380,7 @@ export default function MessageList() {
                 <polyline points="22 12 16 12 14 15 10 15 8 12 2 12"/>
                 <path d="M5.45 5.11L2 12v6a2 2 0 002 2h16a2 2 0 002-2v-6l-3.45-6.89A2 2 0 0016.76 4H7.24a2 2 0 00-1.79 1.11z"/>
               </svg>
-            ) : label}
+            ) : <span style={{ overflow: 'hidden', textOverflow: 'ellipsis', whiteSpace: 'nowrap', minWidth: 0 }}>{label}</span>}
           </h2>
           <div style={{ display: 'flex', alignItems: 'center', gap: 4, flexShrink: 0, marginLeft: 6 }}>
             {messagesTotal > 0 && !searchQuery && (
diff --git a/frontend/src/utils/api.search.test.js b/frontend/src/utils/api.search.test.js
new file mode 100644
index 0000000..a237a77
--- /dev/null
+++ b/frontend/src/utils/api.search.test.js
@@ -0,0 +1,18 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { api } from './api.js';
+
+test('search sends lexical pagination and folder parameters', async () => {
+  let seen;
+  const orig = globalThis.fetch;
+  globalThis.fetch = async (url) => { seen = url; return { ok: true, json: async () => ({ messages: [] }) }; };
+  try {
+    await api.search('quarterly report', 'a1', { offset: 50, limit: 25, folder: 'INBOX' });
+  } finally { globalThis.fetch = orig; }
+  const url = new URL(seen, 'http://localhost');
+  assert.equal(url.searchParams.get('q'), 'quarterly report');
+  assert.equal(url.searchParams.get('accountId'), 'a1');
+  assert.equal(url.searchParams.get('offset'), '50');
+  assert.equal(url.searchParams.get('limit'), '25');
+  assert.equal(url.searchParams.get('folder'), 'INBOX');
+});