Product2 lookup for the Product Catalog Agent

A local-first CrmProduct search with a queued Salesforce/Ampersand fallback, wired into the config-agent chat through a new SSE event and client-side job polling.

1. Why this exists

The Product Catalog Agent stages catalog products with a crmProductId that should map to the Salesforce Product2.Id for writeback. Before this PR, there was no way for the agent to actually look up a Product2: it would either skip CRM mapping entirely or scan up to 5,000 CrmProduct rows unfiltered, with no path to the live Salesforce object.

2. End-to-end flow

The agent always tries local first. Local hits resolve immediately. Local misses (with Salesforce configured) queue a job, end the SSE turn, and let the client take over polling. Step through the lifecycle:

3. The new server-side pieces

defineJob({
  name: 'fetchSalesforceProduct2Candidates',
  schema, // org + lookup fields
  defaultJobOptions: {
    attempts: 2,
    backoff: {
      type: 'exponential',
      delay: 5000,
    },
  },
  handler: async (job) =>
    fetchSalesforceProduct2Candidates(job.data),
});

4. The streaming protocol change

The orchestrator inspects each askProductCatalogAgent result for a salesforceLookupQueued marker. If it finds one, it does not continue iterating tools — it emits a structured event and ends the turn.

// Inline Salesforce call here =
//   stream hangs for ~5–30s,
//   or the model's turn drifts.

// types.ts — new SSE variant
| {
  type: 'product2_lookup_job';
  jobName: 'fetchSalesforceProduct2Candidates';
  jobId: string;
  conversationId: string;
  lookup: {
    product2Id?: string | null;
    skuId?: string | null;
    productCode?: string | null;
    name?: string | null;
  };
}

The orchestrator emits a short user-facing token ("I started a Salesforce Product2 lookup…"), yields the structured event, then yields { type: 'done' }. The SSE stream closes.

Why subAgentSummarizer.ts was touched

The summarizer strips most fields from sub-agent JSON before passing it back up. Without explicitly whitelisting them, the orchestrator would never see the queue marker.

// subAgentSummarizer.ts
const KEPT_FIELDS = [
  // …
  'salesforceLookupQueued',
  'jobName',
  'jobId',
  'lookup',
] as const;

5. The client polling-resume loop

useConfigAgent.ts grows a polling helper and a resume queue. The big idea: do not keep the chat stream open while waiting on a slow job. Instead, poll trpc.jobs.getStatus, then send a new message back into the same thread.

pollProduct2LookupJob(jobId, signal)
  // every 2_000 ms
  // up to 120_000 ms
  // active statuses:
  //   waiting | active |
  //   delayed | waiting-children

Queue, version, and busy-thread guards

Three new refs back the resume machinery, all keyed by thread id:

The finally block of sendMessage now only clears streaming state if the abort controller it owns is still the current one for the thread — preventing a slow finalize from clobbering a fresh send. After clearing, it calls flushProduct2ResumeQueue(threadId) to drain any queued resume messages.

Thread-id migration edge case

An UNASSIGNED thread becomes a real conversation id mid-stream. migrateThreadKey now also moves any queued resume messages and the resume version counter across, so a job that completes after the rename still posts to the right conversation.

6. The local-search performance fix

// effectively:
const all = await prisma.crmProduct
  .findMany({
    where: { organizationId: orgId },
    take: 5000,
  });
// then in-memory string match
// on productCode / name

prisma.crmProduct.findMany({
  where: {
    organizationId: orgId,
    crmProductData: {
      path: ['productCode'],
      equals: 'IDV',
    },
  },
  orderBy: { updatedAt: 'desc' },
  take: limit + 1,
});

7. Agent prompt and tool-rule changes

The orchestrator system prompt gains a matching SALESFORCE PRODUCT2 MAPPING section: include the chosen mapping in the user-facing action_plan, require one approval, and let crmProductId=null through with a warning rather than blocking.

8. What it doesn't change

9. Risks & open questions