Leads, Deals and CRM Management

Purpose

A workspace's sales team can create, read, update, delete and page through their own leads, the one deal each lead can carry through a fixed sales-pipeline stage list, freeform notes and a shared tag vocabulary, and can pull up the subset of leads that arrived with a form submission attached, all of it scoped to their workspace_id and none of it visible to any other workspace (crm_service.py 19 to 160, crm_router.py throughout). The invariant that makes it trustworthy is that every read and write is filtered by the WorkspaceID the auth layer resolved for the caller, never by a workspace value taken from the request body or the path (get_lead, crm_service.py 28 to 36, is the pattern every other lookup in the file repeats: fetch by id AND workspace, 404 rather than 403 on a mismatch, so a foreign id looks identical to a nonexistent one).

When to use

Trigger phrases: "list leads", "CRM inbox", "leads with form submissions", "create a deal", "move a deal to a stage", "pipeline stage", "add a note to a lead", "client tags", "Page[T]", "leads-with-submissions", "why is /api/crm/leads returning double the prefix", "why did tags leak across workspaces". Read this before touching CrmService or any handler in crm_router.py, or before changing Lead, Deal, ClientTag, LeadTag, ClientNote or ClientActivityLog in crm_model.py.

Not this recipe: the cold-outreach campaign system mounted at the same /api/crm prefix by a separate outreach_router.py (app_fastapi.py 878, tag "Outreach"); it imports and queries the same Lead model (outreach_router.py 17, 105) but owns its own campaign/sequence tables and its own endpoints, a neighbouring recipe; the calendar booking to lead bridge that creates Lead(source="booking") rows (LeadSource.booking, added by migration leadsource_booking_001, 2026-09-07) is the producer of some of the rows this recipe manages, not this recipe's subject; the V2 contact spine (contact_id on Lead and Deal, backfilled by an idempotent linking job) is read here only as a plain nullable column, its own resolve/backfill logic lives elsewhere; ClientActivityLog is defined in crm_model.py and read back nowhere in crm_service.py or crm_router.py at this commit, a separate write path this recipe does not cover.

Stack assumptions

At 62ba4fc4: fastapi==0.118.2, synchronous handlers on sqlalchemy.orm.Session (SQLAlchemy==2.0.43), pydantic==2.12.0, psycopg2-binary==2.9.10, alembic==1.16.5, Postgres 15 in prod (docker-compose.prod.yml) / 14 in the test harness (docker-compose.test.yml, service postgres-test on :5434). Foundations assumed: S4 (fastapi-project-layout) for mount_router and the app; S5 (auth-dependency-pattern) for ActiveUser (get_current_active_user, works with both JWT and API-key auth per the 2026-04-17 fix below); S6 (workspace-scoping) for WorkspaceID (_get_workspace_id, dependencies.py 680 to 779), whose 403/400/409 fail shapes this router inherits unchanged; S8 (alembic-migrations) for the baseline pg_dump migration these six tables come from; S9 (test-layout) for the transactional db_session fixture the pagination tests run against.

