feat(WS-26d-autolead): auto-lead from inbound email, OFF behind CRM_AUTO_LEAD (D5)

apps/services/gateway/gateway/routes/email/scheduler_hooks.py

@@ -56,7 +56,8 @@ async def auto_run_rules_for_account(account_id: str) -> None:
 
 async def process_new_mail(account_id: str) -> None:
     """The shared new-mail pipeline (H1): auto-run rules → sweep the leftovers →
-    categorize senders → classify threads (Reply Zero) → auto-archive.
+    categorize senders → classify threads (Reply Zero) → auto-archive → CRM
+    auto-lead.
 
     Order matters. The rules run first and are the only thing that *classifies*.
     The sweep then projects that (plus learned patterns) onto inbox mail the
@@ -66,6 +67,13 @@ async def process_new_mail(account_id: str) -> None:
     invisible to the Email Cleaner. Sender rollup runs after both so it sees the
     complete label set.
 
+    The CRM auto-lead step (WS-26d-autolead) runs LAST, and after auto-archive
+    on purpose: it considers what is still in the INBOX once the account's own
+    automation has finished with it, so mail the user's rules said "I do not
+    need to see this" about never becomes a lead. It is also the only step here
+    that belongs to another app — it lives in ``routes/crm/auto_lead.py``,
+    because what it does is write a CRM record.
+
     Each step is isolated so one failure never skips the rest (same guarantee the
     scheduler loop gave when these were separate steps). Registered as the
     ``on_new_mail`` hook AND called directly by the manual-sync route + webhook,
@@ -110,6 +118,34 @@ async def process_new_mail(account_id: str) -> None:
     except Exception as exc:  # noqa: BLE001
         _log.warning("sync.auto_archive_failed", account_id=account_id,
                      error=str(exc)[:200])
+    try:
+        # WS-26d-autolead. ⚠️ The flag is checked HERE, before the step is
+        # entered — never inside it. With CRM_AUTO_LEAD off the step is not
+        # imported, not called, opens no session and issues no query; a gate
+        # that lived inside `create_leads_from_new_mail` would open a database
+        # session on every sync cycle of every mailbox to discover it had
+        # nothing to do.
+        #
+        # The predicate itself is imported above the gate, and that is
+        # deliberate: `auto_lead_enabled` is the flag's ONE definition, and
+        # reading `settings.crm_auto_lead` here instead would make two places
+        # responsible for agreeing what the flag means — which is how a loop
+        # ends up running with its flag off (the `sync_enabled` precedent).
+        # The cost is one `sys.modules` lookup, since `routes/crm` is mounted
+        # by `main.py` at boot regardless of this flag.
+        #
+        # ⚠️ Divergence from the five steps above, on purpose: the import sits
+        # INSIDE the try. A `routes/crm` module that fails to import must be
+        # logged like any other CRM failure, not raised out of the mail path.
+        from gateway.routes.crm.auto_lead import auto_lead_enabled
+
+        if auto_lead_enabled():
+            from gateway.routes.crm.auto_lead import create_leads_from_new_mail
+
+            await create_leads_from_new_mail(account_id)
+    except Exception as exc:  # noqa: BLE001
+        _log.warning("sync.auto_lead_failed", account_id=account_id,
+                     error=str(exc)[:200])
 
 
 async def learn_label_changes(account_id: str, changes: list) -> None:

infra/AGENTS.md

@@ -5,7 +5,7 @@ Docker Compose, Postgres schema, LiteLLM tier config. LLM routing is via the gat
 
 ## Key Files
 - docker-compose.yml -- core services (Postgres 16 + pgvector, Redis 7)
-- postgres/ -- schema files (00-10) + 09_app_user.sql (NextAuth users) + 11_integration_credentials.sql (unified credential store) + 130_org_access_control.sql (organization, membership lifecycle on app_user, org_role/org_role_permission/user_role, user_permission_override, feature_catalog — spec: ai-company-brain/specs/org_access_control.md) + 131_integration_memory_permissions.sql (additive: grants `integrations:use:*` + org-memory permissions to the seeded roles; `member` reads org memory but does not write it) + 143_access_request.sql (the sign-in queue — one row per address that authenticated with no `app_user` row, unique on `lower(email)`; `status` carries a `CHECK (pending|approved|denied)` because the vocabulary is load-bearing for ACCESS — approve acts only on `pending`, so a typo'd status would fall out of both the queue and the decided record; deliberately a SEPARATE table and not a fifth `app_user.status`, because an `app_user` row IS the org's member record and a stranger who merely knocked must not acquire one that a future join can surface — spec: ai-company-brain/specs/colleague_onboarding.md §6) + 144_crm.sql (the native CRM spine — organizations/contacts/leads/deals, **statuses as data** rather than enums, one `crm_activities` timeline whose four target FKs are all nullable under a CHECK requiring at least one, and a `crm_status_changes` dwell log; also seeds the `crm` feature_catalog row. Contains the schema's **one FK cycle** — `crm_leads.converted_deal_id` ⟷ `crm_deals.lead_id` — closed by a guarded `DO $$` on `pg_constraint`, because `ALTER TABLE … ADD CONSTRAINT` has no `IF NOT EXISTS`. Idempotency is pinned STATICALLY by `tests/unit/test_crm_migration.py`, which reads the file as text: the unit suite runs no database, so an idempotency claim that holds only by inspection is unenforceable. Spec: ai-company-brain/specs/crm_app.md §3) + 145_crm_zoho_sync.sql (what the two-way Zoho sync needs: `zoho_dirty`/`zoho_synced_at` on the four CRM record tables — and **only** those four, since pipeline vocabulary flows down-only and an activity's push signal is its NULL `zoho_id` — plus `crm_zoho_tombstones` (FK-less on purpose: the row it describes is gone by the time anyone reads it) and `crm_sync_cursors` (`module` PK, so the pull cannot silently rewind). Idempotent via `ADD COLUMN IF NOT EXISTS`, which — unlike `ADD CONSTRAINT` — Postgres supports directly, so no guarded `DO $$` is needed. Pinned by the same static `tests/unit/test_crm_migration.py`, which finds BOTH CRM migrations by CONTENT rather than by number. Spec: crm_app.md §7.1). ⚠️ `CREATE TABLE IF NOT EXISTS` means a column or constraint added to an ALREADY-APPLIED migration file is silently skipped on that deployment; 143, 144 and 145 have never been applied anywhere (merging 143 is the OWNER-GATE), which is the only reason they are still editable in place.
+- postgres/ -- schema files (00-10) + 09_app_user.sql (NextAuth users) + 11_integration_credentials.sql (unified credential store) + 130_org_access_control.sql (organization, membership lifecycle on app_user, org_role/org_role_permission/user_role, user_permission_override, feature_catalog — spec: ai-company-brain/specs/org_access_control.md) + 131_integration_memory_permissions.sql (additive: grants `integrations:use:*` + org-memory permissions to the seeded roles; `member` reads org memory but does not write it) + 143_access_request.sql (the sign-in queue — one row per address that authenticated with no `app_user` row, unique on `lower(email)`; `status` carries a `CHECK (pending|approved|denied)` because the vocabulary is load-bearing for ACCESS — approve acts only on `pending`, so a typo'd status would fall out of both the queue and the decided record; deliberately a SEPARATE table and not a fifth `app_user.status`, because an `app_user` row IS the org's member record and a stranger who merely knocked must not acquire one that a future join can surface — spec: ai-company-brain/specs/colleague_onboarding.md §6) + 144_crm.sql (the native CRM spine — organizations/contacts/leads/deals, **statuses as data** rather than enums, one `crm_activities` timeline whose four target FKs are all nullable under a CHECK requiring at least one, and a `crm_status_changes` dwell log; also seeds the `crm` feature_catalog row. Contains the schema's **one FK cycle** — `crm_leads.converted_deal_id` ⟷ `crm_deals.lead_id` — closed by a guarded `DO $$` on `pg_constraint`, because `ALTER TABLE … ADD CONSTRAINT` has no `IF NOT EXISTS`. Idempotency is pinned STATICALLY by `tests/unit/test_crm_migration.py`, which reads the file as text: the unit suite runs no database, so an idempotency claim that holds only by inspection is unenforceable. Spec: ai-company-brain/specs/crm_app.md §3) + 145_crm_zoho_sync.sql (what the two-way Zoho sync needs: `zoho_dirty`/`zoho_synced_at` on the four CRM record tables — and **only** those four, since pipeline vocabulary flows down-only and an activity's push signal is its NULL `zoho_id` — plus `crm_zoho_tombstones` (FK-less on purpose: the row it describes is gone by the time anyone reads it) and `crm_sync_cursors` (`module` PK, so the pull cannot silently rewind). Idempotent via `ADD COLUMN IF NOT EXISTS`, which — unlike `ADD CONSTRAINT` — Postgres supports directly, so no guarded `DO $$` is needed. Pinned by the same static `tests/unit/test_crm_migration.py`, which finds BOTH CRM migrations by CONTENT rather than by number. Spec: crm_app.md §7.1) + 158_crm_auto_lead_cursor.sql (WS-26d-autolead: `crm_auto_lead_cursors`, one row per `email_accounts` id, carrying THREE timestamps because they answer three different questions: `activated_at` (the start of the current ON epoch — `received_at > activated_at` is what tells a deep resync's year-old backlog apart from new mail on the shared `process_new_mail` hook), `processed_watermark` (the incremental cursor, advanced over a batch's successful PREFIX only), and `last_run_at` (the dormancy clock, stamped every cycle — a gap beyond an hour CLAMPS the epoch to `now - 1h`, so a flag turned off for weeks and back on mints nothing from the gap except its final hour, which is the recorded residual). ⚠️ The clamp is not a reset: the step runs only when a sync persisted mail, so the cycle that detects the gap always carries the message that woke it. All three NOT NULL: a NULL cursor is a predicate that matches nothing, which reads exactly like a working feature. ⚠️ **It deliberately adds NO unique index on `crm_leads.email`**: the cross-invocation double-mint race is accepted (one visible, hand-deletable duplicate) because a UNIQUE constraint on a column where 1,516 imported rows may already carry duplicates is a deploy-blocking migration of exactly the shape 148 had to defuse. Inert until `CRM_AUTO_LEAD` is flipped, which is OWNER-GATE. Spec: crm_app.md §9 WS-26d-autolead). ⚠️ `CREATE TABLE IF NOT EXISTS` means a column or constraint added to an ALREADY-APPLIED migration file is silently skipped on that deployment; 143, 144 and 145 have never been applied anywhere (merging 143 is the OWNER-GATE), which is the only reason they are still editable in place.
 
 ## Conventions
 - Postgres migrations are numbered SQL files

infra/postgres/158_crm_auto_lead_cursor.sql

@@ -0,0 +1,160 @@
+-- 158_crm_auto_lead_cursor.sql — WS-26d-autolead
+--
+-- ⚠️ Numbered 158, not 157: open PR #399 (`157_projects_recurrence.sql`) holds
+--    157. Two migrations sharing a number replay in filename order against the
+--    wrong schema, so the ladder carries a deliberate reservation gap at 157
+--    until that PR lands. `tests/unit/test_crm_auto_lead.py` finds this file by
+--    CONTENT rather than by number, so a further renumber in review is free.
+--
+-- What: one row per email account recording the current auto-lead ON epoch for
+--       that mailbox, how far the step has processed, and when it last ran.
+-- Why:  the auto-lead step hangs off `process_new_mail`, and that hook is
+--       reached by DEEP RESYNCS as well as by new mail. `resync_account` runs
+--       a ~1-year all-folder backfill and then fires the hook; a first-ever
+--       sync of a newly connected mailbox is deep by the same heuristic; and
+--       neither path stamps `rules_held_back_at` (its only writer is
+--       `_backfill_and_clean_job`, which does not go through this hook). A
+--       candidate query of "everything classified" would therefore mint a lead
+--       per unknown external sender across a YEAR of mail the moment a second
+--       mailbox connects — each born `zoho_dirty`, each queued for the live
+--       Zoho tenant within one 600s cycle (D-CRM-9), with no confirmation card
+--       anywhere on a scheduler hook and no delete tool to take them back.
+--
+--       THREE columns, because that is three different questions and no one
+--       column answers more than one of them:
+--
+--         activated_at         the start of the CURRENT ON epoch. The backfill
+--                              discriminator is `received_at > activated_at`:
+--                              mail that ARRIVED before this epoch began mints
+--                              nothing, no matter when a resync gets around to
+--                              classifying it.
+--
+--         processed_watermark  the incremental cursor, compared against
+--                              `rules_processed_at`. It advances only over the
+--                              contiguous prefix of a batch that actually
+--                              wrote its leads, so a failure is never mistaken
+--                              for work done.
+--
+--         last_run_at          when the step last RAN — stamped at the end of
+--                              every cycle, including the ones that considered
+--                              nothing and the ones that stalled. This is what
+--                              detects dormancy: an OFF window, or an outage.
+--
+--                              ⚠️ It is its own column for a NARROW reason, and
+--                              not the one you might expect. Both it and the
+--                              watermark freeze on a mailbox that receives
+--                              nothing, because the step is NOT invoked once
+--                              per scheduler period — `email_ingestion/
+--                              scheduler.py:463-472` reads `synced` off the
+--                              sync result and fires the hook only when mail
+--                              was actually PERSISTED. What separates the two
+--                              columns is a cycle that ran and found no
+--                              CANDIDATES (mail outside the inbox, held back,
+--                              or predating the anchor): that advances
+--                              `last_run_at` and not the watermark. It is also
+--                              the state a deliberate stall holds the watermark
+--                              in, so keying dormancy on the watermark would
+--                              re-anchor past a stall and quietly undo it.
+--
+--       WHAT HAPPENS ON A GAP. When `now() - last_run_at` exceeds the step's
+--       REANCHOR_GAP_SECONDS, the step CLAMPS `activated_at` forward to
+--       `now() - REANCHOR_GAP_SECONDS` — it does NOT re-stamp all three, it
+--       does NOT set anything to `now()`, and it does NOT skip the batch:
+--
+--         * `activated_at`        moves forward to the start of the gap window.
+--         * `processed_watermark` is deliberately left ALONE. OFF-window mail
+--                                 is already excluded by the anchor predicate
+--                                 whatever its `rules_processed_at` says, and
+--                                 moving it would skip the triggering batch a
+--                                 second way.
+--         * `last_run_at`         is advanced by the ordinary end-of-cycle
+--                                 stamp, like any other cycle.
+--
+--       Clamp rather than reset, because the step runs only when a sync
+--       persisted mail — so the cycle that DETECTS the gap is always the cycle
+--       carrying the message that woke it. Resetting the anchor to `now()` and
+--       returning early therefore excluded that message permanently: measured
+--       as no mail 22:00→07:30, a prospect writes at 07:30:50, the step runs at
+--       07:31:05, and the one lead it existed to catch fell fifteen seconds the
+--       wrong side of the anchor its own arrival created. Every night, every
+--       weekend.
+--
+--       ⚠️ ACCEPTED RESIDUAL, and it is not "nothing": mail received in the
+--       FINAL REANCHOR_GAP_SECONDS of the window IS minted — one gap-width of
+--       mail (an hour), drained across however many cycles the per-cycle cap
+--       takes. Everything older stays excluded, which is the fail-closed
+--       property that matters: a missed lead is hand-creatable and visible in
+--       the mailbox; 27 unattended pushes into a live tenant are neither.
+--
+--       ⚠️ There is deliberately NO unique index on `crm_leads.email` to go
+--       with this. Two concurrent `process_new_mail` invocations for one
+--       account can read the same watermark and double-mint; the cost is one
+--       visible, hand-deletable duplicate lead, and the alternative — a UNIQUE
+--       constraint minted on a column where 1,516 imported rows may already
+--       carry duplicates — is a deploy-blocking constraint of exactly the shape
+--       migration 148 had to defuse. The accepted race is recorded in
+--       `crm_app.md` §9 WS-26d-autolead. Do not "fix" it with that index.
+--
+-- Spec: ai-company-brain/specs/crm_app.md §9 WS-26d-autolead (the cursor
+--       paragraph) · D-CRM-9.
+-- Depends on: 17_email_accounts.sql (email_accounts, the FK target) and
+--       144_crm.sql (the CRM spine this cursor guards writes into).
+--
+-- Idempotent: CREATE TABLE / CREATE INDEX IF NOT EXISTS only. No seed, no
+-- ALTER, nothing dropped.
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS crm_auto_lead_cursors (
+    -- The mailbox, not the CRM record: the step is per account because
+    -- `process_new_mail` is. CASCADE because a disconnected mailbox's cursor
+    -- describes nothing — the leads it already minted are CRM rows and are
+    -- untouched by this.
+    account_id          UUID PRIMARY KEY
+                            REFERENCES email_accounts (id) ON DELETE CASCADE,
+
+    -- The start of the current ON epoch. Re-stamped on re-anchor, never
+    -- advanced by ordinary progress. See above.
+    activated_at        TIMESTAMPTZ NOT NULL,
+
+    -- Advanced to MAX(rules_processed_at) of each batch's successful prefix.
+    processed_watermark TIMESTAMPTZ NOT NULL,
+
+    -- Stamped at the end of EVERY cycle, including the ones that considered
+    -- nothing and the ones that stalled. The dormancy clock.
+    last_run_at         TIMESTAMPTZ NOT NULL,
+
+    created_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at          TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- ⚠️ `CREATE TABLE IF NOT EXISTS` above is a NO-OP against a database that
+-- already has the table, so it cannot add a column to one. A scratch or dev
+-- database that applied this file while it was still the two-column version
+-- (it was numbered 157 then, and briefly had no `last_run_at`) would keep the
+-- old shape and every cycle would fail on the missing column. These three
+-- statements are the repair, and they are all no-ops on a fresh database:
+--   * ADD COLUMN IF NOT EXISTS — nullable, because an existing row has no
+--     value to give it and NOT NULL would refuse the ALTER outright;
+--   * the backfill takes the best answer available, in order: the row already
+--     tells us how far the step got, and failing that when it was activated;
+--   * SET NOT NULL then restores the invariant, and is itself a no-op when
+--     the column was created NOT NULL by the CREATE TABLE above.
+ALTER TABLE crm_auto_lead_cursors
+    ADD COLUMN IF NOT EXISTS last_run_at TIMESTAMPTZ;
+
+UPDATE crm_auto_lead_cursors
+   SET last_run_at = COALESCE(processed_watermark, activated_at, now())
+ WHERE last_run_at IS NULL;
+
+ALTER TABLE crm_auto_lead_cursors
+    ALTER COLUMN last_run_at SET NOT NULL;
+
+-- The candidate query reads this row by primary key, so no second index is
+-- needed here. This one supports the operator question the log line raises —
+-- "which mailboxes has auto-lead run on, and when?" — without a seq scan
+-- growing with the number of connected accounts.
+CREATE INDEX IF NOT EXISTS idx_crm_auto_lead_cursors_last_run_at
+    ON crm_auto_lead_cursors (last_run_at);
+
+COMMIT;