CSV batch enrichment

Use when the user wants to upload a lead CSV or JSON file, enrich many leads overnight, or resume a paused batch after topping up.

Endpoints

Web UI: /batch and the paperclip on /enrich_agent. Results: /batch/{id}/results. Each summary card links to that row's Enrich Agent chat at /enrich_agent?session=<demo_session_id> when a per-row session exists.

CSV format

Any delimiter and header shape — comma, tab, or semicolon; utility exports, CRM dumps, vendor spreadsheets, etc. JSON uploads use the same three-step path (object keys treated as column headers). The upload path:

1. CRM fast path — obvious headers (company, street, phone, …) and common utility export names (customer_name, service_address, service_city, …) map without an LLM call.
2. Per-agent header cache — when the same header set was mapped before for this agent (fingerprinted by normalized column names), reuse the stored map and skip Crimson Span.
3. Crimson Span mapping — when headers are non-standard and uncached, one data411-crimson-span call reads the header row plus up to five sample rows and returns a column map to canonical lead fields plus optional search_variants (billing names, full mailing/billing addresses, secondary phones — the LLM composes multi-column addresses via compose_columns; ingress code does not split line-2/zip).
4. Thin-result hunts — variant columns are stored on each lead as search_variants and injected into supplement passes only when core coverage is thin (first-pass hunts found little from the primary company/address).

Canonical lead fields (same as enrich/lead-session):

contact_id is optional — when missing, a stable positive int is synthesized from company + street + city + state + zip for LEB storage only (per batch-row session). Non-numeric external ids (docket numbers, alphanumeric CRM keys) are stored verbatim in input_snapshot and hashed to a BIGINT for lead_enrichment_backfill; the original value is kept in lead_ref.external_contact_id.

Prior-enrichment reuse (warm-seed / transcript hydrate) matches the same lead in order:

1. chat_title — same sidebar title on an older agent chat (demo_chat_session.title)
2. Exact input record — canonical CRM JSON matches a prior LEB input_snapshot
3. Batch input record — same canonical JSON on a prior matched batch row (loads LEB or demo_session_debug graph; AGENT_LEAD_BATCH_CACHE_ENABLED=1 default)
4. CRM contact_id — user-supplied int or external string (hashed for LEB lookup)
5. Phone, business name + address, or person name + address

Synthetic per-row contact_id values are for LEB storage only — not used for cross-batch matching. When company and address are pasted into one CSV cell (COMPANY … STREET … CITY, ST, ZIP), the mapper splits them into company, street, city, state, zip before enrichment. At least company or enough address fields must be present.

Comma-separated related names in one company column (common on utility exports: GTI Pennsylvania LLC,Vision Management Services) are split at ingress into one operational primary company plus sibling names in search_variants. The verbatim CRM cell is preserved as company_name_stack_raw on the LEB snapshot. When a single legal entity name contains an internal comma (Smith & Sons, LLC), quote the field per RFC 4180 so it is not split.

Quoted CSV (RFC 4180) is best when company names or addresses contain commas — wrap those fields in double quotes. The upload UI checks the first 10 lines before sending.

Example (standard CRM):

contact_id,company,street,city,state,zip,number1
81555753,Haney Furniture Co,100 Main St,Pittsburgh,PA,15213,7246547732
99201,"Smith & Sons, LLC","100 Main St, Ste 2",Tampa,FL,33602,8135550100

Env: BATCH_ENRICH_LLM_MAP_ENABLED (default 1), BATCH_ENRICH_LLM_MAP_MODEL (default data411-crimson-span), OPENROUTER_API_KEY required for non-standard CSV or JSON shapes (utility CRM exports).

JSON format

Upload a .json file (or JSON content with a .json filename) containing lead objects. Supported shapes:

• Array: [{ "company": "…", "city": "…", … }, …]
• Wrapper: { "leads": [ … ] } or { "rows": [ … ] }
• Single lead object (one-row batch)

Same column-mapping pipeline as CSV: CRM fast path → per-agent header cache → Crimson Span on object keys plus sample rows. Canonical-field keys still map via the fast path; utility export field names (customer_name, service_address, …) go through Crimson Span. Optional per-object search_variants arrays pass through when present. Response includes input_format: "json" and map_source (static, cached, or llm). Enriched download is still CSV (flattened LEB graph).

Example:

{
  "leads": [
    {
      "contact_id": 81555753,
      "company": "Haney Furniture Co",
      "street": "100 Main St",
      "city": "Pittsburgh",
      "state": "PA",
      "zip": "15213",
      "number1": "7246547732"
    }
  ]
}

Batch instructions (batch_prompt)

Optional text applied to every row in the job (same as enrich/lead-session prompts). Set via:

• Upload form field batch_prompt (max 4000 chars)
• POST …/start or /resume body { "batch_prompt": "Find owner mobile numbers" } before the job enters queued

Editable while status is uploaded, pending_funding, or paused_balance, and also when resuming or processing more rows (completed, queued, running with rows still in the file). Clear batch_prompt to run remaining rows with standard enrichment only — already-processed rows keep their prior follow-up in the row chat.

