Chaining lookups

Every method on the SDK returns a promise, so a multi-stage pipeline is ordinary async/await. The interesting part is not the syntax, it is that each stage can depend on what the previous stage discovered: you ingest a document, ask the graph what it mentioned, look those things up in a database Cortex has never seen, and ingest only what matched.

This page builds one such pipeline end to end.

The shape of it

1. Ingest the source document

   A call transcript goes in. Extraction pulls out the companies and people it mentions and links them to that row.

2. Ask what it found

   Read the document's neighbours in the graph and filter to organizations. This is the conditional: what you look up next depends on this answer.

3. Look those up in your own database

   Your CRM, warehouse, whatever. Cortex has no access to it and does not need any.

4. Ingest only the matches

   Not the whole CRM. Just the records for companies that actually came up.

5. Go one level further

   Now query a third source for investors and employees of those same companies, and ingest those too.

Step 1: ingest, and know when the graph is ready

This is the one piece of sequencing that will bite you, so it comes first.

pushRows resolves when the rows are stored, which is not the same moment the graph knows about them. Which of the two you get depends on one option:

Call	Promise resolves in	Entities queryable
pushRows(...)	~1.6s for one row	already there
pushRows(..., { deferEmbed: true })	~0.3s	~11s later

Both numbers are from one row on a live colony. deferEmbed is the right choice for a bulk backfill precisely because it returns before the slow work finishes. It is the wrong choice when your very next line reads the graph.

A small helper covers the deferred case. Poll rather than sleep a fixed duration: enrichment time scales with the document.

import type { CortexAPI, CortexGraphNode } from '@mitosislabs/sdk';

/** Wait until `universalId` has neighbours of `type`, or give up. */
async function waitForEntities(
  cortex: CortexAPI,
  officeId: string,
  universalId: string,
  { type = 'entity:organization', timeoutMs = 60_000, everyMs = 2_000 } = {},
): Promise<CortexGraphNode[]> {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    const { nodes } = await cortex.neighbors(officeId, universalId, { depth: 1 });
    const hits = nodes.filter((n) => n.type === type);
    if (hits.length > 0) return hits;
    await new Promise((r) => setTimeout(r, everyMs));
  }
  // An empty result is a real answer: the document may mention no companies.
  return [];
}

Step 2: ask the graph what the document mentioned

neighbors returns the exact subgraph around one item. After an ingest, its depth-1 neighbours are what extraction pulled out, joined by evidence edges.

const { nodes } = await cortex.neighbors(officeId, transcriptId, { depth: 1 });

const companies = nodes
  .filter((n) => n.type === 'entity:organization')
  .map((n) => n.displayName!);
// -> ['northwind robotics', 'calder biosciences']

Entity labels are normalized, which is what makes them usable as a lookup key across sources that spell things differently. Match on them case-insensitively rather than assuming your database uses the same casing.

Step 3 to 5: the whole pipeline

Three data sources, each consulted only because of what the previous one returned. crm and capTable stand in for your own databases; they are ordinary async functions and Cortex never sees them.

import { MitosisClient } from '@mitosislabs/sdk';

const mi = await MitosisClient.fromConfig();
const cortex = mi.cortex(officeId);