No dedicated environment variable governs this recipe: it reads no feature flag and no credential of its own. crm_router.py imports DatabaseSession (plain get_db, no Row-Level Security context), not DatabaseSessionWithRLS; the workspace filter here is entirely application-level (every query's own .filter(... == workspace_id)), not backed by the Postgres RLS policies described in _shared/workspace-scoping.

Data model

Six tables, all created verbatim from a pg_dump of live staging (baseline_prod_schema_001.py, Create Date: 2026-08-02; column shapes below are that dump, TABLE: leads 1173 to 1201, TABLE: deals 967 to 985, TABLE: client_tags 876 to 887, TABLE: lead_tags 1163 to 1171, TABLE: client_notes 816 to 828, TABLE: client_activity_logs 801 to 814); FKs and uniques were added afterward by deferred_fk_phantom_tables_001.py since a phantom table's constraints could not be trusted to apply cleanly until every referenced table existed.

leads (
  id                    uuid PRIMARY KEY,
  workspace_id          uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,  -- added 2026-04-08, migration leads_ws_001
  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 INDEX ix_leads_calendly_event_id
  onboarding_session_id uuid REFERENCES client_onboarding_sessions(id),
  converted_to_client_at timestamptz,
  assigned_to_id        uuid REFERENCES users(id),
  contact_id            uuid REFERENCES tracking_contacts(id) ON DELETE SET NULL,  -- V2 contact spine, nullable, out of scope
  anonymous_id          varchar, tracking_data text, fingerprint varchar,          -- pre-conversion tracking
  scraped_data          jsonb, scrape_status varchar, scraped_at timestamptz,      -- outreach scraping
  created_at            timestamptz DEFAULT now(), updated_at timestamptz
);
INDEX ix_leads_workspace_id (workspace_id); INDEX ix_leads_email (email);          -- plain, not unique
UNIQUE INDEX uq_leads_workspace_email (workspace_id, lower(email));                -- the real duplicate guard, case-insensitive

deals (
  id            uuid PRIMARY KEY,
  lead_id       uuid NOT NULL REFERENCES leads(id), UNIQUE (lead_id),              -- uq_deals_lead_id: one deal per lead
  project_id    uuid REFERENCES client_projects(id), UNIQUE (project_id),          -- uq_deals_project_id
  contact_id    uuid REFERENCES tracking_contacts(id) ON DELETE SET NULL,
  deal_name     varchar NOT NULL,
  deal_value    numeric(10,2),          -- model default 12000.00, no CHECK enforcing it in the DB
  currency      varchar(3),             -- model default 'CHF'
  stage         dealstage,              -- lead | appointment_booked | essence_submitted | payment_pending | active_project | completed | lost
  probability   integer,                -- a percentage, unvalidated (no CHECK 0-100 at the DB or the schema)
  expected_close_date timestamptz, actual_close_date timestamptz, lost_reason text,
  created_at timestamptz DEFAULT now(), updated_at timestamptz
);
-- deals carries no workspace_id of its own; tenancy is always inherited by joining leads

client_tags (
  id uuid PRIMARY KEY, workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  name varchar NOT NULL, category tagcategory, color varchar(7),
  created_at timestamptz DEFAULT now(), updated_at timestamptz
);
UNIQUE (workspace_id, name);  -- uq_client_tags_workspace_name (model __table_args__)

lead_tags (id uuid PK, lead_id uuid NOT NULL REFERENCES leads(id), tag_id uuid NOT NULL REFERENCES client_tags(id), created_at timestamptz)
-- plain association table; neither FK carries ON DELETE, see Gotchas

client_notes (id uuid PK, lead_id uuid REFERENCES leads(id), project_id uuid REFERENCES client_projects(id),
              note_text text NOT NULL, is_pinned boolean, created_by_id uuid NOT NULL REFERENCES users(id),
              created_at timestamptz DEFAULT now(), updated_at timestamptz)

client_activity_logs (id uuid PK, lead_id uuid REFERENCES leads(id), project_id uuid REFERENCES client_projects(id),
              activity_type clientactivitytype NOT NULL, activity_title varchar NOT NULL, activity_description text,
              metadata text,  -- JSON stored as string; column is named "metadata", mapped attr is meta_data (crm_model.py 191)
              created_by_id uuid REFERENCES users(id), created_at timestamptz DEFAULT now())

What is NOT stored: a balance or price beyond deals.deal_value/currency (this recipe never touches Commerce); a raw form submission body (that lives in form_submissions, joined by lead_id, see API surface); anything for ClientActivityLog beyond the columns above: nothing in crm_service.py or crm_router.py at this commit ever writes a row to client_activity_logs, despite the model existing and stage_changed/note_added members sitting unused in ClientActivityType.

API surface

All routes mounted at /api/crm (app_fastapi.py 877, mount_router(app, crm_router.router, prefix="/api/crm", tags=["CRM"], surfaces=LABS_SURFACES)). Page[T] (crm_router.py 30 to 36) is {items: T[], total: int, limit: int, offset: int, has_more: bool}; only three routes return it, the rest return a plain list or a bare object, see the Notes column.

Route	Auth	Tenancy	Codes	Notes
GET /leads (41 to 71)	ActiveUser	WorkspaceID, Lead.workspace_id filter	200 Page[LeadResponse]	limit Query(50, ge=1, le=100), offset Query(0, ge=0); ordered created_at desc; two queries per call (a COUNT then the page), flat regardless of row count (test_crm_leads_query_count_constant)
POST /leads (73 to 76)	ActiveUser	body has no workspace field; workspace_id from WorkspaceID	201 LeadResponse; 409 if email already exists in this workspace	duplicate check is case-sensitive ==, the DB unique index is lower(email), see Gotchas
GET /leads/{lead_id} (78 to 81)	ActiveUser	id + WorkspaceID	200; 404
PUT /leads/{lead_id} (83 to 86)	ActiveUser	id + WorkspaceID	200; 404	exclude_unset=True, partial update
DELETE /leads/{lead_id} (88 to 92)	ActiveUser	id + WorkspaceID	204; 404	ORM cascade deletes the lead's Deal, ClientNote rows and ClientActivityLog rows (cascade="all, delete-orphan", crm_model.py 109, 111, 112); lead_tags rows are cleared by SQLAlchemy's own secondary-table handling, not a DB ON DELETE
GET /deals (96 to 130)	ActiveUser	WorkspaceID, joined through Lead	200 Page[DealResponse]	same Page[T] shape and query-count guarantee as /leads, joined via Deal.lead_id == Lead.id
POST /deals (132 to 135)	ActiveUser	get_lead first (404 if the lead isn't in this workspace)	201 DealResponse	no duplicate-lead-id guard in the service; a second POST for the same lead_id reaches db.commit() and raises an unhandled IntegrityError on uq_deals_lead_id, see Gotchas
GET /deals/{deal_id} (137 to 140)	ActiveUser	id + workspace via join	200; 404
PUT /deals/{deal_id} (142 to 145)	ActiveUser	id + workspace via join	200; 404	exclude_unset=True; setting stage does not append to client_activity_logs
GET /leads/{lead_id}/notes (149 to 152)	ActiveUser	get_lead first	200 List[NoteResponse]	not paginated; unbounded
POST /leads/{lead_id}/notes (154 to 157)	ActiveUser	get_lead first	200 NoteResponse (not 201)	created_by_id = current_user.id
GET /tags (161 to 163)	ActiveUser	WorkspaceID	200 List[TagResponse]	not paginated
POST /tags (165 to 167)	ActiveUser	workspace_id from WorkspaceID	201 TagResponse	no duplicate-name guard in the service; a second name in the same workspace hits uq_client_tags_workspace_name and raises unhandled
PUT /tags/{tag_id} (169 to 171)	ActiveUser	id + WorkspaceID	200; 404
DELETE /tags/{tag_id} (173 to 176)	ActiveUser	id + WorkspaceID	204; 404	lead_tags.tag_id carries no ON DELETE; deleting a tag still attached to a lead raises an unhandled FK violation, see Gotchas
GET /leads/{lead_id}/submissions (181 to 205)	ActiveUser	get_lead first, then FormSubmission.lead_id	200 raw list of dicts	not paginated; not a Page[T]; joins models.forms.FormSubmission/FormTemplate
GET /leads-with-submissions (208 to 284)	ActiveUser	WorkspaceID, Lead inner-joined to FormSubmission	200 {items, total, limit, offset, has_more}	the CRM inbox search: optional status (a LeadStatus value, silently ignored if invalid, 227 to 232) and search (ilike over email/first_name/last_name/company_name, 234 to 241) query params, limit/offset same bounds as /leads; submission counts are batched in one GROUP BY per page (252 to 261), not one query per lead; the envelope is hand-built as a dict, not the Page[T] Pydantic model the other three paginated routes use, so it is structurally identical but not the same type

Application logic

CRUD is uniform across Lead, Deal and ClientTag, each following the same four-step shape in CrmService (crm_service.py):

1. Lookup helpers scope by id and workspace together, never id alone: get_lead (28 to 36) filters Lead.id == lead_id, Lead.workspace_id == workspace_id and raises 404 if nothing matches; get_deal (81 to 90) does the same by joining Deal to Lead and filtering the join on Lead.workspace_id, because Deal itself carries no workspace_id column. update_lead/update_deal/update_tag/delete_tag all call the matching lookup first, so a write to a foreign-workspace id 404s before any setattr happens.
2. Create checks for a pre-existing row only on Lead: create_lead (38 to 52) runs its own SELECT for (email, workspace_id) before inserting and answers a clean 409 on a match; create_deal (92 to 101) and create_tag (138 to 143) insert directly, relying on get_lead's 404 (for deals, to confirm the lead exists in-workspace) but nothing to guard the DB's own unique constraints, so a colliding POST surfaces as an unhandled IntegrityError rather than a 4xx (Gotchas).
3. Update is exclude_unset=True, applied field by field: update_lead (54 to 62), update_deal (103 to 111) and update_tag (145 to 153) all do for key, value in update_data.dict(exclude_unset=True).items(): setattr(obj, key, value) then commit()/refresh(); a field the client omits is left untouched, a field explicitly sent as null is set to null.
4. Delete is a plain db.delete() + commit(): delete_lead (64 to 68) relies on the ORM relationship cascades declared on Lead (Data model) to remove the deal/notes/activity logs; delete_tag (155 to 160) does not check for existing lead_tags rows first.

Router-level logic beyond straight service delegation, both in crm_router.py: get_leads/get_deals/get_leads_with_form_submissions build the Page[T] (or, for the last one, an equivalent hand-built dict) by running a COUNT query then a LIMIT/OFFSET query against the same filtered base query (57 to 64, 112 to 123, 223 to 249), so the count and the page can never disagree on the filter; get_leads_with_form_submissions additionally batches per-lead submission counts in one GROUP BY FormSubmission.lead_id IN (...) (252 to 261) instead of querying per lead.

Invariants

• Every lookup, update and delete is scoped by (id, workspace_id) together; none of the eleven id-taking handlers in crm_router.py resolves a row by id alone (verified by reading every get_lead/get_deal/tag-filter call site in crm_service.py).
• Deal, ClientNote and ClientActivityLog carry no workspace_id column of their own; their tenancy is always derived by joining through Lead (get_deal, 81 to 90) or by calling get_lead first (add_note_to_lead, 114 to 127; get_notes_for_lead, 129 to 132).
• A page's total and its items come from one filtered base query split into a COUNT and a LIMIT/OFFSET, never two independently-filtered queries (get_leads, get_deals, get_leads_with_form_submissions), which is what test_crm_leads_first_page_matches_unbounded and test_crm_leads_query_count_constant hold constant.
• has_more is computed as (offset + len(items)) < total everywhere it appears (three call sites), never offset + limit < total, so the last, possibly-short page correctly reports has_more = False (test_crm_leads_paginated_caps_at_100, the exactly-100-of-100 case).
• Only Lead creation checks for a pre-existing row before inserting; Deal and ClientTag creation rely entirely on the database's unique constraints, which are not caught anywhere in crm_service.py (no except IntegrityError), so a collision on either becomes an unhandled 500, not a 409 (measured by reading create_deal/create_tag, no test found exercising this path).

Frontend

No module in alchemy_labs's own frontend calls any route in crm_router.py at this commit (grep for /api/crm/leads, /api/crm/deals, /api/crm/tags or leads-with-submissions under frontend/src returns nothing outside this router's own file). The nearest real, measured connection is the cold-outreach campaign system (AdminOutreach.tsx, AdminOutreachCampaign.tsx, backed by outreachService.ts's BASE = '/api/crm/outreach', a sibling router): it reads and writes rows in the same leads table this recipe's CrmService manages (outreach_router.py 17, 105, from models.crm_model import Lead), just through its own endpoints rather than GET /leads or GET /leads-with-submissions. That campaign workspace is the frontend surface Prima's rebuild targets:

prospecting/ui/outreach-campaign-workspace   renders and edits rows from the same `leads` table this
                                                                recipe's CRUD manages (via a sibling router at this
                                                                commit, not via /api/crm/leads itself); the CRM-inbox
                                                                search (GET /leads-with-submissions) this recipe
                                                                exposes has no confirmed frontend consumer in
                                                                alchemy_labs at 62ba4fc4

Minimum admin/debug surface: none dedicated exists at this commit. A human check is a direct query: SELECT id, email, status, source, workspace_id FROM leads WHERE workspace_id = :ws ORDER BY created_at DESC LIMIT 50 (the same shape GET /leads runs); SELECT l.email, d.stage, d.deal_value FROM deals d JOIN leads l ON l.id = d.lead_id WHERE l.workspace_id = :ws for the pipeline view GET /deals returns.

Security & tenancy

Auth dependency: ActiveUser (Annotated[User, Depends(get_current_active_user)], dependencies.py 566), which accepts both a JWT and an API key (see the 2026-04-17 fix in Gotchas); no surface_exempt call on this router, so it is a normal authenticated Labs surface, not a public webhook. Tenancy: WorkspaceID (Annotated[UUID, Depends(_get_workspace_id)], dependencies.py 680 to 779) resolves the workspace from, in order, request.state.workspace_id (set by WorkspaceMiddleware for JWT requests, already access-checked), the X-Workspace-ID header (re-validated against the caller for API-key auth, 403 "You don't have access to this workspace" on a mismatch), a single-accessible-workspace fallback, or a 400/409 asking the client to disambiguate or bootstrap a workspace; this recipe never resolves or trusts a workspace id from its own request body or path.

Fail direction: every id-scoped lookup in CrmService closes on a mismatch (Lead/Deal/ClientTag not found for this workspace $\Rightarrow$ 404, same status as genuinely nonexistent, so a foreign-workspace id cannot be distinguished from a nonexistent one by response shape); crm_router.py uses DatabaseSession (plain get_db), not DatabaseSessionWithRLS, so there is no Postgres Row-Level Security backstop under this router's application-level filters, unlike the defense-in-depth pattern _shared/workspace-scoping documents for RLS-protected routes. Secrets at rest: none of the six tables stores a credential, a token or a card detail; client_activity_logs.metadata and leads.tracking_data are free-text/JSON columns that could carry arbitrary caller-supplied strings, nothing in this recipe redacts them. Rate limits: no per-route limiter decorator found on any handler in crm_router.py; the router relies on whatever global rate limiting app_fastapi.py applies to all authenticated LABS_SURFACES routes, not documented here. Attacker-controlled inputs: every path/body field on every route is workspace-unchecked user input by construction (ids, email, search, status) except workspace_id itself, which never comes from the request; search is passed into .ilike(f"%{search}%") as a bound SQLAlchemy parameter (234 to 241), not string-interpolated SQL.

Tests

The only test file measured for this recipe, tests/v2/test_s5_pagination.py (250 lines, module docstring "V2 - S5 - Unbounded lists + N+1"), exercises the router functions and CrmService directly against the transactional db_session fixture from tests/conftest.py, using a _QueryCounter (86 to 102) that listens on before_cursor_execute to assert flat query counts. Run 2026-09-14 against deap-postgres-test (Postgres 14, localhost:5434):

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

prints 4 passed. The four: test_crm_leads_first_page_matches_unbounded (Gate A: the paginated first page, limit=50 offset=0, is byte-identical in id order to the old unbounded endpoint's first 50 of 120 leads); test_crm_leads_paginated_caps_at_100 (Gate B: a 100-lead workspace's limit=100 page returns all 100 with has_more=False, and an offset=100 second page is empty with has_more still False); test_crm_leads_query_count_constant (the query count for GET /leads at 10 leads equals the count at 100 leads, proving the page and the count together stay O(1)); test_backfill_uses_constant_queries (a TrackingService N+1 fix adjacent to but not inside this recipe's own files, included in the same test file, cited for completeness only).

Hermetic: the whole file runs against the real transactional test database, nothing is mocked; no live external service is involved.

What a rebuild adds and this file does not cover: no test in the repository calls create_lead, update_lead, delete_lead, create_deal, update_deal, add_note_to_lead, list_tags, create_tag, update_tag or delete_tag (grep -rn "crm_service\.\(create_lead\|update_lead\|delete_lead\|create_deal\|update_deal\|add_note_to_lead\|create_tag\|update_tag\|delete_tag\)" across backend/tests/ returns nothing); every CRUD write path in this recipe beyond GET /leads is exercised only by hand, not by the suite; test_leads_isolated/test_client_tags_isolated in backend/tests/integration/test_workspace_isolation.py (107 to 125) cover cross-workspace isolation at the model level (inserting two workspaces' rows and asserting a filtered query only returns one), not through crm_router.py's handlers; no test exercises the case-sensitivity gap in create_lead's duplicate check or the unhandled IntegrityError paths on create_deal/create_tag/delete_tag (Gotchas).

Deploy & ops

No dedicated config key: this router reads no env var of its own, only the ambient DATABASE_URL/SYNC_DATABASE_URL every FastAPI route shares. Schema: the six tables were captured into Alembic tracking by baseline_prod_schema_001 (down_revision = None, Create Date: 2026-08-02, "Reconciliation migration ... the rest (including users and workspaces) were created solely by SQLAlchemy's create_all() at app boot"); their FKs were added afterward, once every referenced table was guaranteed to exist, by deferred_fk_phantom_tables_001; leads.workspace_id itself predates the baseline capture and was added by leads_ws_001 (2026-04-08), which also converted the lead-email uniqueness from a bare column constraint to (workspace_id, lower(email)); leadsource_booking_001 (2026-09-07) is the most recent schema touch, adding 'booking' to the live leadsource Postgres enum for the calendar booking bridge. No scheduler job and no background worker reads or writes any of these six tables in this recipe (nothing found calling CrmService or querying leads/deals/client_tags outside crm_router.py, outreach_router.py, and the booking bridge). No Argus-style health check or monitoring probe dedicated to this router was found; rollback is ordinary code rollback, no migration in this recipe's chain is destructive to run backward except leadsource_booking_001's downgrade, which is a documented no-op (Postgres cannot drop an enum value).

Gotchas

• 2026-04-17 (eea28a1a4b6cb2e07eccba4ea958f148681c18e5, "Fix CRM router: remove double prefix, use ActiveUser auth"): the router previously declared its own prefix="/crm" on top of the app-level prefix="/api/crm", so every route 404'd at /api/crm/crm/...; the same commit switched auth from AdminUser (JWT-only) to ActiveUser (works with API keys too), for consistency with the other business routers. A rebuild that adds a router-level prefix on top of an app-level one reproduces exactly this outage.
• 2026-05-28 (37c7157c767f230a92ae1c3903fef16f78da5634, "fix: enforce workspace isolation across all routers + demo seed"): before this commit, GET/POST/PUT/DELETE /tags had no workspace_id parameter at all; crm_service.list_tags/create_tag/update_tag/delete_tag took no workspace argument and any authenticated user could read or modify any workspace's tags. The fix threaded WorkspaceID through all four tag handlers and both matching service methods; the CRM tags surface is explicitly named in that commit's own list ("CRM tags (4)") alongside forms, outreach, design, tracking and SEO routers that had the same gap.
• 2026-06-10 (b71e1ba3b6ae8262900050025235eb9a63d019c4, "Wave-2 hardening ... S5 crm pagination + N+1 batch"): GET /leads and GET /deals were unbounded before this commit (the whole table, every call); GET /leads-with-submissions ran one SELECT count(*) per lead to compute submission_count (an N+1). Both were fixed in the same commit that introduced Page[T], DEFAULT_PAGE_SIZE/MAX_PAGE_SIZE, and the batched GROUP BY count (crm_router.py 252 to 261) this recipe documents.
• 2026-09-14 (measured this session, not dated in the source): create_lead's pre-insert duplicate check is Lead.email == lead_data.email (case-sensitive, crm_service.py 42), but the DB's own uniqueness guard is uq_leads_workspace_email on (workspace_id, lower(email)) (leads_ws_001, 2026-04-08). A POST /leads with Jane@Example.com in a workspace that already has jane@example.com passes the service's check, then fails at db.commit() with an unhandled IntegrityError, surfacing as a generic 500 instead of the clean 409 same-case duplicates get.
• 2026-09-14 (measured this session): create_deal and create_tag have no pre-insert duplicate check at all (unlike create_lead); a second POST /deals for a lead_id that already has one, or a second POST /tags with a name already used in the workspace, both reach db.commit() and raise an unhandled IntegrityError on uq_deals_lead_id / uq_client_tags_workspace_name rather than a 409. delete_tag has the mirror problem on the way out: lead_tags.tag_id carries no ON DELETE clause, so deleting a tag still attached to any lead raises an unhandled foreign-key violation instead of a clean 409 "tag in use".

Reference implementation

Paths relative to the repo root; all at main 62ba4fc41009b454e1c033e5036c7e8577db2061 unless noted.

• alchemy_labs/backend/services/crm_service.py (162 lines, whole file in scope): list_leads/get_lead (19 to 36); create_lead (38 to 52); update_lead (54 to 62); delete_lead (64 to 68); list_deals/get_deal (71 to 90); create_deal (92 to 101); update_deal (103 to 111); add_note_to_lead/get_notes_for_lead (114 to 132); list_tags/create_tag/update_tag/delete_tag (135 to 160).
• alchemy_labs/backend/api/routers/crm_router.py (284 lines, whole file in scope): Page[T] (30 to 36); lead routes (41 to 92); deal routes (96 to 145); note routes (149 to 157); tag routes (161 to 176); get_lead_submissions (181 to 205); get_leads_with_form_submissions (208 to 284).
• alchemy_labs/backend/models/crm_model.py (198 lines, whole file in scope): the five enums (16 to 66); Lead (70 to 112); Deal (114 to 139); ClientTag (141 to 159); LeadTag (161 to 166); ClientNote (168 to 181); ClientActivityLog (183 to 197).
• alchemy_labs/backend/schemas/crm_schemas.py (110 lines, whole file in scope): tag schemas (12 to 28); note schemas (32 to 51); lead schemas (55 to 87); deal schemas (91 to 110).
• alchemy_labs/backend/api/dependencies.py: get_db (27 to 33, no RLS); get_current_active_user (338 onward); ActiveUser (566); _get_workspace_id (680 to 779); WorkspaceID (779).
• alchemy_labs/backend/app_fastapi.py: router mount (877, prefix="/api/crm", tags=["CRM"], surfaces=LABS_SURFACES); the sibling outreach_router mount at the same prefix (878).
• alchemy_labs/backend/api/routers/outreach_router.py: from models.crm_model import Lead (17); the same-table lookup (105) cited in Frontend.
• alchemy_labs/backend/migrations/versions/baseline_prod_schema_001.py: module docstring (1 to 24); the six TABLE: blocks cited in Data model.
• alchemy_labs/backend/migrations/versions/deferred_fk_phantom_tables_001.py: the FK and unique-constraint additions for these six tables cited in Data model.
• alchemy_labs/backend/migrations/versions/20260408_add_workspace_id_to_leads.py (leads_ws_001): whole file in scope, the workspace_id backfill and the case-insensitive unique index.
• alchemy_labs/backend/migrations/versions/20260907_0001_leadsource_add_booking.py (leadsource_booking_001): whole file in scope.
• alchemy_labs/backend/tests/v2/test_s5_pagination.py (250 lines, whole file in scope, see Tests).
• alchemy_labs/backend/tests/integration/test_workspace_isolation.py: test_leads_isolated (107 to 115), test_client_tags_isolated (118 to 125), cited in Tests for completeness only.

Deviations of the reference from an ideal implementation, dated 2026-09-14 unless noted: no IntegrityError handling anywhere in CrmService beyond create_lead's own proactive check, so three write paths (duplicate deal, duplicate tag, delete-tag-in-use) surface as unhandled 500s; the duplicate-lead check is case-sensitive while the DB constraint it guards is not; GET /leads/{lead_id}/notes, GET /tags and GET /leads/{lead_id}/submissions remain unbounded lists, never converted to Page[T] the way /leads//deals//leads-with-submissions were in the 2026-06-10 pagination pass; ClientActivityLog is a fully modeled table with five CRM-relevant enum members nothing in this recipe ever writes to; crm_router.py uses plain get_db, forgoing the RLS defense-in-depth DatabaseSessionWithRLS offers elsewhere in the app.

Acceptance checks

Run from alchemy_labs/backend at 62ba4fc4 with the deap-postgres-test container (Postgres 14, port 5434) up.

1. TEST_DATABASE_URL=postgresql://test_user:test_password@localhost:5434/test_db python -m pytest tests/v2/test_s5_pagination.py -q --no-cov -o addopts="" -p no:cacheprovider prints 4 passed (measured 2026-09-14).
2. grep -c "^def test_" tests/v2/test_s5_pagination.py prints 4.
3. grep -n "prefix=\"/api/crm\"" app_fastapi.py shows the mount at line 877, with no prefix= argument inside APIRouter() in api/routers/crm_router.py (grep -n "APIRouter(" api/routers/crm_router.py prints line 21 with no arguments), confirming the 2026-04-17 double-prefix bug stays fixed.
4. grep -n "workspace_id: WorkspaceID" api/routers/crm_router.py | grep -c tag prints 4 (get_tags, create_tag, update_tag, delete_tag all still take the parameter the 2026-05-28 fix added).
5. grep -n "class Page" api/routers/crm_router.py shows line 30; grep -n "response_model=Page\[" api/routers/crm_router.py prints exactly two lines (get_leads, get_deals); GET /leads-with-submissions has no response_model=Page[...] (it returns a hand-built dict, confirmed by grep -n "def get_leads_with_form_submissions" -A2 api/routers/crm_router.py showing no response_model on the decorator).
6. grep -n "except.*IntegrityError" services/crm_service.py prints nothing (0 matches), confirming the unhandled-IntegrityError gap in Gotchas.