Salesforce ingestion for Call Context Service

This Dealops 2 server PR wires Salesforce call notes and opportunity context into the AI quoting Call Context Service instead of relying on hand-seeded rows.

Author: @pk675 PR: dealops#6784 Ticket: DEA-7334 Area: Dealops 2 / AI quoting Files: 18 Diff: +2841 / -37 State: open

What it adds

A Salesforce-backed ingestion path for Dealops_Call_Note__c and Dealops_Opp_Context__c.

Rows land in CallRecord, OppContext, and CallSyncState, where the agent already reads evidence.

What changes

CallContextService.sync() is no longer a stub. It pulls one deal on demand through the same ingestion engine used by the scheduled sweep.

Operational shape

A 6-hour cron fans out one BullMQ job per enabled org. Enablement is data-driven: an org is in scope only when it has a CallContextOrgSyncState row.

Review focus

Cursor semantics, provenance-preserving context merges, migration safety, and the Salesforce SOQL contract are the important edges.

Org sync state

Salesforce source

Ingestion engine

Scheduled jobs

Call Context tables

Service API

1. Why this exists

Before

sync() returned an honest error: no live CRM pull wired up yet.

Real orgs had empty CallRecord and OppContext rows.
The agent could not quote from calls unless someone manually seeded evidence.
evidence() stayed gated on missing or stale call sync state.

After

Salesforce becomes the first live CRM source for Call Context ingestion.

Call notes and opp context are pulled from Dealops-owned Salesforce custom objects.
Both scheduled backfill and on-demand deal refresh share the same dedupe and merge rules.
Touched opportunities are marked fresh, making evidence quotable.

Design stance: this is opt-in per org. No CallContextOrgSyncState row means the background job does not touch the org, which keeps orgs without the Dealops Salesforce objects out of this path.

2. What changes

Schema

Adds CallContextOrgSyncState as the org-level enablement gate and cursor store.

Splits call provenance into crmType and tool-level source.

CRM source

Adds salesforceCallContextSource.ts, the only Salesforce-aware ingestion file.

It reads fixed Dealops object and field API names by SOQL.

Ingestion

Adds CRM-agnostic planners and writers in ingestCallContext.ts.

Dedupe uses organizationId + crmType + crmRecordId; unchanged calls skip writes via content hash.

Jobs

Adds syncCallContext and syncCallContextForOrg.

The cron dispatches; per-org jobs own retry and failure isolation.

File map

apps/server/src/dealops2/aiQuoting/callContext/

├─ callContextService.ts sync() + syncOrg() now call ingestion

├─ types.ts crmType/tool source split + org sync result

└─ ingestion/

├─ dealopsObjects.ts fixed Salesforce object contract

├─ salesforceCallContextSource.ts SOQL adapter

├─ ingestCallContext.ts CRM-agnostic merge/write engine

├─ types.ts source boundary

└─ __tests__/ingestCallContext.test.ts real DB e2e ingestion suite

apps/server/src/jobs/

├─ definitions/syncCallContext.ts

├─ definitions/syncCallContextForOrg.ts

├─ registry.ts

└─ startWorkers.ts

packages/prisma/

├─ migrations/20260824200000_call_context_ingestion/migration.sql

└─ schema/models/v2/call-context.prisma

Schema delta

Entity	Change	Reason
CallContextOrgSyncState	New table keyed by `organizationId`; stores `callNoteCursorAt`, `oppContextCursorAt`, `status`, and `lastError`.	One row both enables an org and remembers independent cursors per object.
CallRecord	Replaces old `source` CRM enum with `crmType`; adds nullable `source` as call-tool enum.	Salesforce is the CRM, while Gong/Fireflies/Chorus/manual is the note source. Those are different dimensions.
CallRecord unique key	`organizationId_source_crmRecordId` → `organizationId_crmType_crmRecordId`	Dedupe should not depend on call-tool label; the same CRM record should remain one row if Source changes.
OppContext	Adds `includeInGoldenSet Boolean @default(false)`.	Makes the “teaching example” flag queryable and preserves `false` as a real value.

Diff weight

+2841 additions are mostly new ingestion code, tests, jobs, and migration; -37 deletions are primarily old stub/provenance assumptions.

3. How it works

Runtime flow

1. Enable org

Create one CallContextOrgSyncState row during onboarding.

2. Cron fan-out

syncCallContext runs every 6 hours and queues one org job per enabled row.

3. Salesforce pull

Source queries Dealops_Call_Note__c and Dealops_Opp_Context__c.

4. Ingest

Resolve Opportunity links, dedupe, hash-skip unchanged calls, merge context.

5. Quote gate

Touched deals get fresh CallSyncState; evidence() can return quotable context.

Two doors, one engine

Scheduled org sweep

CallContextService.syncOrg(orgId)

Used by syncCallContextForOrg.
Fetch scope: modifiedSince.
Advances org cursors only for successful halves.
6-hour cadence: bulk backfill and safety net, not freshness.