Each row runs a standard initial enrich/lead-session (including synthesis_merge). When batch_prompt is set, the worker then appends that text as a second user message on the row chat and runs a follow-up hunt (prior_backfill + prompts) on the default first-pass parallel websearch stack (WEBSEARCH_PARALLEL_MODELS / Hunt Ground, Hunt Native, Hunt Sonar, …) — not the dig-deeper slots (Hunt Forge / Swift / Deep). Then synthesis_merge runs again on the merged graph. Instructions are not injected into the initial hunt. Web UI: upload dialog textarea and /batch/{id} status page before Start.

Max rows and file size: BATCH_ENRICH_MAX_ROWS (default 10 000) and BATCH_ENRICH_MAX_UPLOAD_BYTES (default 5 MB).

Concurrency: BATCH_ENRICH_MAX_JOBS (default 5) caps how many CSV batch files hold a worker lease at once globally. Row parallelism uses two gates enforced in Claim_NextBatchEnrichRow:

Legacy BATCH_ENRICH_ROW_CONCURRENCY is an alias for the soft limit when BATCH_ENRICH_ROW_SOFT_LIMIT is unset. Long hunts (>3 min) let a single batch pipeline beyond five concurrent rows until the global hard cap is reached.

Optional upload form field row_limit (process_limit on the job) caps how many rows the worker enriches. The full CSV is parsed and every mapped row is stored in batch_enrich_row (status skipped beyond the limit; web UI Rows to process, default 10). Response: stored_row_count, row_count (pending), file_row_count. Enrich-agent uploads auto-call /start when balance allows. Crimson Span column mapping receives up to min(row_limit, 10) sample rows (BATCH_ENRICH_LLM_PREVIEW_ROWS, default 10) and returns chat_label (primary_column + optional location_columns) so sidebar titles and each row's chat_title are set at upload — enrichment does not re-derive names per record.

How processing works

1. Upload creates a batch_enrich_job and one batch_enrich_row per lead.
2. When a bearer is present, demo sessions are created: one batch_file parent and one batch_row chat per lead (seeded with the CRM JSON). While the batch worker runs, each row chat mirrors live enrich_lead_session hunt progress (tool progress + parallel websearch tails) — open Files in /enrich_agent and click any lead to watch. Row chats also hydrate from prior agent enrichment when chat_title, exact input record, or company + address matches an earlier chat for this agent — not by per-row synthetic contact_id.
3. Start checks virtual balance ≥ BATCH_ENRICH_MIN_BALANCE_USD (default $10).
4. Worker claims rows and runs the same path as enrich/lead-session + backfill persist; pauses when balance cannot cover the next row. Matched rows with a persist failure still show status: matched but backfill_id: null — check stats_json.persist on the row. Ops: backfill from debug with python3 scripts/backfill_leb_from_session_debug.py --all-matched --agent-id N --link-batch-rows --commit.
5. Bill-on-match — empty rows auto-credit like single-lead hunts.
6. Preview and results pages show counts only (people, phones, emails) — never contact names or numbers in the UI cards. The authenticated CSV download is a flattened LEB graph (backfill_json): one row per lead with dotted/indexed columns (contacts.0.display_name, phones.0.e164, emails.0.address, entities.0.legal_name, confidence, summary_markdown, …). The internal tool_trace is omitted. Columns are the union across all rows, prefixed with row_index, enrich_status, enrich_error_code, enrich_lead_type, enrich_matched, enrich_spent_usd.

Commercial vs residential prefilter (C vs R)

Optional lead-type prefilter on upload (prefilter_lead_type: commercial or residential). Flow:

1. In-scope rows start as preprep (name-only classify via data411-hunt-economy, batch-50, global fingerprint cache).
2. Prep worker sets each row to pending (passes filter) or skipped (error_code=prefilter_lead_type) when classified type ≠ selected filter. unknown is never skipped by prefilter.
3. Enrichment worker claims pending only.

Match rate = matched ÷ enrichment attempts (matched + no_match + failed); prefilter skips are excluded from the denominator and from rows_remaining.

Classification billing: global cache hit → $0; cache miss → COGS × LEAD_SHAPE_CLASSIFY_MARKUP (default 1.15) debited from virtual balance. Upload response includes estimated_classify_usd and classify_uncached_count.

Manual override: POST /api/v1/agents/lead-type with { "primary_name", "lead_type" } updates the global cache (source=manual). Does not re-apply batch prefilter — call POST …/re-prep with optional new prefilter_lead_type after manual changes.

Env: LEAD_SHAPE_CLASSIFY_ENABLED (default 1), LEAD_SHAPE_CLASSIFY_MODEL (default data411-hunt-economy), LEAD_SHAPE_BATCH_SIZE (default 50), LEAD_SHAPE_CLASSIFY_MARKUP (default 1.15).

Upload form fields: prefilter_lead_type, row_limit, batch_prompt.

Billing order

Same as all paid routes: virtual USD balance first; wallet x402 only when balance is exhausted. Top up at /connect?action=topup.

Anonymous flow

1. Upload without bearer → upload_token in response.
2. Store token in sessionStorage (SPA does this automatically).
3. Mint / connect with return=/batch/{id}.
4. POST …/link with { "upload_token": "…" }.
5. Fund to $10+ → POST …/start.

Headless parallel REST

For integrations that already have row iterators (no CSV file), see batch_and_concurrency and event_callbacks.