async function importCallAndExpand(transcript: string, callId: string) {
  // 1. The transcript itself. Inline, because step 2 depends on the graph.
  const { feedKey } = await cortex.ensureFeed(officeId, 'calls');
  const transcriptId = `ext_calls:integration_feed:${callId}`;
  await cortex.pushRows(officeId, feedKey, [
    { external_id: callId, title: `Call ${callId}`, content: transcript, modality: 'document' },
  ]);

  // 2. What did it mention? Everything downstream is conditional on this.
  const mentioned = await waitForEntities(cortex, officeId, transcriptId);
  const names = mentioned.map((n) => n.displayName!.toLowerCase());
  if (names.length === 0) return { companies: [], people: 0 };

  // 3. Your CRM. One round trip per company, in parallel: these do not depend
  //    on each other, so awaiting them in a loop would just be slower.
  const records = (await Promise.all(names.map((n) => crm.findCompany(n))))
    .filter((r): r is CompanyRecord => r !== null);

  // 4. Ingest ONLY the matches, not the CRM. Deferred: nothing below reads
  //    these back, so there is no reason to wait for embeddings.
  const { feedKey: crmFeed } = await cortex.ensureFeed(officeId, 'crm');
  await cortex.pushRows(
    officeId,
    crmFeed,
    records.map((c) => ({
      external_id: `company/${c.id}`,
      title: c.name,
      content: `${c.name}. ${c.description} Stage: ${c.stage}. HQ: ${c.hq}.`,
      modality: 'record',
      // Extra top-level keys stay on the row and are addressable by
      // extraction rules, so mentioned_in becomes a real edge back to the call.
      mentioned_in: transcriptId,
      company_id: c.id,
    })),
    { deferEmbed: true },
  );

  // 5. A third source, keyed on what step 3 actually matched.
  const people = (
    await Promise.all(records.map((c) => capTable.peopleFor(c.id)))
  ).flat();

  const { feedKey: peopleFeed } = await cortex.ensureFeed(officeId, 'people');
  await cortex.pushRows(
    officeId,
    peopleFeed,
    people.map((p) => ({
      external_id: `person/${p.id}`,
      title: p.name,
      content: `${p.name}, ${p.role} at ${p.companyName}.`,
      modality: 'record',
      company_id: p.companyId,
      relation: p.kind, // 'investor' | 'employee'
    })),
    { deferEmbed: true },
  );

  return { companies: records.map((c) => c.name), people: people.length };
}

Read the awaits as the dependency graph: steps 3 and 5 fan out with Promise.all because their calls are independent of each other, while the stages themselves stay sequential because each one needs the previous result.

Do not fan out without a bound

Promise.all over a handful of companies is fine. Promise.all over ten thousand opens ten thousand connections at once and takes down whichever database is smallest. Cap the width:

async function mapLimit<T, R>(items: T[], limit: number, fn: (t: T) => Promise<R>) {
  const out: R[] = [];
  for (let i = 0; i < items.length; i += limit) {
    out.push(...(await Promise.all(items.slice(i, i + limit).map(fn))));
  }
  return out;
}

const records = await mapLimit(names, 8, (n) => crm.findCompany(n));

The same applies to pushRows: batch around 25 rows per call rather than sending thousands in one request.

Partial failure

One dead lookup should not lose the work that succeeded. Promise.allSettled keeps the good results and lets you decide about the rest:

const settled = await Promise.allSettled(names.map((n) => crm.findCompany(n)));

const records = settled
  .filter((r): r is PromiseFulfilledResult<CompanyRecord | null> => r.status === 'fulfilled')
  .map((r) => r.value)
  .filter((c): c is CompanyRecord => c !== null);

const failed = names.filter((_, i) => settled[i].status === 'rejected');
if (failed.length) console.warn('CRM lookup failed for', failed);

Ingest is idempotent per external_id, so re-running the pipeline after fixing the outage updates the rows it already wrote instead of duplicating them. That is what makes a retry safe.

The callback form

If you would rather not hold the pipeline open, drive it from the change feed instead. pollLiveData calls you back as rows land, and returns a function that stops it:

import { pollLiveData } from '@mitosislabs/sdk';

const stop = pollLiveData(cortex, officeId, 'calls', async ({ events }) => {
  for (const e of events) {
    if (e.layer !== 'raw' || e.sourceTable !== 'integration_feed') continue;
    await expandOneCall(e.universalId); // steps 2 to 5, per row
  }
});

// later
stop();

Undoing a run

A pipeline that ingests derived records will eventually ingest wrong ones. Each stage wrote under its own feed, so each can be removed on its own, and purge previews by default:

// What would removing this run's CRM records take with it?
const plan = await cortex.purge(officeId, { source_schema: 'ext_crm' });
console.log(plan.matched, 'rows,', plan.entities_deleted, 'entities');

await cortex.purge(officeId, { source_schema: 'ext_crm', dry_run: false });