Lead Capture and CRM Sync

Purpose

When a visitor submits a public form, this module turns the raw {field: value} payload into a workspace-scoped CRM Lead (created or, on a repeat email, updated without ever overwriting a field a human already filled in), links it to a TrackingContact and the append-only V2 ContactTimeline, and fans the same event out to Meta/LinkedIn/Google Ads conversion APIs; the same module exposes the /forms/leads list, detail and status endpoints the leads inbox reads. The invariant that makes it trustworthy is that lead identity is always (workspace_id, lower(email)), enforced twice independently, once by the application's own dedup query (lead_integration.py 228 to 233) and once by a real Postgres unique index the ORM model does not even declare (uq_leads_workspace_email, Data model), so the same email in two different workspaces can never collide into one lead and the same email in one workspace can never silently fork into two.

When to use

Read it before touching process_submission_lead, extract_email_from_data/extract_name_from_data/extract_phone_from_data/extract_company_from_data, link_lead_to_tracking_contact, link_submission_to_contact, or the /forms/leads* routes; before changing what counts as a duplicate lead, what fields are allowed to overwrite an existing lead, or which of Meta/LinkedIn/Google Ads a submission fires to. Trigger phrases: "lead capture", "form to lead", "lead dedup", "why is this lead a duplicate", "why didn't this submission create a lead", "leads inbox", "TrackingContact", "contact spine", "form_submitted timeline", "conversion event_id".