On-demand deal refresh

CallContextService.sync(deal)

Fetch scope: one Opportunity__c CRM id.
Ignores org cursors because reading one deal says nothing about the rest of the org.
Skips CRM calls when mode: ifStale and the deal is already fresh.
mode: force always pulls.

Key ingestion rules

Rule	Implementation	Why it matters
Org opt-in	Background jobs read enabled orgs from `CallContextOrgSyncState`.	No allowlist drift; orgs without Dealops Salesforce objects are never polled.
Independent cursors	`callNoteCursorAt` and `oppContextCursorAt` advance separately.	A call-note failure cannot skip past opp-context changes, or vice versa.
Cursor overlap	`cursorAfterPoll()` rewinds poll start by 15 minutes.	Records modified during a poll or under clock skew get re-read instead of skipped forever.
Call dedupe	Existing records load by `crmType + crmRecordId`; content hash decides unchanged vs update.	Re-reading overlap windows is cheap and safe.
Context provenance	`customer` and `extracted` outrank `synced`.	A CRM sync never overwrites a human-entered or extracted value.
Partial records	Missing required call-note identity/body fields are warnings and skips, not poll failures.	One bad CRM row does not block the org.

Salesforce contract

Fixed object names

The API names are centralized in dealopsObjects.ts because they are not per-org mapped fields.

Dealops_Call_Note__c Dealops_Opp_Context__c Opportunity__c LastModifiedDate

The PR intentionally does not use crmObjectService.queryObjects(), because that path only emits equality filters and this needs LastModifiedDate > cursor.

Service API changes

Old contract assumption

CallRecord.source:
  'salesforce' | 'hubspot' | 'manual'

sync():
  'error' unless already fresh

New contract

CallRecord.crmType:
  'salesforce' | 'hubspot'

CallRecord.source:
  'gong' | 'fireflies' | 'chorus' | 'manual' | 'other' | null

sync():
  pulls one deal via CRM source

syncOrg():
  sweeps enabled org via cursors

4. Tests and coverage

Main suite

ingestCallContext.test.ts adds database-backed coverage around the real ingestion engine.

Only the CRM network boundary is faked.

Fixtures

Each case creates its own org, user, and OpportunityV2, then tears them down.

No customer org is required.

Service readback

Assertions read through the real CallContextService, including evidence().

Covered behavior	Why reviewers should care
Happy path calls + context land end-to-end	Confirms the agent reads what ingestion wrote.
Unchanged records are no-ops; edits update in place	Validates content hash and CRM-record dedupe.
Customer provenance survives sync	Prevents CRM backfill from clobbering higher-trust values.
Empty CRM fields do not clear stored values	Matches partial-object semantics.
Unknown opportunities are skipped and counted	Normal CRM noise does not fail a poll.
Failed poll marks org broken with cursors intact	Prevents silent data loss across failures.
On-demand `sync()` and org `syncOrg()` share dedupe	Makes two entry points safe to run against the same CRM record.
Large batch of 50 call notes	Exercises bulk ingest behavior without pinning query counts.

5. What it doesn't change

Explicit non-goals

Does not add a new tRPC route; this changes server service behavior and background jobs.
Does not make the 6-hour job a freshness mechanism for a rep actively opening a deal; on-demand sync() is the freshness door.
Does not onboard every org automatically; orgs need an explicit CallContextOrgSyncState row.
Does not make call body cleanup or summarization live; bodyType remains served as raw.
Does not use per-org crmObjectSpecs mappings for these Salesforce objects; the object contract is fixed.
Does not add HubSpot ingestion, though the source interface leaves room for it.
Does not hard-delete call context data; the existing soft-delete posture remains.

6. Risks / rollback / open questions

Before merge blocker: the PR description says the migration still needs to be applied via Prisma workflow, and tests fail until the new schema exists. Confirm the generated migration is the one intended to land and that local/CI DB setup runs it.

Migration risk

The migration drops the old CallRecord.source enum column and re-adds source as call-tool provenance.

Review whether existing rows need data backfill beyond defaulting crmType to salesforce.

Salesforce query contract

The source type says implementations must return oldest-modified first, while the SOQL implementation orders by Id.

Verify this is harmless with the poll-start cursor model, or align the query/comment before merge.

Failure behavior

Partial object failures mark the org broken, preserve failed cursors, and still keep the successful half.

The per-org BullMQ job then throws so retries and queue failure visibility still work.

Rollback lever

Soft-delete or remove an org's CallContextOrgSyncState row to stop background ingestion for that org.

Unschedule/disable syncCallContext to stop all fan-out while leaving on-demand service code in place.

Before enabling the first org

Run and verify the Prisma migration against the target environment.
Insert the first CallContextOrgSyncState row deliberately, likely for Merge as called out in the PR.
Confirm the org has both Salesforce custom objects and the exact fixed field API names.
Watch lastError, job failures, and warning telemetry for missing objects, bad JSON, unknown opportunities, and unrecognized call-tool labels.