Not this recipe: the public submit endpoint's own mechanics, bot protection, the embed widget and answer piping, which is forms-leads/public-submission-and-widget (public_forms_router.py's bot/honeypot/rate-limit block at 626 to 661, this recipe only picks up after that block has already let a submission through and only from the line that calls process_submission_lead); form sessions and abandonment recovery, which is forms-leads/form-sessions-and-abandonment-recovery (session_service.py's complete_session never calls process_submission_lead at all, see Gotchas); the actual Meta/LinkedIn/Google Ads CAPI client code (services/meta_capi_service.py, services/linkedin_capi_service.py, services/google_ads_service.py) and the workspace tracking-settings UI that configures them, this recipe only calls them and is not responsible for what they do once called; Divination's own identify implementation (services/divination_client.py), called but not owned here; commerce checkout, the Quentn newsletter sync, the Telegram/email creator notifications and the lead-magnet email delivery, all of which live in the same submit_public_form handler right after the process_submission_lead call (public_forms_router.py 744 to 863) but are separate, best-effort side effects this recipe does not cover.

Stack assumptions

At 62ba4fc: fastapi==0.118.x, synchronous handlers on a sqlalchemy.orm.Session (SQLAlchemy 2.0), Postgres 14 in the test harness (docker-compose.test.yml, port 5434) / matches the baseline prod schema. lead_integration.py's own imports (1 to 31): dataclasses, uuid, json, logging, sqlalchemy.orm.Session, plus the app's own models.forms, models.crm_model, services.divination_client, services.meta_capi_service, services.linkedin_capi_service, services.google_ads_service, services.tracking_settings_service, services.contacts_service (a re-export shim over services/contacts/contact_service.py, confirmed by reading contacts_service.py itself, which is nothing but from services.contacts.contact_service import ...). Foundations assumed: S4 (fastapi-project-layout) for the app and router mount; S5 (auth-dependency-pattern) for get_current_user on every /forms/leads* route; S6 (workspace-scoping) for the WorkspaceID dependency (api/dependencies.py _get_workspace_id, 680 to 779) those same routes take; S8 (alembic-migrations) for the migration chain under Data model; S7 (error-envelope) for how the plain HTTPException(detail=...) calls in forms_router.py are shaped into a response, this module never builds its own envelope.

No environment variable is read directly by lead_integration.py. The CAPI fan-out reads workspace-level TrackingSettings via get_tracking_config (tracking_settings_service.py 93 to 178), which falls back to environment variables only when a workspace has not configured its own: META_PIXEL_ID, META_ACCESS_TOKEN, META_AD_ACCOUNT_ID, LINKEDIN_ACCESS_TOKEN, LINKEDIN_AD_ACCOUNT_ID, LINKEDIN_CONVERSION_ID_LEAD, LINKEDIN_CONVERSION_ID_HIGH_INTENT, GOOGLE_ADS_CUSTOMER_ID, GOOGLE_ADS_CONVERSION_ACTION_LEAD, GOOGLE_ADS_DEVELOPER_TOKEN, GOOGLE_ADS_REFRESH_TOKEN, GOOGLE_ADS_CLIENT_ID/GOOGLE_CLIENT_ID, GOOGLE_ADS_CLIENT_SECRET/GOOGLE_CLIENT_SECRET (tracking_settings_service.py 108 to 175, core/config.py settings). Fail behaviour: every external side effect inside process_submission_lead, the pre-conversion tracking capture (274 to 283), the tracking-contact link (286 to 289), the contact-spine write (296 to 310), the whole Divination/CAPI block (330 to 504), is wrapped in its own try/except that logs a warning and continues; the function's own db.commit() is itself wrapped (507 to 511, rolls back and logs on failure). Nothing inside process_submission_lead can raise past its own body; the one caller in public_forms_router.py still wraps the call in a second try/except anyway (734 to 742).

Data model

Both tables predate this recipe (baseline_prod_schema_001.py) and are extended by three later migrations. leads (models/crm_model.py 70 to 107; DDL baseline_prod_schema_001.py 1175 to 1200):

leads (
  id                    uuid PRIMARY KEY,
  workspace_id          uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  email                 varchar NOT NULL,
  first_name            varchar, last_name varchar, phone varchar, company_name varchar, website varchar,
  source                leadsource,           -- calendly | manual | referral | website_form | cold_outreach | booking | other
  status                leadstatus,           -- visitor | contact | new | contacted | qualified | unqualified | converted | lost
  lead_score            integer,
  calendly_event_id     varchar UNIQUE,
  onboarding_session_id uuid REFERENCES client_onboarding_sessions(id) UNIQUE,
  converted_to_client_at timestamptz,
  assigned_to_id        uuid REFERENCES users(id),
  contact_id            uuid REFERENCES tracking_contacts(id) ON DELETE SET NULL,   -- added by contact_spine_001
  anonymous_id          varchar,              -- pre-conversion tracking session
  tracking_data         text,                 -- JSON, stored as a STRING (see Gotchas), not jsonb
  fingerprint           varchar,
  scraped_data           jsonb, scrape_status varchar, scraped_at timestamptz,       -- outreach scraping, unrelated to this recipe
  created_at timestamptz DEFAULT now(), updated_at timestamptz
);
-- UNIQUE INDEX uq_leads_workspace_email ON leads (workspace_id, lower(email))
--   present in baseline_prod_schema_001.py line 4153 directly; 20260408_add_workspace_id_to_leads.py
--   (revision leads_ws_001) re-derives the same index but is a documented no-op on a
--   freshly-baselined database (its own comment, lines 23 to 26). NOT declared in the
--   SQLAlchemy model's __table_args__ (there is none on Lead) -- the ORM does not know
--   about this constraint, only Postgres enforces it.

form_submissions (models/forms.py 443 to 489; DDL baseline_prod_schema_001.py 1091 to 1109), the columns this recipe writes: email varchar(255) (extracted, indexed), lead_id uuid REFERENCES leads(id) ON DELETE SET NULL, contact_id uuid REFERENCES tracking_contacts(id) ON DELETE SET NULL (added by contact_spine_001, nullable, index ix_form_submissions_contact_id), data jsonb NOT NULL (the raw submitted payload this recipe reads from), tracking_data jsonb DEFAULT '{}' (the UTM/attribution dict, contrast with leads.tracking_data above which is text).

tracking_contacts (models/tracking_model.py 117 to 213; __table_args__ 202 to 213): the columns link_lead_to_tracking_contact/link_submission_to_contact touch are workspace_id, email, status (VisitorStatus enum), lead_id, first_name/last_name/phone/company, total_sessions/total_events/total_page_views. Two composite unique constraints, both added by 20260610_0001_tracking_contacts_workspace_unique.py (revision tracking_contacts_ws_unique_001): uq_tracking_contacts_workspace_email (workspace_id, email) is the identity resolve_or_create_contact matches on; uq_tracking_contacts_workspace_lead_id (workspace_id, lead_id) means one lead can back at most one contact per workspace.

contact_timeline (models/tracking_model.py 263 to 320; created by 20260610_0003_contact_spine_001.py, revision contact_spine_001): append-only, workspace_id, contact_id, kind varchar(100), payload jsonb, denormalized form_submission_id/lead_id (not real FKs, by design, so the audit trail survives deletion of the row it references). This recipe writes exactly one kind, the module constant TIMELINE_FORM_SUBMITTED = "form_submitted" (lead_integration.py 38).

What is NOT stored: no PSP or payment data (this recipe has none); no honeypot field value (public_forms_router.py strips HONEYPOT_FIELDS from data before a FormSubmission is even created, upstream of this recipe); ClientActivityLog.meta_data is written via Python str(dict), not json.dumps (lead_integration.py 321 to 325), so it is not valid JSON despite living in a column named metadata (see Gotchas).

API surface

The endpoint that actually triggers this recipe's Lead/Contact side of the work is NOT in forms_router.py (see Gotchas for why it looks like it should be). It is the public embed submit routes in public_forms_router.py:

Route	Auth	Tenancy	Codes	Notes
POST /f/{embed_code}/submit (public_forms_router.py 591)	none, public	resolved from embed_code → FormTemplate.workspace_id (_find_embed, 606)	200 (incl. a fake success: true when bot protection trips, 654 to 661, so a bot never learns it was blocked); 404 embed/template not found or not public	Calls process_submission_lead at line 735 inside its own try/except (739 to 742); on any exception the submission itself still succeeds, lead_id in the response is simply None
POST /f/{embed_code}/submit-ai (public_forms_router.py 1037)	none, public	same as above	200; 404	Same pattern, calls process_submission_lead at line 1184

Neither call site validates submission.data against the template's FormField definitions first (no ValidationEngine/ConditionalLogicEngine anywhere in public_forms_router.py, confirmed by grep), unlike the authenticated /forms/templates/{id}/submit route which does run both (forms_router.py 374 to 394) but never reaches process_submission_lead at all.

The lead-management endpoints this recipe exposes, all in forms_router.py, all behind get_current_user + WorkspaceID:

Route	Codes	Notes
GET /forms/leads (632 to 788)	200	Joined to FormSubmission (leads with zero submissions are excluded by the join itself, not a filter), ?status=/?search= filters, batches submission counts and per-page tracking-event counts in two grouped queries each (was N+1, per the router's own comments at 685, 697)
GET /forms/leads/{lead_id}/submissions (791 to 829)	200; 404 lead not in workspace	Reimplements its own FormSubmission query rather than calling the module-level get_lead_submissions at lead_integration.py 522 (see Gotchas, that function is unreachable from any router)
PUT /forms/leads/{lead_id}/status (832 to 869)	200; 404 lead not in workspace; 400 invalid LeadStatus value	Body {"status": "<LeadStatus value>"}

The module-level Python entrypoints, not routes, that a rebuild reimplements:

Function	Workspace-scoped	Returns
process_submission_lead(db, submission, template, tracking_data=None) (177 to 519), async	yes, via template.workspace_id (or its owner fallback)	LeadConversionResult(lead, event_id, meta_event), lead is None when no email was extractable
link_lead_to_tracking_contact(db, lead, anonymous_id=None, fingerprint=None, email=None, workspace_id=None) (705 to 817)	yes	the TrackingContact or None on failure (own SAVEPOINT)
link_submission_to_contact(db, *, submission, workspace_id, email, ..., lead=None, template=None) (820 to 892)	yes, via contacts_service.resolve_or_create_contact	the TrackingContact or None on failure/no email (own SAVEPOINT)
get_lead_by_email(db, email) (540 to 542)	no	one Lead or None, matches across ALL workspaces
get_leads_with_submissions(db, status=None, source=None, search=None, ...) (545 to 583)	no	list of Lead, no workspace_id parameter exists
get_lead_submissions(db, lead_id, limit=50, offset=0) (522 to 537)	no (takes no workspace_id)	list of FormSubmission

Application logic

process_submission_lead, in commit order:

1. extract_email_from_data(submission.data) (196 to 197): normalizes keys to lowercase, checks EMAIL_FIELD_NAMES (email, e-mail, email_address, emailaddress, contact_email, your_email) in that order, requires the value to be a str containing @, returns it lowercased and stripped. No email anywhere in the payload: returns LeadConversionResult(lead=None) immediately (199 to 201), nothing downstream runs, no Lead/Contact/ClientActivityLog row is written.
2. submission.email = email (204), so the submission itself is queryable by email even before any lead exists.
3. extract_name_from_data/extract_phone_from_data/extract_company_from_data (207 to 209): each does its own lowercase-key scan over its own alias list (NAME_FIELD_NAMES, LAST_NAME_FIELD_NAMES, PHONE_FIELD_NAMES, COMPANY_FIELD_NAMES, 49 to 67), all deliberately including German (vorname, nachname, telefon, firma) and French (prenom, nom) aliases, per the module's own comment (43 to 48) documenting a dated regression where every German-language form produced anonymous leads until these aliases were added. extract_name_from_data only splits a value on its first space when the matched field name is name/full_name/fullname (117); a first_name field with a space in it, e.g. "Mary Jane", is kept whole (test test_first_name_not_split_even_with_space).
4. anonymous_id/fingerprint read straight off tracking_data (212 to 213), no validation.
5. Workspace resolution (215 to 226): template.workspace_id if set; otherwise a raw-SQL fallback, SELECT id FROM workspaces WHERE owner_id = :uid ORDER BY created_at ASC LIMIT 1 against template.created_by. No workspace resolvable at all: returns LeadConversionResult(lead=None), logged as a warning.
6. Dedup and create/update (228 to 269): db.query(Lead).filter(Lead.email == email, Lead.workspace_id == workspace_id).first(). Found: every optional field (first_name, last_name, phone, company_name, anonymous_id, fingerprint) is set ONLY if the incoming value is truthy AND the lead's current value is falsy, never overwriting a value that is already there (240 to 252). Not found: a new Lead(source=LeadSource.website_form, status=LeadStatus.new, ...) is created and db.flush()ed for its id.
7. Pre-conversion tracking capture (271 to 283), only if anonymous_id or fingerprint: _capture_pre_conversion_tracking_from_db (586 to 702) looks up the TrackingSession by anonymous_id, pulls every TrackingEvent for that session (and, if a fingerprint is known, every session sharing that fingerprint, no workspace filter on this inner query), and builds a summary dict (page views, distinct pages visited, CTA/video counts, max scroll depth, time on site). The result is written to lead.tracking_data via json.dumps ONLY if the lead does not already have tracking data (279), so a second submission from the same lead never overwrites the first journey snapshot. Wrapped in try/except, a failure here only logs a warning.
8. link_lead_to_tracking_contact (286 to 289, body at 705 to 817): its own db.begin_nested() SAVEPOINT. Finds or creates a TrackingContact by (workspace_id, email.lower()); on create, seeds it from the lead's own fields and sets status = VisitorStatus.lead; on an existing contact missing lead_id, links it and bumps status to lead. If anonymous_id is given, links the matching TrackingSession (workspace-scoped) and its TrackingEvents to the contact. If fingerprint is given, backfills lead.fingerprint if unset and merges every OTHER unlinked session sharing that fingerprint in the SAME workspace onto the contact. Recomputes total_sessions/total_events/total_page_views by counting, not incrementing, so the numbers are always exact even after a merge. A failure anywhere rolls back only the savepoint (814 to 817).
9. link_submission_to_contact (296 to 310, body at 820 to 892): the "canonical Contact writer". Calls the SHARED contacts_service.resolve_or_create_contact (own SAVEPOINT, 852), stamps submission.contact_id, and appends one ContactTimeline row via append_timeline with kind=TIMELINE_FORM_SUBMITTED, a payload of {form_name, template_id, form_category} when a template was passed, and the denormalized form_submission_id/lead_id refs. Idempotent: re-running for the same email reuses the same contact (module docstring 846 to 847; test_reuses_existing_contact_and_timeline_is_append_only), the timeline itself is append-only so two submissions produce two events.
10. submission.lead_id = lead.id (313).
11. ClientActivityLog(activity_type=ClientActivityType.form_submitted, ...) (316 to 327), meta_data=str({...}), a Python dict repr, NOT json.dumps (see Gotchas).
12. Divination + CAPI fan-out (329 to 504), all inside one try/except that only logs on failure (503 to 504): divination.identify(...) with attribution stripped of None values; a single event_id = uuid.uuid4() shared across all three ad platforms for cross-platform deduplication (364); event_source_url prefers tracking_data['page_url'], falls back to landing_page; the conversion event name per platform comes from a form-specific mapping (get_form_conversion_mapping) if enabled, else CATEGORY_TO_CONVERSION_EVENT keyed by template.category (71 to 79, e.g. lead_magnet → Lead, appointment → Schedule). consent_blocks_marketing (398 to 400, true only when tracking_data['marketing_consent'] is False) gates ONLY the Meta send (403); LinkedIn (451) and Google Ads (478) have no equivalent consent check, gated only by their own workspace send_form_submissions flag.
13. db.commit() (507 to 511), its own try/except: on failure, rolls back and logs, the function still returns normally.
14. Returns LeadConversionResult(lead=lead, event_id=locals().get('event_id'), meta_event=locals().get('meta_event')) (515 to 519): event_id/meta_event are None if the try-block in step 12 died before they were assigned, e.g. a workspace tracking-settings load error.

Invariants

• No extractable email means no Lead, no TrackingContact, no ContactTimeline, no activity log row, at all (test_process_no_email_returns_no_lead, asserts Lead count stays 0).
• Dedup is scoped to (workspace_id, email): the same email in a different workspace always creates a SEPARATE lead, never reuses one (test_process_dedup_is_workspace_scoped).
• An existing lead's already-set fields are never overwritten by a later submission, only empty fields are backfilled (test_process_updates_existing_lead_without_overwriting, explicitly checks status stays qualified and first_name/company_name stay the original values).
• The same email in two different workspaces resolves to two DISTINCT TrackingContact rows, never one shared across tenants (test_workspace_isolation_same_email_distinct_contacts, on the contacts_service layer link_submission_to_contact calls into).
• Re-running link_submission_to_contact for the same email in the same workspace reuses the one TrackingContact (idempotent resolve) while the form_submitted timeline itself is append-only, one row per submission (test_reuses_existing_contact_and_timeline_is_append_only).
• A TrackingSession is only ever linked to a TrackingContact within its own workspace_id, both by anonymous_id match (763 to 766) and by fingerprint merge (785 to 789); the module comment at 759 to 762 names this explicitly as a fix for a prior cross-tenant leak.
• link_lead_to_tracking_contact and link_submission_to_contact each fail inside their own SAVEPOINT (db.begin_nested()), so either one raising rolls back only that nested unit, never the Lead row already flushed earlier in the same call (module code at 719/852; no test in the two given test files forces an exception inside either savepoint to prove the outer lead survives, so this is read from the code, not proven by a test at this commit).

Frontend

frontend_modules for this recipe is forms-leads/ui/form-builder-page, the surface for building the form whose submissions this module turns into leads. The actual reference render for the endpoints in this recipe (GET/PUT /forms/leads*) is a DIFFERENT, not-yet-named-in-this-kit screen: alchemy_labs/frontend/src/pages/FormLeads.tsx (1365 lines, routed at frontend/src/App.tsx 986, calls GET /api/forms/leads, GET /api/forms/leads/{id}/submissions and PUT /api/forms/leads/{id}/status at lines 256, 287 and 433 respectively, exactly matching the routes above). No recipe in this kit currently documents a "leads inbox" frontend module; that is a genuine gap, not something to guess a module id for.

Minimum admin/debug surface: the two grouped-query endpoints above ARE the admin surface, there is no separate one. A quick health check by hand: SELECT count(*) FROM leads WHERE workspace_id = :ws AND created_at > now() - interval '1 day' against the count of /f/*/submit calls in the access log for the same window is the fastest way to notice process_submission_lead silently swallowing an exception (step 12's outer try/except logs a warning but the submission still returns 200).

Security & tenancy

Auth dependency: the two routes that actually create leads (POST /f/{embed_code}/submit, /submit-ai) have NONE, by design, public_forms_router.py mounts under /f with no Depends(get_current_user) anywhere in either handler. Tenancy on that path comes ENTIRELY from resolving embed_code to a FormTemplate.workspace_id server-side; nothing the caller sends names a workspace directly. The /forms/leads* routes are the opposite: get_current_user (S5) plus the WorkspaceID dependency (S6, X-Workspace-ID header, access-checked against the authenticated user, api/dependencies.py 680 to 779), and every query in forms_router.py's handlers explicitly filters Lead.workspace_id == str(workspace_id).

Fail direction, inconsistent by function: process_submission_lead's own dedup query and link_lead_to_tracking_contact/link_submission_to_contact are all workspace-scoped and closed; but get_lead_by_email and get_leads_with_submissions (lead_integration.py 540 to 583) take NO workspace_id parameter at all and are open by construction, a lead in workspace A is findable by get_lead_by_email regardless of who is asking. Confirmed at this commit that neither function is called from anywhere except the two given test files (grep across api/ and services/ for both names), so this is a latent gap in unused code, not a live leak, but it means EITHER function reused as-is for a new endpoint reintroduces cross-tenant lookup with no additional guard.

Attacker-controlled inputs on the public path: the entire submission.data dict, key names and values both, every extractor (extract_email_from_data etc.) trusts whatever key names happen to match its alias list, and (confirmed by grep) NOTHING validates data against the template's real FormField definitions before process_submission_lead reads it, unlike the authenticated /forms/templates/{id}/submit route. tracking_data['marketing_consent'] is caller-supplied and gates the Meta CAPI send (see Application logic step 12); nothing signs or verifies it, a caller claiming marketing_consent: true when the real visitor said no is indistinguishable from a genuine consent signal at this layer.

Secrets and PII at rest: ClientActivityLog.meta_data and lead.tracking_data both carry form/session metadata, not raw CAPI tokens; the actual Meta/LinkedIn/Google Ads access tokens live in workspace TrackingSettings, read (and, per its own module docstring, dual-read decrypted) by get_workspace_tracking_settings, out of this recipe's scope. Rate limiting on the public submit path is entirely public_forms_router.py's own bot-protection block (out of scope, see When to use), this recipe has none of its own.

Tests

Run 2026-09-12 at 62ba4fc, against the already-running deap-postgres-test container (Postgres 14, port 5434, the same docker-compose.test.yml postgres-test service tests/conftest.py defaults to), from alchemy_labs/backend with its own .venv:

TEST_DATABASE_URL=postgresql://test_user:test_password@localhost:5434/test_db \
    python -m pytest \
    tests/v2/coverage/test_forms_g5_lead_integration_field_extraction_lead_creation_dedup_process_submission_lead_.py \
    tests/test_forms_contact_writer.py \
    -q --no-cov -o addopts="" -p no:cacheprovider

prints 39 passed (326 warnings, all Pydantic v2/SQLAlchemy deprecation noise unrelated to this recipe, 9.75s). grep -c "^def test_\|async def test_" on the two files individually prints 33 (G5) and 6 (contact writer), 33 + 6 = 39, matching exactly, no test is skipped or xfailed at this commit (unlike the billing-credits scratch-database suite, these two files carry no pytestmark/skip gate on the database).

Hermetic vs live: the patched_externals fixture (G5 file, 158 to 174) patches li.divination, li.get_tracking_config (returns a MagicMock with test_mode=True and every platform's send_form_submissions=False) and li.get_form_conversion_mapping (returns None), so NEITHER test file's 39 tests ever call the real meta_capi/linkedin_capi/google_ads services or a real Divination endpoint; both are fully hermetic against network. Known-answer vectors: extract_email_from_data case-insensitive key AND value normalization ({"Email": " Foo@Bar.COM "} → "foo@bar.com"), the @ guard rejects a value with the right key but no @, a non-string value is ignored entirely; extract_name_from_data's full-name split happens ONLY for name/full_name/fullname keys and splits on the FIRST space only ("Jean Luc Picard" → ("Jean", "Luc Picard")), name is checked before first_name in NAME_FIELD_NAMES so a form carrying both prefers name.

What a new app adds and this branch lacks: no test in either given file exercises process_submission_lead's own db.query(Lead)...first() dedup racing a concurrent second submission for the same brand-new email (contrast with the billing-credits recipe's two explicit threading.Barrier concurrency tests), and the Lead create/query path itself has no savepoint or IntegrityError handling the way link_lead_to_tracking_contact/link_submission_to_contact do, so this is a genuinely untested edge, not merely an undocumented one; no test in these two files exercises the actual meta_capi.send_event/linkedin_capi.send_lead_event/google_ads.send_lead_event call bodies, only that they are correctly SKIPPED when the workspace config says so.

Deploy & ops

No config keys are owned by this recipe directly (see Stack assumptions for the CAPI env var fallbacks, owned by tracking_settings_service.py). Migrations, in the order that matters for this recipe's tables: baseline_prod_schema_001.py creates leads and form_submissions with uq_leads_workspace_email already present as a raw functional unique index (line 4153); 20260408_add_workspace_id_to_leads.py (leads_ws_001) re-derives the same column/index sequence but is a documented no-op on a database that started from the baseline (its own comment, lines 23 to 26); 20260610_0001_tracking_contacts_workspace_unique.py (tracking_contacts_ws_unique_001) adds the two composite uniques on tracking_contacts; 20260610_0003_contact_spine_001.py (contact_spine_001) creates contact_timeline and adds the nullable contact_id FK to leads, orders, form_submissions, calendar_bookings and deals (its own _CONTACT_FK_TABLES list, lines 33 to 39), additive only, guarded to be safely re-runnable.

Scheduler: this recipe itself runs entirely inline inside the HTTP request, no job of its own. A related but separate job, run_contact_linking_job (services/scheduled_jobs.py 393 to 406, IntervalTrigger(hours=1), coalesce=True, max_instances=1, registered inside the advisory-lock singleton block), backfills contact_id on OLDER leads/orders/form_submissions/calendar_bookings/deals rows created before the contact spine existed; it does not touch rows this recipe already links inline and is out of this recipe's scope.

Argus-style checks: POST /f/{embed_code}/submit with a valid data.email against a real public template returns 200 with a non-null lead_id in the body; SELECT count(*) FROM leads l JOIN form_submissions s ON s.lead_id = l.id WHERE s.contact_id IS NULL AND s.created_at > now() - interval '1 hour' should stay near 0 in steady state, a nonzero count means link_submission_to_contact is failing silently inside its own try/except (309 to 310) for recent submissions. Rollback: no data is ever deleted by this recipe (Lead/TrackingContact/ContactTimeline rows all persist through a code rollback); a rolled-back deploy leaves whatever leads/contacts/timeline rows it already wrote in place, the next deploy's writes are still idempotent on top of them.

Gotchas

• 2026-09-12 (measured at 62ba4fc): forms_router.py, one of this recipe's own named source files, does NOT call process_submission_lead anywhere (grep -n "process_submission_lead" api/routers/forms_router.py prints nothing). Its authenticated /forms/templates/{id}/submit (338 to 440) and the session-based /forms/sessions/{token}/complete (session_service.py complete_session, 212 to 279) both create a bare FormSubmission row and stop. Only the PUBLIC, unauthenticated embed routes in public_forms_router.py (/f/{embed_code}/submit and /submit-ai) ever create a Lead. A rebuild that wires process_submission_lead into the authenticated submit path or session completion would be adding new behavior, not restoring removed behavior.
• 2026-09-12 (measured at 62ba4fc): get_lead_by_email and get_leads_with_submissions (lead_integration.py 540 to 583) are dead code at this commit, tested, but called from nowhere in api/ or services/ outside the test files themselves (confirmed by grep). Neither takes a workspace_id, so wiring either into a new endpoint without adding workspace scoping first reintroduces a cross-tenant lookup.
• 2026-09-12 (measured at 62ba4fc): ClientActivityLog.meta_data is written with Python str({...}) (lead_integration.py 321 to 325), not json.dumps. The column is genuinely named metadata and genuinely intended to hold structured data, but its stored value is a Python dict repr (single quotes, None/True/False unquoted), not parseable JSON. json.loads on it will raise; only ast.literal_eval round-trips it.
• 2026-09-12 (measured at 62ba4fc): leads.tracking_data is a text column written via json.dumps (lead_integration.py 280), while form_submissions.tracking_data is a real jsonb column holding the same shape of data natively. Two different storage representations of "tracking data" coexist on two tables this recipe writes to in the same call.
• 2026-09-12 (measured at 62ba4fc): consent gating (consent_blocks_marketing, tracking_data.get('marketing_consent') is False) is checked only before the Meta CAPI send (403). The LinkedIn (451) and Google Ads (478) sends have no equivalent check, gated only by their own workspace send_form_submissions flag. Whether this is intentional (Meta being the platform under the most scrutiny) or an oversight is not determinable from the code alone.
• 2026-09-12 (measured at 62ba4fc): the public submit path (public_forms_router.py /f/{embed_code}/submit, /submit-ai) never runs ValidationEngine/ConditionalLogicEngine against the template's FormField definitions before process_submission_lead reads submission.data (confirmed absent by grep), unlike the authenticated /forms/templates/{id}/submit route (forms_router.py 374 to 394) which does, but which never reaches this recipe's lead-creation logic at all (see the first Gotcha above).
• 2026-09-12 (measured at 62ba4fc): the Lead dedup query-then-create in process_submission_lead (228 to 269) takes no row lock and has no IntegrityError handling, unlike link_lead_to_tracking_contact/link_submission_to_contact right after it, which both wrap their writes in a SAVEPOINT. A real Postgres unique index (uq_leads_workspace_email) backs the identity this query assumes, so two concurrent submissions racing on the same brand-new email in the same workspace could raise an uncaught IntegrityError on the second db.flush(); no test in either given test file exercises this race, and the function's own outer behavior for that specific exception is not determinable from the code (it is not one of the named try/except blocks, which start after the lead is already created).

Reference implementation

Paths relative to the repo root; all at main 62ba4fc41009b454e1c033e5036c7e8577db2061 unless noted (verified git branch --show-current = main, git log -1 --format=%H = this commit, i.e. this recipe is measured on the tip of main, not a feature branch).

• alchemy_labs/backend/services/forms/lead_integration.py (893 lines): field alias lists (41 to 67); extract_email_from_data/extract_name_from_data/extract_phone_from_data/extract_company_from_data (82 to 160); LeadConversionResult (163 to 175); process_submission_lead (177 to 519); get_lead_submissions/get_lead_by_email/get_leads_with_submissions (522 to 583, all three unreachable from any router, see Gotchas); _capture_pre_conversion_tracking_from_db (586 to 702); link_lead_to_tracking_contact (705 to 817); link_submission_to_contact (820 to 892).
• alchemy_labs/backend/api/routers/forms_router.py (1572 lines): submit_form (338 to 440, does NOT call this recipe); get_form_leads (632 to 788); get_lead_submissions route (791 to 829); update_lead_status (832 to 869).
• alchemy_labs/backend/api/routers/public_forms_router.py (1815 lines): submit_public_form (591 to 989), the process_submission_lead call (735) and its surrounding try/except (734 to 742); submit_public_form_ai (1037 onward), the second call (1184).
• alchemy_labs/backend/models/forms.py: FormTemplate (18 to 137); FormSubmission (443 to 519, contact_id 455 to 460).
• alchemy_labs/backend/models/crm_model.py: LeadSource/LeadStatus (16 to 35); Lead (70 to 107); ClientActivityLog (183 to 197).
• alchemy_labs/backend/models/tracking_model.py: TrackingContact (117 to 213, __table_args__ 202 to 213); ContactTimeline (263 to 320).
• alchemy_labs/backend/services/contacts/contact_service.py: resolve_or_create_contact (79 to 222); append_timeline_event (224 to 306); append_timeline (308 to 327). Re-exported at alchemy_labs/backend/services/contacts_service.py (a pure shim, whole file in scope).
• alchemy_labs/backend/services/tracking_settings_service.py: get_tracking_config (93 to 178).
• alchemy_labs/backend/services/scheduled_jobs.py: contact_linking_job registration (393 to 406).
• alchemy_labs/backend/migrations/versions/baseline_prod_schema_001.py: leads DDL (1175 to 1200), form_submissions DDL (1091 to 1109), uq_leads_workspace_email (4151 to 4153).
• alchemy_labs/backend/migrations/versions/20260408_add_workspace_id_to_leads.py (leads_ws_001); 20260610_0001_tracking_contacts_workspace_unique.py (tracking_contacts_ws_unique_001); 20260610_0003_contact_spine_001.py (contact_spine_001, _CONTACT_FK_TABLES 33 to 39).
• Tests: alchemy_labs/backend/tests/v2/coverage/test_forms_g5_lead_integration_field_extraction_lead_creation_dedup_process_submission_lead_.py (528 lines, 33 tests), alchemy_labs/backend/tests/test_forms_contact_writer.py (263 lines, 6 tests).

Deviations of the reference from an ideal implementation, all dated 2026-09-12: forms_router.py never calls this recipe despite owning the endpoints the recipe's own scope description implies it should; two of the six module-level functions are dead code with no workspace scoping; the Lead create path has no concurrency guard where its siblings do; consent gating is asymmetric across the three ad platforms; ClientActivityLog.meta_data is not valid JSON despite the column's intent.

Acceptance checks

Run from alchemy_labs/backend at 62ba4fc (or an app built to this recipe), against a Postgres reachable at TEST_DATABASE_URL. Expected observations are exact, as measured 2026-09-12.

1. TEST_DATABASE_URL=postgresql://test_user:test_password@localhost:5434/test_db python -m pytest tests/v2/coverage/test_forms_g5_lead_integration_field_extraction_lead_creation_dedup_process_submission_lead_.py tests/test_forms_contact_writer.py -q --no-cov -o addopts="" -p no:cacheprovider prints 39 passed.
2. grep -c "^def test_\|async def test_" tests/v2/coverage/test_forms_g5_lead_integration_field_extraction_lead_creation_dedup_process_submission_lead_.py tests/test_forms_contact_writer.py prints 33 and 6, summing to the 39 passed of check 1.
3. grep -n "process_submission_lead" api/routers/forms_router.py prints nothing (exit code 1); grep -c "process_submission_lead" api/routers/public_forms_router.py prints 4 (one import, two call sites, one comment).
4. grep -n "uq_leads_workspace_email" migrations/versions/baseline_prod_schema_001.py shows the CREATE UNIQUE INDEX uq_leads_workspace_email ON public.leads USING btree (workspace_id, lower((email)::text)) line, confirming the workspace+email uniqueness this recipe's dedup query assumes is a real Postgres constraint, not just application logic.
5. grep -rln "get_lead_by_email\|get_leads_with_submissions" api/ services/ | grep -v lead_integration.py | grep -v /tests/ prints nothing, confirming both functions are unreachable from any router at this commit.
6. grep -c "meta_data=str(" services/forms/lead_integration.py prints 1, confirming the single call site that writes the non-JSON activity-log payload named in Gotchas.
7. git branch --show-current prints main and git log -1 --format=%H prints 62ba4fc41009b454e1c033e5036c7e8577db2061, confirming maturity: prod's premise, this recipe is measured on the tip of the branch that ships.