---
title: "Primary care: review, triage, extract — on a tokenized lake"
summary: "The whole primary-care surface in one walk. Stand up a tenant, a raw datalake and the tokenized and redacted copies, then run three scenarios end to end — a post-visit review request that only reaches patients whose appointment was actually fulfilled, an LLM agent triaging inbound contact-us messages into a category-keyed fan-out, and a vision agent reading a scanned document. Ends by reading one appointment back through all three lakes to show what each reader sees, then inspecting how far the tokenized copy has got and repairing one of its tables."
use_case: primary-care-feedback
slug: primary-care
status: draft
---

# Problem

A primary-care practice generates three kinds of work that all touch the same
patient and none of which a person should have to remember.

A visit finishes and the practice wants to ask how it went — but only about
visits that actually happened. A patient who cancelled must not be asked to
review the appointment they did not attend, and the difference between those
two cases is one field on one row.

Messages arrive through the website and most of them are not clinical. Some
are. Something has to read each one and route it before a human decides
whether it was urgent.

And documents arrive as scans. A referral, a registration form, an insurance
letter — pixels, from which someone has to type the same six fields into the
same system every time.

**This surface is tokenized, and that is the whole reason it is separate from
the other three cookbooks.** Everything here is patient data. The triage agent
in §023 reads inbound messages and has no business knowing who sent them; the
extraction agent in §031 reads documents. Both are machines, and the tokenized
copy exists so a machine can be given the signal without the identity. The
redacted copy exists for the other reader — a person on a support screen who
needs to see that a message came in without seeing whose it was.

# Composition

| Resource                          | Why it exists here                         |
|-----------------------------------|--------------------------------------------|
| Tenant + raw, tokenized, redacted | Three lakes, because two readers are masked|
| Appointments table                | The review workflow's event source         |
| Contact Us table                  | The triage workflow's event source         |
| SMS tool, LLM tool, vision tool   | One dispatcher, two models                 |
| Two AI agents                     | Triage, and document extraction            |
| Two workflows                     | Review request, and category fan-out       |
| Authenticated link                | The review form the SMS links to           |

# Walkthrough

This cookbook is **self-reliant**: it stands up everything it uses and depends
on no other file. It is also **idempotent** — every creating step looks first
and creates only what is missing, so running it twice costs what running it
once cost. That matters more here than on the raw surfaces: this walk
provisions three datalakes rather than one, and nothing in the platform
reclaims an abandoned lake.

## 001 — sign in as root, and make sure the admin user exists

Authenticate as the platform's root admin (`admin@dev.local` /
`devpassword` in local dev) via the tenantless bootstrap login — keyless by
structural necessity, since no tenant exists yet to scope a key to, and a
dev/test-only surface.

Then make sure the admin user this walk runs as exists. **The email is
stable, not per-run**, which is what makes the step idempotent — and it means
the second run finds the user already signed up. A duplicate signup is
refused, so the refusal is caught and inspected: if it says the email is
taken, that is the idempotent path and the walk continues. Any other failure
is re-raised, because swallowing it would turn a real auth problem into a
confusing failure three steps later.

```typescript
ctx.rootSession = await createBootstrapSession({
  baseUrl: process.env.ALVERA_BASE_URL!,
  email: process.env.ALVERA_ROOT_EMAIL!,
  password: process.env.ALVERA_ROOT_PASSWORD!,
})
ctx.rootApi = createIsolatedPlatformApi({
  baseUrl: process.env.ALVERA_BASE_URL!,
  sessionToken: ctx.rootSession.sessionToken,
  apiKey: '',
})

ctx.practiceEmail = 'cookbook-primary-care@dev.local'
ctx.practicePassword = 'CookbookPass1!'

try {
  const signUpResp = await ctx.rootApi.admin.signUp({
    email: ctx.practiceEmail,
    password: ctx.practicePassword,
    first_name: 'Cookbook',
    last_name: 'Practice',
  })
  await ctx.rootApi.admin.confirmUser(signUpResp.data.id!)
} catch (err) {
  const detail = JSON.stringify((err as { errors?: unknown }).errors ?? err)
  if (!/taken|already|exist/i.test(detail)) throw err
}
```

## 002 — the find-or-create helper every later step uses

Idempotence is one question asked over and over: *is this already here?*
Rather than answer it a dozen different ways, the walk defines it once.

`ensure` takes a label, a lookup and a create. It runs the lookup, returns
what it finds, and only creates when the lookup comes back empty. The label
is not decoration — when a run reuses something you did not expect it to, the
log line naming it is how you find out.

Alongside it, `firstNamed` — the lookup half. **A list endpoint pages at
twenty and is not newest-first**, so a `.find()` over the default page
silently stops finding things as soon as a lake has a few of them, and the
failure looks like the resource was never created. `page_size: 100` is the
cheap fix.

Both live on `ctx` rather than as bare functions because each numbered step
compiles into its own `it()` block.

```typescript
ctx.ensure = async <T>(
  label: string,
  find: () => Promise<T | undefined>,
  create: () => Promise<T>,
): Promise<T> => {
  const existing = await find()
  if (existing !== undefined) {
    console.log(`  ↻ reusing ${label}`)
    return existing
  }
  console.log(`  + creating ${label}`)
  return await create()
}

ctx.firstNamed = <T extends { name?: string }>(
  rows: readonly T[] | undefined,
  name: string,
): T | undefined => (rows ?? []).find((r) => r.name === name)
```

## 003 — the tenant

The practice admin signs in without a tenant scope — they may not belong to
one yet — and the walk finds or creates the tenant by its stable name. The
server derives the slug; capture it, because every later call is addressed by
it.

```typescript
ctx.practiceTenantlessSession = await createBootstrapSession({
  baseUrl: process.env.ALVERA_BASE_URL!,
  email: ctx.practiceEmail,
  password: ctx.practicePassword,
})
ctx.practiceTenantlessApi = createIsolatedPlatformApi({
  baseUrl: process.env.ALVERA_BASE_URL!,
  sessionToken: ctx.practiceTenantlessSession.sessionToken,
  apiKey: '',
})

const TENANT_NAME = 'Cookbook Primary Care'

const tenant = await ctx.ensure(
  `tenant ${TENANT_NAME}`,
  async () => {
    const { data } = await ctx.practiceTenantlessApi.tenants.list()
    return (data.data ?? []).find((t: { name?: string }) => t.name === TENANT_NAME)
  },
  async () => {
    const { data } = await ctx.practiceTenantlessApi.tenants.create({ name: TENANT_NAME })
    return data
  },
)
tenantSlug = tenant.slug!
```

## 004 — the tenant-scoped client, at the raw ceiling

A tenant-scoped login requires `X-API-Key`, so a key has to exist before the
admin can sign in against the tenant. Mint one through the platform-admin
side door with the root bearer; in the web console this is *Settings → API
Keys*.

`data_access_mode: 'raw'` is the **ceiling**, not the default view, and on a
tokenized surface the distinction finally matters. The key says how far this
session is *allowed* to see; each read then names the tier it wants. A raw
ceiling is what lets §033 ask the same row for all three answers and compare
them. A tokenized-ceiling key asking for `raw` is refused — which is the
right behaviour, and the reason a support console gets its own key.

**This is the one step in the walk that is not idempotent.** There is no
endpoint that lists a tenant's API keys, so `ensure` has no lookup to run and
each run mints another. A key row is cheap where a datalake is not.

```typescript
const { data: mintedKey } = await ctx.rootApi.admin.createTenantApiKey(tenantSlug, {
  name: 'Cookbook Primary Care Key',
  data_access_mode: 'raw',
})
ctx.tenantApiKey = mintedKey.api_key

const practiceTenantSession = await createSession({
  baseUrl: process.env.ALVERA_BASE_URL!,
  email: ctx.practiceEmail,
  password: ctx.practicePassword,
  tenantSlug,
  apiKey: ctx.tenantApiKey,
})
api = createIsolatedPlatformApi({
  baseUrl: process.env.ALVERA_BASE_URL!,
  sessionToken: practiceTenantSession.sessionToken,
  apiKey: ctx.tenantApiKey,
})
```

## 005 — the raw datalake

A datalake is created `raw` and stays raw. The derived copies are a separate
call, in §007, and this one has to be `ready` before that call is made.

The lookup filters on `type === 'raw'`, and here that is load-bearing rather
than defensive: from §007 onward `datalakes.list` returns **three** rows, two
of which are copies. A lookup that matched on name alone would start
returning whichever came first and the walk would begin writing to a lake
that is meant to be read-only.

Local-dev defaults match the seeded `dev.exs` setup — `postgres` on
`localhost:5432`, database `alvera_dev_foundation`, LocalStack S3 on
`localhost:4566` — and the schema name is stable, because a fresh schema per
run is the same leak as a fresh tenant per run.

```typescript
const DB = { host: 'localhost', port: 5432, user: 'postgres', pass: 'postgres', name: 'alvera_dev_foundation' }
const DB_SCHEMA = `cookbook_primary_care_${runSuffix}`
const LAKE_NAME = `Cookbook Primary Care Datalake ${runSuffix}`

const S3 = {
  cloud_storage_type: 'aws' as const,
  region: 'us-east-1',
  access_key_id: 'localkey',
  secret_access_key: 'localsecret',
  endpoint: 'http://localhost:4566',
}

const datalake = await ctx.ensure(
  `datalake ${LAKE_NAME}`,
  async () => {
    const { data } = await api.datalakes.list(tenantSlug)
    return (data.data ?? []).find(
      (l: { name?: string; type?: string }) => l.name === LAKE_NAME && l.type === 'raw',
    )
  },
  async () => {
    const { data } = await api.datalakes.create(tenantSlug, {
      name: LAKE_NAME,
      description: 'Primary care datalake provisioned by the cookbook doctest.',
      timezone: 'America/New_York',
      pool_size: 3,
      type: 'raw',

      db_writer_host: DB.host,
      db_writer_port: DB.port,
      db_writer_name: DB.name,
      db_writer_schema: DB_SCHEMA,
      db_writer_auth_method: 'password',
      db_writer_user: DB.user,
      db_writer_pass: DB.pass,
      db_writer_enable_ssl: false,
      db_reader_host: DB.host,
      db_reader_port: DB.port,
      db_reader_name: DB.name,
      db_reader_schema: DB_SCHEMA,
      db_reader_auth_method: 'password',
      db_reader_user: DB.user,
      db_reader_pass: DB.pass,
      db_reader_enable_ssl: false,

      cloud_storage: { ...S3, bucket: 'alvera-platform-dev', base_path: 'cookbook/primary-care' },
    })
    return data
  },
)
datalakeSlug = datalake.slug!
ctx.datalakeId = datalake.id!
```

## 006 — run the migrations, and wait for ready

`datalakes.create` persists the row at `status: 'new'`; it does not run the
schema DDL. Migration is triggered separately so the operator decides when
the potentially-slow part happens. `datalakes.migrate` enqueues the job and
returns immediately with `status: 'enqueued'`; the poll after it is what
waits for the worker.

**This wait is a precondition for §007, not politeness.** A derived lake is
provisioned empty and its own migration is what builds its tables and starts
the copy. Provision before the raw lake is ready and you get two lakes that
exist and stay empty forever.

```typescript
const migrateResp = await api.datalakes.migrate(tenantSlug, datalakeSlug)
if (migrateResp.data.status !== 'enqueued') {
  throw new Error(`datalake migration not enqueued (status: ${migrateResp.data.status})`)
}

const READY_TIMEOUT_MS = 5 * 60_000
const readyDeadline = Date.now() + READY_TIMEOUT_MS
let datalakeStatus: string | undefined
while (Date.now() < readyDeadline) {
  const { data } = await api.datalakes.get(tenantSlug, ctx.datalakeId)
  datalakeStatus = data.status
  if (datalakeStatus === 'ready') break
  await new Promise((r) => setTimeout(r, 5_000))
}
if (datalakeStatus !== 'ready') {
  throw new Error(`datalake did not reach :ready within ${READY_TIMEOUT_MS}ms (last: ${datalakeStatus})`)
}
```

## 007 — turn tokenization on, because two machines are going to read this

`provisionTokenization` creates both derived lakes and enqueues a migration
for each. It is **both-or-neither**: a run that provisions the tokenized lake
and cannot provision the redacted one reports the failure rather than leaving
half a pair. It is **idempotent**, so this step needs no `ensure` around it —
already-provisioned lakes come back as they are.

Poll the **derived** ids. The raw lake's own status says nothing about
whether its copies are ready, and a walk that waits on the wrong row gets a
`ready` it did not earn.

The statuses a derived lake passes through on the way are worth watching
rather than merely waiting out, because they are the difference between *this
run built these lakes* and *this run found them* — and on a cookbook whose
whole point is masking, that is the difference between having tested the copy
and having tested a copy someone else made.

The enum is `provisioning | new | processing | ready`, and **a derived lake
does not use the first of those.** `:provisioning` is set by exactly one
function, `Platform.Datalakes.provision_datalake/2`, which is deliberately
unreachable from `attrs` — and the tokenization path does not call it.
`provision_derived_datalakes/2` reuses the ordinary `create_datalake/2`
rather than opening a second create path, so a derived lake starts at the
default `:new` and climbs from there. A client watching for `provisioning`
would wait forever and conclude nothing was being built.

**That is the settled design, not an oversight to route around.** The stage
is reserved for a topology where each copy is a machine of its own and
standing one up takes minutes — there, a row exists while its database is
still being built, and `provisioning` is the honest thing to show. Today
every derived database is created *ahead* of its row, so no such moment
exists: the stage would be unreachable by any operator, and wiring it would
cost the connection probe on the one path that deliberately keeps it. A
missing database would become a row that quietly sits there instead of a
refusal you can read.

So the observation is keyed on `new` or `processing`, and it is a **report,
not an assertion**. A fast enough provision could have its first poll already
read `ready`, and failing the walk on that would be failing it for finishing
quickly.

```typescript
const { data: provisioned } = await api.datalakes.provisionTokenization(tenantSlug, datalakeSlug)
const derivedLakes = provisioned.derived_datalakes ?? []
if (derivedLakes.length !== 2) {
  throw new Error(`expected a tokenized + redacted pair, got ${derivedLakes.length} derived lake(s)`)
}

const DERIVED_TIMEOUT_MS = 5 * 60_000
const derivedDeadline = Date.now() + DERIVED_TIMEOUT_MS
const seenStatuses = new Set<string>()
let pendingLakes = derivedLakes.map((d) => d.id!)
while (Date.now() < derivedDeadline && pendingLakes.length > 0) {
  const stillPending: string[] = []
  for (const id of pendingLakes) {
    const { data } = await api.datalakes.get(tenantSlug, id)
    if (data.status) seenStatuses.add(data.status)
    if (data.status !== 'ready') stillPending.push(id)
  }
  pendingLakes = stillPending
  if (pendingLakes.length > 0) await new Promise((r) => setTimeout(r, 5_000))
}
if (pendingLakes.length > 0) {
  throw new Error(`derived datalakes did not reach :ready: ${pendingLakes.join(', ')}`)
}

ctx.provisionedFresh = seenStatuses.has('new') || seenStatuses.has('processing')
console.log(
  `  derived lakes ${ctx.provisionedFresh ? 'BUILT by this run' : 'already ready when first polled'} ` +
    `— statuses observed: ${[...seenStatuses].join(', ')}`,
)

// Capture the tokenized lake's slug — §012 reads its physical column types,
// and §033 reads a row through it.
const tokenizedLake = derivedLakes.find((d) => d.type === 'tokenized')
if (!tokenizedLake?.slug) throw new Error('no tokenized lake in the provisioned pair')
ctx.tokenizedLakeSlug = tokenizedLake.slug
```

## 008 — three helpers the scenarios below share

**`waitForFiredRun`.** `workflows.run` only *schedules* a run. It returns
immediately with a `workflow_run_id`, and the `workflow_run_log_id` and
`batch_id` a scenario needs are written later, when the run actually fires.
Poll until `workflow_run_log_id` is a **string** — not until `status` leaves
`'scheduled'`. Those are different moments, and because `typeof null ===
'object'` a predicate on status alone produces a baffling *"expected string,
got object"* rather than an obvious nil. Raise on `failed` carrying
`failure_reason` rather than polling to the deadline.

**`waitForBatches`.** Ingestion is async. **Gate on `dataset_updated`, never
on `rows_ingested`** — received is not written, and a batch whose row is
refused reports `rows_ingested: 1, dataset_updated: 0, status: 'partial'`
with the reason in `error`.

**`deployGenericTable`.** Creating a generic table does not deploy it. On
this surface the migration has more to do than on a raw one: it builds the
table in the raw lake **and** in both copies. The wait that means anything
asks for the table the same way the next step will, and retries until it
stops erroring.

```typescript
ctx.waitForFiredRun = async (
  runDatalakeSlug: string,
  runId: string,
  timeoutMs = 120_000,
): Promise<{ workflowRunLogId: string; batchId: string | null }> => {
  const deadline = Date.now() + timeoutMs
  let lastStatus: string | undefined
  while (Date.now() < deadline) {
    const { data } = await api.workflowRuns.get(tenantSlug, runDatalakeSlug, runId)
    lastStatus = data.status
    if (data.status === 'failed') {
      throw new Error(`workflow run ${runId} failed: ${data.failure_reason ?? 'no failure_reason given'}`)
    }
    if (typeof data.workflow_run_log_id === 'string') {
      return { workflowRunLogId: data.workflow_run_log_id, batchId: data.batch_id ?? null }
    }
    await new Promise((r) => setTimeout(r, 1_000))
  }
  throw new Error(`workflow run ${runId} never fired within ${timeoutMs}ms (last status: ${lastStatus})`)
}

ctx.waitForBatches = async (
  dacSlug: string,
  batchIds: readonly string[],
  timeoutMs = 120_000,
): Promise<void> => {
  const targets = new Set(batchIds)
  const deadline = Date.now() + timeoutMs
  let greenCount = 0
  while (Date.now() < deadline) {
    const { data } = await api.dataActivationClients.logs.list(tenantSlug, datalakeSlug, dacSlug)
    const green = new Set<string>()
    for (const row of (data.data ?? []) as Array<Record<string, unknown>>) {
      const b = row.batch_id
      if (typeof b !== 'string' || !targets.has(b)) continue
      if (row.status === 'partial' || row.status === 'failed') {
        throw new Error(
          `batch ${b} on ${dacSlug} did not persist its rows (status: ${row.status}): ` +
            `${String(row.error ?? 'no reason given')}`,
        )
      }
      if (typeof row.dataset_updated !== 'number' || row.dataset_updated < 1) continue
      const files = row.output_files
      if (!Array.isArray(files) || files.length === 0) continue
      green.add(b)
    }
    greenCount = green.size
    if (greenCount === targets.size) return
    await new Promise((r) => setTimeout(r, 1_000))
  }
  throw new Error(`only ${greenCount}/${targets.size} batches on ${dacSlug} persisted within ${timeoutMs}ms`)
}

ctx.deployGenericTable = async (tableName: string, timeoutMs = 180_000): Promise<void> => {
  await api.datalakes.migrate(tenantSlug, datalakeSlug)
  const deadline = Date.now() + timeoutMs
  let lastError: unknown
  while (Date.now() < deadline) {
    try {
      await api.datalakes.executeSql(tenantSlug, datalakeSlug, {
        sql: `SELECT 1 FROM ${tableName} LIMIT 1`,
        mode: 'raw',
      })
      return
    } catch (err) {
      lastError = err
      await new Promise((r) => setTimeout(r, 1_000))
    }
  }
  throw new Error(
    `generic table ${tableName} was not readable within ${timeoutMs}ms ` +
      `(last error: ${lastError instanceof Error ? lastError.message : String(lastError)})`,
  )
}
```

## 009 — the SMS tool and the review page

One SMS tool dispatches every outbound message in this walk. **The
per-message body lives on the workflow action, never on the tool**, which is
what lets one tool serve both the review request and the triage fan-out.
Local dev points at LocalStack's SNS on `:4566`, so nothing leaves the
machine.

The review form the SMS links to lives outside the platform, and it is two
addresses, neither registered with the platform. The **page** is the URL
written into the SMS body, exactly as the patient should end up seeing it.
The **link prefix** is the host you run in front of it: at send the platform
swaps every URL in the body for `<link_prefix>/<token>`, a link minted for
this one message, and a tap reaches that host first. The host asks the
platform what the link is, then sends the patient on to the form — §019
does exactly that. A path is allowed in the prefix; a query, a fragment or a
trailing slash is not, because the platform appends `/<token>` itself.

```typescript
const smsTool = await ctx.ensure(
  'SMS tool',
  async () => {
    const { data } = await api.tools.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook Practice SMS Tool')
  },
  async () => {
    const { data } = await api.tools.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook Practice SMS Tool',
      description: 'SNS-backed SMS dispatcher for review requests and triage, wired to LocalStack.',
      intent: 'sms',
      status: 'active',
      datalake_id: ctx.datalakeId,
      body: {
        tool_body_type: 'sns',
        auth_method: 'access_key',
        region: 'us-east-1',
        phone_number: '+15551234567',
        endpoint_url: 'http://localhost:4566',
        access_key_id: 'localkey',
        secret_access_key: 'localsecret',
      },
    })
    return data
  },
)
toolId = smsTool.id!
ctx.smsToolId = smsTool.id!

ctx.reviewFormUrl = 'https://review.example.local/review/visit'
ctx.reviewLinkPrefix = 'https://go.review.example.local/l'
```

## 010 — the appointments table, and the one column that exercises everything

GH-859 removed the `appointment` dataset along with every other per-domain
resource, so an appointment is a table you declare and the columns are yours
rather than FHIR's.

`appointment_id` is unique, so re-ingesting the same export updates those
rows instead of doubling them. `status` is the canonical value the filter
reads — the vendor's `appt_slot_status` string is mapped onto it by the
contract in §012, so the filter never has to know about `"3 - checked out"`.
`patient_name` and `patient_mobile` are `tokenize`; they ride on the row
because that is what lets the action templates read `event_dataset.*` rather
than reaching back through the resolved subject.

**`care_gaps` is the column worth reading closely.** It is `jsonb`, **and
`is_array: true`, and `redact`** — all three at once, which no other column
here is. Each half matters:

- `is_array: true` is not optional and its absence does not look like a type
  error. Without it the column is a single JSON document, and a template that
  renders a list into it is refused per row with
  `upsert_row_failed %{care_gaps: ["is invalid"]}` while the batch reports
  itself green.
- `redact` is what puts it in front of the masking machinery. A column that
  is only ever empty proves nothing about masking, which is why §014 ingests
  gaps that actually have entries in them.

Do **not** declare a `legal_entity_id` column. The platform stamps it from
Contract B's resolve, and the name is reserved — declaring it does not shadow
the stamp, it 422s the create, so you lose the table until you drop it.

```typescript
const appointments = await ctx.ensure(
  'appointments table',
  async () => {
    const { data } = await api.genericTables.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return (data.data ?? []).find((t: { title?: string }) => t.title === 'Cookbook CAHPS Appointments')
  },
  async () => {
    const { data } = await api.genericTables.create(tenantSlug, datalakeSlug, {
      title: 'Cookbook CAHPS Appointments',
      description: 'Appointments loaded from the practice management CAHPS export.',
      columns: [
        { name: 'appointment_id', title: 'Appointment ID', type: 'string', description: 'Vendor appointment id — unique per appointment', is_unique: true, privacy_requirement: 'none' },
        { name: 'status', title: 'Status', type: 'string', description: 'Canonical status: fulfilled / cancelled / arrived / booked / pending', is_unique: false, privacy_requirement: 'none' },
        { name: 'appt_start', title: 'Appointment Start', type: 'string', description: 'Local start timestamp, ISO 8601', is_unique: false, privacy_requirement: 'none' },
        { name: 'minutes_duration', title: 'Duration (minutes)', type: 'integer', description: 'Slot length in minutes', is_unique: false, privacy_requirement: 'none' },
        { name: 'description', title: 'Description', type: 'string', description: 'Appointment type as the practice records it', is_unique: false, privacy_requirement: 'none' },
        { name: 'department', title: 'Department', type: 'string', description: 'Servicing department', is_unique: false, privacy_requirement: 'none' },
        { name: 'provider_name', title: 'Provider Name', type: 'string', description: 'Rendering provider display name', is_unique: false, privacy_requirement: 'none' },
        { name: 'patient_name', title: 'Patient Name', type: 'string', description: 'Patient display name — rendered into the SMS greeting', is_unique: false, privacy_requirement: 'tokenize' },
        { name: 'patient_mobile', title: 'Patient Mobile', type: 'string', description: 'Patient mobile — the SMS destination', is_unique: false, privacy_requirement: 'tokenize' },
        { name: 'patient_ref', title: 'Patient Ref', type: 'string', description: "The vendor's own patient id, for tracing a row back to the export", is_unique: false, privacy_requirement: 'tokenize' },
        { name: 'care_gaps', title: 'Care Gaps', type: 'jsonb', description: 'Open care gaps, as a JSON array', is_unique: false, is_array: true, privacy_requirement: 'redact' },
      ],
    })
    return data
  },
)
genericTableId = appointments.id!
ctx.appointmentsTableId = appointments.id!
ctx.appointmentsTableName = appointments.name!

await ctx.deployGenericTable(ctx.appointmentsTableName)
```

## 011 — the data source and the manual-upload tool

The chain's first link is a **data source** — a registration of where the
rows originate. Its `uri` is the system-of-record address, and both contracts
render it onto the identifier, so it is half of the key that collapses them
onto one patient.

The **manual-upload tool** needs no endpoint and no credentials, because the
rows arrive in the ingest call's own body rather than being fetched.

```typescript
const dataSource = await ctx.ensure(
  'athenahealth data source',
  async () => {
    const { data } = await api.dataSources.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook Athenahealth Source')
  },
  async () => {
    const { data } = await api.dataSources.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook Athenahealth Source',
      uri: 'athenahealth.example.com/cookbook-primary-care',
      description: 'Practice management system — origin of the CAHPS appointment export.',
      status: 'active',
      is_default: false,
    })
    return data
  },
)
dataSourceId = dataSource.id!

const manualUploadTool = await ctx.ensure(
  'manual-upload tool',
  async () => {
    const { data } = await api.tools.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook Practice Manual Upload Tool')
  },
  async () => {
    const { data } = await api.tools.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook Practice Manual Upload Tool',
      description: 'Manual-upload data-exchange tool — backs the appointment ingest.',
      intent: 'data_exchange',
      status: 'active',
      datalake_id: ctx.datalakeId,
      data_source_id: dataSourceId,
      body: { tool_body_type: 'manual_upload' },
    })
    return data
  },
)
ctx.manualUploadToolId = manualUploadTool.id!
```

## 012 — the CAHPS contract pair, on two clients

One inbound CAHPS row becomes two things. **Contract A**
(`resource_type: 'legal_entity'`) writes the patient — the subject — and its
`mdm_input_config` is `{ type: 'null' }`, because a template that already
*is* the subject has nothing to resolve. **Contract B**
(`resource_type: 'generic_table'`) writes the appointment row, and its
`mdm_input_config` emits the same `(uri, patient_id)` pair so both converge
on one patient and the row earns its `legal_entity_id` stamp.

**The pair is split across two clients, and the ingest in §014 is staged.** A
client fans out per `(row, contract)` and the job key staggers by *row*, so a
pair bound to one client runs both contracts on the same row at the same
instant. Both find-or-create the same patient, both miss the other's
uncommitted write, and the unique index on `(uri, id_type, id_number)`
refuses the loser. The legal entity is deliberately not in that key, which is
what makes the refusal correct rather than a bug.

**Watch the identifier shapes.** `Platform.MDMInput` takes `uri` / `type` /
`value`; a legal-entity identification takes `id_type` / `id_number` / `uri`.
`system` is a genuine alias on the MDM side — cast and normalised onto `uri`
— but `id_type` and `id_number` are not cast at all, and unknown keys are
dropped rather than refused. And the reason the two look interchangeable is
that one becomes the other: `MDMInput.to_identification/1` takes
`(uri, type, value)` and returns `(uri, id_type, id_number)`, so the
identifier reads back under names it will not accept on the way in.

```typescript
const { readFileSync } = await import('node:fs')
const { join } = await import('node:path')
const read = (f: string) =>
  readFileSync(join(process.env.COOKBOOK_FIXTURES_DIR!, 'primary-care-feedback', f), 'utf8')

const leContract = await ctx.ensure(
  'patient contract',
  async () => {
    const { data } = await api.interoperabilityContracts.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook CAHPS Patient Contract')
  },
  async () => {
    const { data } = await api.interoperabilityContracts.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook CAHPS Patient Contract',
      description: 'CAHPS appointment row → the patient LegalEntity.',
      resource_type: 'legal_entity',
      type: 'identity',
      generic_table_id: null,
      template_config: { type: 'custom', body: read('_cahps_appointments_legal_entity.liquid') },
      mdm_input_config: { type: 'null' },
    })
    return data
  },
)
ctx.leContractId = leContract.id!

const gtContract = await ctx.ensure(
  'appointment contract',
  async () => {
    const { data } = await api.interoperabilityContracts.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook CAHPS Appointment Contract')
  },
  async () => {
    const { data } = await api.interoperabilityContracts.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook CAHPS Appointment Contract',
      description: 'CAHPS appointment row → appointments table row, stamped with its patient.',
      resource_type: 'generic_table',
      type: 'identity',
      generic_table_id: ctx.appointmentsTableId,
      template_config: { type: 'custom', body: read('_cahps_appointments_generic_table.liquid') },
      mdm_input_config: { type: 'custom', body: read('_cahps_appointments_mdm.liquid') },
    })
    return data
  },
)
interopContractId = gtContract.id!
ctx.gtContractId = gtContract.id!
```

## 013 — the two activation clients

Both share the tool and the data source. Only the contract list differs, and
the order they run in is the whole point: identity first, then the row that
resolves against it.

```typescript
const identityDac = await ctx.ensure(
  'patient identity client',
  async () => {
    const { data } = await api.dataActivationClients.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook CAHPS Patient Identity DAC')
  },
  async () => {
    const { data } = await api.dataActivationClients.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook CAHPS Patient Identity DAC',
      description: 'Writes the patient LegalEntity for each appointment row. Runs FIRST.',
      tool_id: ctx.manualUploadToolId,
      data_source_id: dataSourceId,
      tool_call: { tool_call_type: 'manual_upload' },
      interoperability_contract_ids: [ctx.leContractId],
    })
    return data
  },
)
ctx.identityDacSlug = identityDac.slug!

const dataDac = await ctx.ensure(
  'appointment data client',
  async () => {
    const { data } = await api.dataActivationClients.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook CAHPS Appointment Data DAC')
  },
  async () => {
    const { data } = await api.dataActivationClients.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook CAHPS Appointment Data DAC',
      description: 'Writes the appointment row and stamps it with the patient. Runs SECOND.',
      tool_id: ctx.manualUploadToolId,
      data_source_id: dataSourceId,
      tool_call: { tool_call_type: 'manual_upload' },
      interoperability_contract_ids: [ctx.gtContractId],
    })
    return data
  },
)
dacId = dataDac.id!
ctx.dataDacSlug = dataDac.slug!
```

## 014 — two appointments, one attended and one not

Two rows in **the vendor's own shape**, which is why every key below looks
slightly wrong: `appt_slot_status` rather than `status`,
`patient_mobile_no` rather than `patient_mobile`, `rndrng_provider` rather
than `provider_name`. Renaming those onto the table's columns is Contract B's
entire job, and ingesting the canonical names instead would bypass the
contract and prove nothing about it.

The status mapping is the visible half — the export says `"3 - checked out"`
and `"x - cancelled"`, and the contract maps them onto `fulfilled` and
`cancelled` so the filter never has to know about vendor strings. The field
renames are the quiet half, and they fail differently: a key the template does
not read is not an error. It is simply absent, the column lands `null`, the
ingest reports itself green, and the failure surfaces much later as an SMS
action rejecting an empty recipient. §015 exists to catch that on the row
rather than three steps downstream.

**Both rows carry `care_gaps` with real entries.** That is deliberate. A
jsonb array column that is only ever `[]` still exists and still replicates,
so a walk that never populates it will go green while proving nothing about
whether a *populated* array survives the copy — which is the only version of
the question anyone cares about. §016 reads them back.

```typescript
ctx.fulfilledId = 'APPT-COOKBOOK-0001'
ctx.cancelledId = 'APPT-COOKBOOK-0002'

const appointmentRows = [
  {
    appointment_id: ctx.fulfilledId,
    appt_slot_status: '3 - checked out',
    appt_date: '3/12/2026',
    appt_start_time: '2:30 PM',
    appt_slot_duration: 30,
    appt_type: 'Annual wellness visit',
    svc_department: 'Family Medicine',
    rndrng_provider_id: 'PROV-114',
    rndrng_provider: 'Dr. Amara Osei',
    patient_name: 'Rosalind Achterberg',
    patient_mobile_no: '+12025554101',
    patient_id: 'PAT-COOKBOOK-0001',
    care_gaps: [
      { code: 'A1C', description: 'HbA1c not recorded in 12 months', due_date: '2026-04-30' },
      { code: 'BP', description: 'Blood pressure follow-up overdue', due_date: '2026-03-15' },
    ],
  },
  {
    appointment_id: ctx.cancelledId,
    appt_slot_status: 'x - cancelled',
    appt_date: '3/13/2026',
    appt_start_time: '9:00 AM',
    appt_slot_duration: 20,
    appt_type: 'Follow-up',
    svc_department: 'Family Medicine',
    rndrng_provider_id: 'PROV-114',
    rndrng_provider: 'Dr. Amara Osei',
    patient_name: 'Teodoro Vasquez-Lindqvist',
    patient_mobile_no: '+12025554102',
    patient_id: 'PAT-COOKBOOK-0002',
    care_gaps: [{ code: 'FLU', description: 'Influenza vaccination not on file', due_date: '2026-10-01' }],
  },
]

// Stage one — the patients.
const identityIngests = await Promise.all(
  appointmentRows.map((row) =>
    api.dataActivationClients.ingest(tenantSlug, datalakeSlug, ctx.identityDacSlug, { data: row }),
  ),
)
await ctx.waitForBatches(
  ctx.identityDacSlug,
  identityIngests.map((r: { data: { batch_id?: string } }) => r.data.batch_id!),
)

// Stage two — the appointment rows, now that every patient exists.
const dataIngests = await Promise.all(
  appointmentRows.map((row) =>
    api.dataActivationClients.ingest(tenantSlug, datalakeSlug, ctx.dataDacSlug, { data: row }),
  ),
)
await ctx.waitForBatches(
  ctx.dataDacSlug,
  dataIngests.map((r: { data: { batch_id?: string } }) => r.data.batch_id!),
)
```

## 015 — the rows landed, the mapping ran, and the stamp is on

Four assertions in one read, and each catches a different failure.

The **row count** catches an ingest that reported green over nothing. The
**status values** catch a contract whose mapping silently fell through to the
`pending` default — which would leave the filter in §018 with nothing to
route and make the whole scenario a false green. The **stamp** catches a
resolve that never ran, which would leave every link naming nobody — so the
fulfilled visit's stamp is kept, and §019 checks the review link names that
same patient.

And **`patient_mobile` must not be null**, which is the one that catches a
misspelled vendor key. A template reading a field the payload does not carry
is not an error — Liquid renders nothing, the column lands `null`, and the
batch reports itself green. The consequence appears two steps later as an SMS
action failing `output_schema_violation` on an empty recipient, which reads
like a broken tool rather than a missing letter in an ingest key. Assert the
column on the row, where the answer is unambiguous.

```typescript
const stampDeadline = Date.now() + 120_000
let rows: unknown[][] = []
while (Date.now() < stampDeadline && rows.length < 2) {
  const { data: result } = await api.datalakes.executeSql(tenantSlug, datalakeSlug, {
    sql: `SELECT appointment_id, status, legal_entity_id, patient_mobile
          FROM ${ctx.appointmentsTableName}
          WHERE appointment_id IN ('${ctx.fulfilledId}', '${ctx.cancelledId}')
          ORDER BY appointment_id`,
    mode: 'raw',
  })
  if (typeof result !== 'string') rows = result.data
  if (rows.length < 2) await new Promise((r) => setTimeout(r, 1_000))
}
if (rows.length !== 2) {
  throw new Error(`expected 2 appointment rows, found ${rows.length} within 120s`)
}

const byId = new Map(
  rows.map((r) => [String(r[0]), { status: String(r[1]), stamp: r[2], mobile: r[3] }]),
)
if (byId.get(ctx.fulfilledId)?.status !== 'fulfilled') {
  throw new Error(
    `the contract did not map "3 - checked out" onto fulfilled — got ` +
      `${byId.get(ctx.fulfilledId)?.status}`,
  )
}
if (byId.get(ctx.cancelledId)?.status !== 'cancelled') {
  throw new Error(
    `the contract did not map "x - cancelled" onto cancelled — got ` +
      `${byId.get(ctx.cancelledId)?.status}`,
  )
}
for (const [id, row] of byId) {
  if (!row.stamp) throw new Error(`appointment ${id} landed with no legal_entity_id stamp`)
  if (!row.mobile) {
    throw new Error(
      `appointment ${id} landed with a null patient_mobile — the contract template reads ` +
        `the vendor key 'patient_mobile_no', so a payload using the column name writes nothing`,
    )
  }
}
ctx.fulfilledPatientId = String(byId.get(ctx.fulfilledId)!.stamp)
```

## 016 — a populated masked array survives the copy

This is the step that proves the derived lakes are real, and it is the
cheapest one in the file to get wrong by writing something weaker.

`care_gaps` is `jsonb` + `is_array` + `redact`, and what happens to that
column across the three lakes is worth stating exactly, because the obvious
guess is wrong.

**The column is `jsonb` in the raw lake and `text` in both copies. The types
are deliberately different, and that is what makes replication work.** Logical
replication ships the publisher's text representation and hands it to the
subscriber column's input function. Any jsonb value renders to a string and
any string is valid `text`, so `jsonb → text` is always accepted. What is
refused is `jsonb → text[]`, because `[{"code":"A1C",…}]` is JSON and not a
Postgres array literal — and a masking policy that widened the column to an
array type in the copies would fail the apply worker on the first row
carrying one.

So the invariant is **scalar-ness, not type parity**: a jsonb column is never
a Postgres array type, in any lake. `is_array: true` describes the JSON
document the column holds. It is not an instruction to make the Postgres
column an array, and reading it that way is the mistake that produces a
derived lake which exists, reports `ready`, and silently never fills — a
failure with no error surface a client can see.

None of that is checkable from here. `executeSql` refuses any query touching
`information_schema` or `pg_catalog` — they are blocked tables — so a client
cannot read a column's physical type at all, in any mode. The mechanism is
therefore something to know rather than something to assert, and **the walk
asserts its consequence instead**: a row carrying a populated array must
arrive in both copies. That is directly observable, and it is the claim
anyone actually depends on.

It also explains the shape of the masked value below, which is otherwise
baffling. The copies hand back a masked *string*, not an array, because on
that side the column really is text and carries no declared semantic kind —
so the anonymizer masks its string rendering the way it masks any free text.

Read the same row through all three lakes. `mode` is a parameter of the read,
not a different slug: one lake is addressed and `mode` selects which copy
answers. This works only because §004 built its client at the `raw` ceiling.

```typescript
const gapsSql =
  `SELECT appointment_id, care_gaps, minutes_duration ` +
  `FROM ${ctx.appointmentsTableName} WHERE appointment_id = '${ctx.fulfilledId}'`

const readGapsAs = async (mode: 'raw' | 'tokenized' | 'redacted') => {
  const deadline = Date.now() + 120_000
  while (Date.now() < deadline) {
    const { data } = await api.datalakes.executeSql(tenantSlug, datalakeSlug, { sql: gapsSql, mode })
    const row = (data.data ?? [])[0]
    if (row) {
      const cell = (column: string): unknown => row[data.meta.columns.indexOf(column)]
      return { gaps: cell('care_gaps'), minutes: cell('minutes_duration') }
    }
    await new Promise((r) => setTimeout(r, 2_000))
  }
  throw new Error(`the appointment never became readable through the ${mode} lake within 120s`)
}

const rawGaps = await readGapsAs('raw')
const tokenizedGaps = await readGapsAs('tokenized')
const redactedGaps = await readGapsAs('redacted')

console.log('  raw       →', JSON.stringify(rawGaps))
console.log('  tokenized →', JSON.stringify(tokenizedGaps))
console.log('  redacted  →', JSON.stringify(redactedGaps))

// The array is populated in the primary. A walk whose array is always empty
// proves nothing about arrays.
const rawText = JSON.stringify(rawGaps.gaps)
if (!rawText.includes('A1C') || !rawText.includes('BP')) {
  throw new Error(`the raw lake does not hold the two care gaps that were ingested: ${rawText}`)
}

// The row REACHED both copies. Had the masked column been widened to a
// Postgres array type, the apply worker would have stopped on this row and
// these reads would time out against a lake that reports itself ready.
for (const [mode, read] of [['tokenized', tokenizedGaps], ['redacted', redactedGaps]] as const) {
  if (read.gaps === undefined || read.gaps === null) {
    throw new Error(`the ${mode} lake has no care_gaps value — replication did not apply the row`)
  }
  if (Number(read.minutes) !== Number(rawGaps.minutes)) {
    throw new Error(
      `the ${mode} lake changed a 'none' column: minutes_duration ` +
        `${read.minutes} vs ${rawGaps.minutes}`,
    )
  }
}

// And the clinical detail does NOT survive it. `redact` is the strictest
// tier, so the copies must not hand back the gap codes.
for (const [mode, read] of [['tokenized', tokenizedGaps], ['redacted', redactedGaps]] as const) {
  const masked = JSON.stringify(read.gaps)
  if (masked.includes('A1C') || masked.includes('HbA1c')) {
    throw new Error(`the ${mode} lake handed back the real care gaps — redaction is not in force: ${masked}`)
  }
}
```

## 017 — the Review SMS workflow

Standard shape — filter, decision, one action — targeting §010's table.

`skip_mdm_resolution: false` is set explicitly, and must be. A generic-table
row is never its own subject; the subject is the patient §012's pair stamped
onto it. With resolution on, the platform loads that subject, and the message
— and the link minted with it — names *this patient* as its
`legal_entity_id`. Skip it and the link still resolves, but it cannot say
whose review it is.

The body carries §009's form URL itself; the action's `link_prefix` is what
turns it into this message's own link at send, and it is required on every
SMS call. `metadata_template` renders once per send and rides on the
message, so the host gets the appointment back with the link.

The filter is one comparison — `status == "fulfilled"` — and it carries the
whole clinical rule this scenario exists for: never ask someone to review a
visit they did not attend. That the rule is one line of Liquid on the
workflow, rather than a branch in a service, is the point of the primitive.
A reviewer can read it.

```typescript
const REVIEW_KEY = 'send_review_request'

const reviewWorkflow = await ctx.ensure(
  'Review SMS workflow',
  async () => {
    const { data } = await api.workflows.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook Visit Review Workflow')
  },
  async () => {
    const { data } = await api.workflows.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook Visit Review Workflow',
      description: 'Asks for a review after a visit that actually happened.',
      dataset_type: 'generic_table',
      generic_table_id: ctx.appointmentsTableId,
      status: 'live',
      tags: ['cahps', 'review'],
      skip_mdm_resolution: false,
      filter_config: {
        type: 'custom',
        body: '{% if event_dataset.status == "fulfilled" %}true{% endif %}',
      },
      decision_config: {
        type: 'custom',
        body: `["${REVIEW_KEY}"]`,
        output_schema: { type: 'array', items: { type: 'string' } },
      },
      context_datasets: [
        {
          dataset_type: 'message',
          where_clause:
            `m.legal_entity_id = '{{ legal_entity_id }}' AND m.decision_key = '${REVIEW_KEY}' ` +
            "AND m.sent_at > NOW() - INTERVAL '90 days'",
          limit: 1,
          position: 0,
        },
      ],
      actions: [
        {
          action_type: 'sms',
          tool_id: ctx.smsToolId,
          decision_key: REVIEW_KEY,
          position: 0,
          trigger_template: 'now',
          idempotency_template: `{{ event_dataset.appointment_id }}-${REVIEW_KEY}`,
          metadata_template: '{"appointment_id":"{{ event_dataset.appointment_id }}"}',
          tool_call: {
            tool_call_type: 'sms_request',
            to: { type: 'custom', body: '{{ event_dataset.patient_mobile }}' },
            body: {
              type: 'custom',
              body:
                'Hi {{ event_dataset.patient_name }}, thanks for visiting ' +
                '{{ event_dataset.department }}. How did we do? ' +
                ctx.reviewFormUrl,
            },
            sms_type: 'transactional',
            link_prefix: ctx.reviewLinkPrefix,
          },
        },
      ],
    })
    return data
  },
)
workflowId = reviewWorkflow.id!
ctx.reviewWorkflowSlug = reviewWorkflow.slug!

if (reviewWorkflow.skip_mdm_resolution !== false) {
  throw new Error('a workflow that mints a per-patient link must keep MDM resolution ON')
}
```

## 018 — run it, and the cancelled visit is filtered

`manual_override: false` is what makes this a real test — the filter is
evaluated, so `status` genuinely routes each row. With `true` the filter is
bypassed and the cancelled appointment would get a review request, which is
the exact outcome the scenario exists to prevent.

The assertion is at the execution-log level, which is what lets it survive a
rerun: the action inside the passing row's log refuses to fire a second time
because its idempotency key already fired, but the row still passes the
filter, so the one-passed-one-filtered split is identical on every run.

```typescript
const runResp = await api.workflows.run(tenantSlug, datalakeSlug, ctx.reviewWorkflowSlug, {
  sql_where_clause: `appointment_id IN ('${ctx.fulfilledId}', '${ctx.cancelledId}')`,
  mode: 'live',
  manual_override: false,
})
const fired = await ctx.waitForFiredRun(datalakeSlug, runResp.data.workflow_run_id)
ctx.reviewRunBatchId = fired.batchId!

const refreshDeadline = Date.now() + 120_000
let status: string | null = null
while (Date.now() < refreshDeadline) {
  const { data: log } = await api.workflows.batchLogs.refresh(
    tenantSlug, datalakeSlug, ctx.reviewWorkflowSlug, fired.workflowRunLogId,
  )
  status = log.status ?? null
  if (status && status !== 'pending') break
  await new Promise((r) => setTimeout(r, 2_000))
}
if (status === 'failed') throw new Error('the review run reached :failed')
if (!status || status === 'pending') throw new Error('the review run did not leave :pending within 120s')

const welDeadline = Date.now() + 90_000
let ourLogs: Array<Record<string, unknown>> = []
let byStatus: Record<string, number> = {}
while (Date.now() < welDeadline) {
  const { data: wfLogs } = await api.workflows.workflowLogs.list(
    tenantSlug, datalakeSlug, ctx.reviewWorkflowSlug, { page_size: 100 },
  )
  ourLogs = (wfLogs.data ?? [])
    .map((w) => w as Record<string, unknown>)
    .filter((w) => w.batch_id === ctx.reviewRunBatchId)
  byStatus = {}
  for (const w of ourLogs) {
    const st = (w.status as string | undefined) ?? 'unknown'
    byStatus[st] = (byStatus[st] ?? 0) + 1
  }
  const settled = (byStatus.completed ?? 0) + (byStatus.executing ?? 0)
  if (ourLogs.length === 2 && settled === 1 && (byStatus.filtered ?? 0) === 1) break
  await new Promise((r) => setTimeout(r, 2_000))
}
if (ourLogs.length !== 2) {
  throw new Error(`expected 2 execution logs for the review run, got ${ourLogs.length}`)
}
if ((byStatus.executing ?? 0) + (byStatus.completed ?? 0) !== 1) {
  throw new Error(`expected 1 fulfilled visit past the filter — ${JSON.stringify(byStatus)}`)
}
if ((byStatus.filtered ?? 0) !== 1) {
  throw new Error(`expected the cancelled visit to be :filtered — ${JSON.stringify(byStatus)}`)
}
```

## 019 — the review request went out, and its link resolves

Search on the **idempotency key**, never on `workflow_id`. The key is
`<appointment id>-send_review_request` and the appointment id is stable, so it
names the same message forever. A workflow id is stable only as long as the
workflow resource is.

The stored body is the body as sent, so the form URL is not in it: in its
place stand the link prefix and a fifteen-character token. Finding the form
URL itself would mean no link was minted.

Then play the host behind the prefix. Hand the token to
`authenticatedLinks.exchange` with what the tap's request said about itself
— a phone's browser, because the platform judges from those details whether
a person tapped, and only a person's tap stamps `link_clicked_at` and the
message's `clicked_at`. The answer names the patient and carries the
appointment on `message.metadata`. Both calls need only the tenant's
publishable key.

Then the form reports what the patient did, with the same token. The message
status only moves forward, and `opened` ranks below `clicked`, so reporting
`opened` after the tap leaves the status at `clicked` while `opened_at`
still lands; `form_submitted` ranks above both, so the message ends there.

```typescript
const { data: search } = await api.datasets.createUserSearch(tenantSlug, datalakeSlug, 'message', {
  search_query: `m.idempotency_key = '${ctx.fulfilledId}-send_review_request'`,
})
if (search.status !== 'completed') {
  throw new Error(`message user-search status=${search.status} error=${search.error_message ?? '(none)'}`)
}

const msgDeadline = Date.now() + 90_000
let reviewBody: string | undefined
while (Date.now() < msgDeadline && !reviewBody) {
  const { data } = await api.datasets.search(tenantSlug, datalakeSlug, 'message', {
    userSearchId: search.id!,
    dataAccessMode: 'raw',
  })
  reviewBody = ((data.data ?? []) as Array<Record<string, unknown>>)
    .map((m) => String(m.body ?? ''))
    .find((b) => b.includes('How did we do?'))
  if (!reviewBody) await new Promise((r) => setTimeout(r, 2_000))
}
if (!reviewBody) {
  throw new Error(`no rendered review SMS for ${ctx.fulfilledId} within 90s`)
}
if (reviewBody.includes(ctx.reviewFormUrl)) {
  throw new Error(`the form URL went out as written, so no link was minted: ${reviewBody}`)
}
const reviewLinkStart = `${ctx.reviewLinkPrefix}/`
const reviewToken = reviewBody.split(reviewLinkStart)[1]?.match(/^[A-Za-z0-9_-]{15}/)?.[0]
if (!reviewToken) throw new Error(`no ${reviewLinkStart}<token> in the review body: ${reviewBody}`)

const { data: tap } = await api.authenticatedLinks.exchange(tenantSlug, datalakeSlug, {
  token: reviewToken,
  method: 'GET',
  user_agent:
    'Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 ' +
    '(KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1',
  headers: { 'accept-language': 'en-US' },
})
if (tap.destination_url !== ctx.reviewFormUrl) {
  throw new Error(`the review link should lead to ${ctx.reviewFormUrl}, got ${tap.destination_url}`)
}
if (tap.legal_entity_id !== ctx.fulfilledPatientId) {
  throw new Error(`the review link should name patient ${ctx.fulfilledPatientId}, got ${tap.legal_entity_id}`)
}
if (tap.message.metadata?.appointment_id !== ctx.fulfilledId) {
  throw new Error(`metadata_template did not reach the message: ${JSON.stringify(tap.message.metadata)}`)
}
if (!tap.link_clicked_at || !tap.message.clicked_at) {
  throw new Error("a person's tap should stamp link_clicked_at and the message's clicked_at")
}

const now = new Date().toISOString()
await api.authenticatedLinks.updateMessageTracking(tenantSlug, datalakeSlug, {
  token: reviewToken,
  status: 'opened',
  opened_at: now,
})
const { data: tracked } = await api.authenticatedLinks.updateMessageTracking(tenantSlug, datalakeSlug, {
  token: reviewToken,
  status: 'form_submitted',
  form_submitted_at: now,
})
if (!tracked.message.opened_at || !tracked.message.form_submitted_at) {
  throw new Error('message tracking did not persist opened_at + form_submitted_at')
}
if (tracked.message.status !== 'form_submitted') {
  throw new Error(`the message should end at form_submitted, got ${tracked.message.status}`)
}
```

## 020 — the Contact Us table, and its client you did not create

Inbound website messages do not fit any canonical entity, so they are a table
you declare. `message` is what the agent reads; `category` is what a
production system writes back.

**Creating a generic table provisions a default Data Activation Client for
you**, named after the table, with `tool_call: manual_upload`. For plain row
ingestion into a table you declared there is nothing to build — no data
source, no tool, no contract, no client. That is the whole reason this
scenario has none of the machinery §011–§013 needed: those existed because
the appointment rows also had to become *patients*, and this table's rows are
their own subject.

**Scope the lookup server-side.** The listing pages and is not newest-first,
so a `.find()` over page one starts missing the client as soon as the lake
has a few tables — and the failure looks like it was never provisioned.

```typescript
const contactUs = await ctx.ensure(
  'contact-us table',
  async () => {
    const { data } = await api.genericTables.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return (data.data ?? []).find((t: { title?: string }) => t.title === 'Cookbook Contact Us')
  },
  async () => {
    const { data } = await api.genericTables.create(tenantSlug, datalakeSlug, {
      title: 'Cookbook Contact Us',
      description: 'Inbound contact-us form submissions the triage agent classifies.',
      columns: [
        { name: 'submission_id', title: 'Submission ID', type: 'string', description: 'Vendor-supplied unique submission id', is_unique: true, privacy_requirement: 'none' },
        { name: 'name', title: 'Name', type: 'string', description: 'Submitter full name', is_unique: false, privacy_requirement: 'tokenize' },
        { name: 'email', title: 'Email', type: 'string', description: 'Submitter email', is_unique: false, privacy_requirement: 'tokenize' },
        { name: 'message', title: 'Message', type: 'string', description: 'Free-text message — the agent classifies this', is_unique: false, privacy_requirement: 'none' },
        { name: 'category', title: 'Category', type: 'string', description: 'Triage bucket assigned by the agent', is_unique: false, privacy_requirement: 'none' },
      ],
    })
    return data
  },
)
ctx.contactTableId = contactUs.id!
ctx.contactTableName = contactUs.name!

await ctx.deployGenericTable(ctx.contactTableName)

const dacDeadline = Date.now() + 120_000
let defaultDac: { slug?: string | null } | undefined
while (Date.now() < dacDeadline && !defaultDac) {
  const { data } = await api.dataActivationClients.list(tenantSlug, datalakeSlug, {
    filters: [{ field: 'name', op: 'ilike', value: ctx.contactTableName }],
  })
  defaultDac = (data.data ?? [])[0]
  if (!defaultDac) await new Promise((r) => setTimeout(r, 1_000))
}
if (!defaultDac?.slug) {
  throw new Error(`no default client found for table ${ctx.contactTableName} within 120s`)
}
ctx.contactDacSlug = defaultDac.slug
```

## 021 — the LLM tool and the triage agent

The tool is a **provider adapter**: `base_body` authors Ollama's native
`/api/chat` request with `think: false` and a `format` schema so the model
returns schema-constrained JSON, and `response_extractor` maps the reply back
onto the canonical `{ output_json, … }`.

The agent's `llm_response_schema` pins the vocabulary with an `enum`, and
that is load-bearing rather than decorative: the workflow interpolates the
category straight into its decision array, so a model answering
`"a clinical question"` instead of `clinical` would produce a decision key no
action is registered for and the row would silently do nothing.

**`data_access: 'tokenized'` is the decision this whole surface exists for.**
The agent reads inbound messages to decide how urgent they are. It does not
need to know who sent them, and on this lake it cannot: the tokenized
projection is what it sees. Compare `subscription-saas`, where the equivalent
agent runs `raw` — there the prompt carries an account number and an account
type and no identity at all, so there is nothing to hold back. The tier
follows what the prompt can reach, not a house style.

```typescript
const ENRICHMENT_OUTPUT_SCHEMA = {
  type: 'object',
  properties: {
    output_json: {},
    input_tokens: { type: ['integer', 'null'] },
    output_tokens: { type: ['integer', 'null'] },
    total_tokens: { type: ['integer', 'null'] },
    explanation: { type: ['string', 'null'] },
  },
  required: ['output_json'],
}

const OLLAMA_BASE_BODY =
  '{"model": "{{ model }}", "messages": [{"role": "user", "content": "{{ rendered_prompt | json_escape }}", ' +
  '"images": [{% for img in images %}{% unless forloop.first %}, {% endunless %}"{{ img.data }}"{% endfor %}]}], ' +
  '"stream": false, "think": false, "options": {"temperature": {{ temperature }}, "num_predict": {{ max_tokens }}, "num_ctx": 40960}, "format": {{ schema | to_json }}}'

const OLLAMA_EXTRACTOR =
  '{"output_json": "{{ msg.message.content | json_escape }}", ' +
  '"explanation": "{{ msg.message.thinking | json_escape }}", ' +
  '"input_tokens": {{ msg.prompt_eval_count | default: 0 }}, ' +
  '"output_tokens": {{ msg.eval_count | default: 0 }}, ' +
  '"total_tokens": {{ msg.prompt_eval_count | default: 0 | plus: msg.eval_count }}}'

const llmTool = await ctx.ensure(
  'LLM tool',
  async () => {
    const { data } = await api.tools.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook Practice LLM Tool')
  },
  async () => {
    const { data } = await api.tools.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook Practice LLM Tool',
      description: 'Ollama-backed chat-completion adapter for contact-us triage.',
      intent: 'llm_enrichment',
      status: 'active',
      datalake_id: ctx.datalakeId,
      response_extractor: { type: 'custom', body: OLLAMA_EXTRACTOR, output_schema: ENRICHMENT_OUTPUT_SCHEMA },
      body: {
        tool_body_type: 'rest_api',
        base_url: 'http://localhost:11434',
        base_path: { type: 'custom', body: '/api/chat' },
        auth_method: 'api_key',
        api_key: 'stub-key',
        api_key_name: 'Authorization',
        api_key_location: 'header',
        request_type: 'json',
        response_type: 'json',
        timeout_ms: 60_000,
        base_body: { type: 'custom', body: OLLAMA_BASE_BODY },
      },
    })
    return data
  },
)
ctx.llmToolId = llmTool.id!

const TRIAGE_PROMPT = `You are a triage assistant for a primary-care practice website. Read the inbound message and classify it into EXACTLY ONE category:

- "clinical"      — a symptom, a medication question, anything a clinician must see
- "billing"       — an invoice, a statement, insurance or payment
- "appointment"   — booking, rescheduling or cancelling a visit
- "spam"          — marketing, solicitation, or nothing to do with the practice

Message: {{ message }}

Respond with JSON: {"category": "<one of clinical|billing|appointment|spam>"}`

const triageAgent = await ctx.ensure(
  'Contact Us triage agent',
  async () => {
    const { data } = await api.aiAgents.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook Contact Us Triage Agent')
  },
  async () => {
    const { data } = await api.aiAgents.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook Contact Us Triage Agent',
      tool_id: ctx.llmToolId,
      model: 'qwen3-vl:8b-instruct',
      data_access: 'tokenized',
      temperature: 0.0,
      max_tokens: 1024,
      enabled: true,
      input_schema: {
        type: 'object',
        properties: { message: { type: 'string' } },
        required: ['message'],
      },
      llm_response_schema: {
        type: 'object',
        properties: {
          category: { type: 'string', enum: ['clinical', 'billing', 'appointment', 'spam'] },
        },
        required: ['category'],
      },
      prompt_config: { type: 'custom', body: TRIAGE_PROMPT },
    })
    return data
  },
)
aiAgentId = triageAgent.id!
ctx.triageAgentSlug = triageAgent.slug!
```

## 022 — the triage workflow, and three submissions to run it on

The agent is **nested in the create body** — `workflow_ai_agents` is a
`cast_assoc`, not a separate attach call — with a `context_mapping_config`
that projects each row into the agent's input schema. **Bracket access is
required, not stylistic**: the agent's slug contains hyphens and Liquid's dot
parser would read `additional_context.cookbook-contact-us-triage-agent` as a
subtraction.

The three submissions are deliberately unambiguous — chest pain, an invoice
query, and obvious solicitation — so the classification is not a coin toss
and the assertion in §023 means something. `category` is left unset on
ingest; assigning it is the agent's job.

`link_prefix` is required on every SMS call, these routing alerts
included, though their bodies carry no URL — a body with none is sent
unchanged. Leave it off and the workflow is refused at create with a 422 on
`/actions/<n>/tool_call/link_prefix`.

```typescript
const CATEGORIES = ['clinical', 'billing', 'appointment', 'spam'] as const

const triageWorkflow = await ctx.ensure(
  'Contact Us triage workflow',
  async () => {
    const { data } = await api.workflows.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook Contact Us Triage Workflow')
  },
  async () => {
    const { data } = await api.workflows.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook Contact Us Triage Workflow',
      description: 'Classifies inbound messages into four buckets; one SMS action per bucket.',
      dataset_type: 'generic_table',
      generic_table_id: ctx.contactTableId,
      status: 'live',
      tags: ['triage', 'llm'],
      skip_mdm_resolution: true,
      filter_config: { type: 'custom', body: 'true' },
      decision_config: {
        type: 'custom',
        body: `["{{ additional_context["${ctx.triageAgentSlug}"].category }}"]`,
        output_schema: { type: 'array', items: { type: 'string' } },
      },
      actions: CATEGORIES.map((category) => ({
        decision_key: category,
        action_type: 'sms' as const,
        tool_id: ctx.smsToolId,
        position: 0,
        trigger_template: 'now',
        idempotency_template: `{{ event_dataset.submission_id }}-${category}`,
        tool_call: {
          tool_call_type: 'sms_request' as const,
          to: { type: 'custom' as const, body: '+15555550100' },
          body: {
            type: 'custom' as const,
            body: `[${category}] contact-us submission {{ event_dataset.submission_id }} needs routing.`,
          },
          sms_type: 'transactional' as const,
          link_prefix: 'https://go.triage.example.local/l',
        },
      })),
      workflow_ai_agents: [
        {
          ai_agent_id: aiAgentId,
          position: 0,
          context_mapping_config: {
            type: 'custom',
            body: JSON.stringify({ message: '{{ event_dataset.message }}' }),
          },
        },
      ],
    })
    return data
  },
)
ctx.triageWorkflowSlug = triageWorkflow.slug!

ctx.clinicalId = 'SUB-COOKBOOK-0001'
ctx.billingId = 'SUB-COOKBOOK-0002'
ctx.spamId = 'SUB-COOKBOOK-0003'

const submissions = [
  {
    submission_id: ctx.clinicalId,
    name: 'Rosalind Achterberg',
    email: 'rosalind.achterberg@example.com',
    message:
      'I have had chest tightness and shortness of breath since yesterday evening, ' +
      'and it gets worse when I walk up stairs. Should I come in?',
  },
  {
    submission_id: ctx.billingId,
    name: 'Teodoro Vasquez-Lindqvist',
    email: 'teodoro.vl@example.com',
    message:
      'I received an invoice for my visit last month but my insurance should have ' +
      'covered it. Can someone check the claim and re-send the statement?',
  },
  {
    submission_id: ctx.spamId,
    name: 'Growth Partners',
    email: 'outreach@growth-partners.example',
    message:
      'Boost your clinic revenue 300% with our SEO package! Limited slots. ' +
      'Reply STOP to opt out. Click here for a free audit.',
  },
]

const ingests = await Promise.all(
  submissions.map((row) =>
    api.dataActivationClients.ingest(tenantSlug, datalakeSlug, ctx.contactDacSlug, { data: row }),
  ),
)
await ctx.waitForBatches(
  ctx.contactDacSlug,
  ingests.map((r: { data: { batch_id?: string } }) => r.data.batch_id!),
)
```

## 023 — the agent's category steered the fan-out

Each submission produces an execution log carrying **four** action logs, one
per category. On the first run for a given submission exactly **one** is
matched (`:pending` or `:completed` — the category the agent chose) and
**three** are `:skipped`.

**On a later run all four are `:skipped`, and that is correct.** The
idempotency key is `<submission id>-<category>` and the submission ids are
stable, so a key that already fired refuses to fire again. Both shapes are
accepted; anything else is a real failure — two matched means the decision
matched more than one category, zero matched with zero skipped means the
fan-out never happened.

```typescript
const runResp = await api.workflows.run(tenantSlug, datalakeSlug, ctx.triageWorkflowSlug, {
  sql_where_clause: `submission_id IN ('${ctx.clinicalId}', '${ctx.billingId}', '${ctx.spamId}')`,
  mode: 'live',
  manual_override: false,
})
const fired = await ctx.waitForFiredRun(datalakeSlug, runResp.data.workflow_run_id)
ctx.triageRunBatchId = fired.batchId!

const refreshDeadline = Date.now() + 180_000
let status: string | null = null
while (Date.now() < refreshDeadline) {
  const { data: log } = await api.workflows.batchLogs.refresh(
    tenantSlug, datalakeSlug, ctx.triageWorkflowSlug, fired.workflowRunLogId,
  )
  status = log.status ?? null
  if (status && status !== 'pending') break
  await new Promise((r) => setTimeout(r, 2_000))
}
if (status === 'failed') throw new Error('the triage run reached :failed')
if (!status || status === 'pending') throw new Error('the triage run did not leave :pending within 180s')

const welDeadline = Date.now() + 180_000
let ourLogs: Array<Record<string, unknown>> = []
while (Date.now() < welDeadline) {
  const { data: wfLogs } = await api.workflows.workflowLogs.list(
    tenantSlug, datalakeSlug, ctx.triageWorkflowSlug, { page_size: 100 },
  )
  ourLogs = (wfLogs.data ?? [])
    .map((w) => w as Record<string, unknown>)
    .filter((w) => w.batch_id === ctx.triageRunBatchId)
  const settled = ourLogs.filter(
    (w) => ((w.action_execution_logs as unknown[] | undefined) ?? []).length === 4,
  )
  if (ourLogs.length === 3 && settled.length === 3) break
  await new Promise((r) => setTimeout(r, 2_000))
}
if (ourLogs.length !== 3) {
  throw new Error(`expected 3 execution logs for the triage run, got ${ourLogs.length}`)
}

for (const wel of ourLogs) {
  const welId = (wel as { id?: string }).id
  const aels = (wel as { action_execution_logs?: Array<{ status?: string }> }).action_execution_logs ?? []
  if (aels.length !== 4) {
    throw new Error(`log ${welId}: expected 4 action logs (one per category), got ${aels.length}`)
  }
  const byStatus: Record<string, number> = {}
  for (const ael of aels) {
    const st = ael.status ?? 'unknown'
    byStatus[st] = (byStatus[st] ?? 0) + 1
  }
  const matched = (byStatus.pending ?? 0) + (byStatus.completed ?? 0)
  const skipped = byStatus.skipped ?? 0
  if (matched === 1 && skipped === 3) continue   // first run for this submission
  if (matched === 0 && skipped === 4) continue   // already fired, correctly refusing
  throw new Error(
    `log ${welId}: expected 1 matched + 3 skipped, or 4 skipped on a repeat run — ${JSON.stringify(byStatus)}`,
  )
}
```

## 024 — the chest-pain message was routed clinical

§023 proved the fan-out chose one action per row. This proves *which* one,
and it is the assertion that survives every rerun because it reads the
durable message rather than the run that made it.

The chest-pain submission is the one asserted, because it is the only one
whose category is genuinely beyond argument — tightness, breathlessness and
exertional worsening is clinical to any classifier worth running, and a model
that files it as billing is a model to find out about immediately.

The key is `<submission id>-clinical`, so **finding the message at all is the
category assertion**. There is no need to parse the bucket back out of the
body.

```typescript
const { data: search } = await api.datasets.createUserSearch(tenantSlug, datalakeSlug, 'message', {
  search_query: `m.idempotency_key = '${ctx.clinicalId}-clinical'`,
})
if (search.status !== 'completed') {
  throw new Error(`message user-search status=${search.status} error=${search.error_message ?? '(none)'}`)
}

const msgDeadline = Date.now() + 90_000
let clinicalBody: string | undefined
while (Date.now() < msgDeadline && !clinicalBody) {
  const { data } = await api.datasets.search(tenantSlug, datalakeSlug, 'message', {
    userSearchId: search.id!,
    dataAccessMode: 'raw',
  })
  clinicalBody = ((data.data ?? []) as Array<Record<string, unknown>>)
    .map((m) => String(m.body ?? ''))
    .find((b) => b.includes('[clinical]'))
  if (!clinicalBody) await new Promise((r) => setTimeout(r, 2_000))
}
if (!clinicalBody) {
  throw new Error(
    `no clinical routing message for ${ctx.clinicalId} within 90s — ` +
      'the agent filed the chest-pain message somewhere else, or the action never fired',
  )
}
if (!clinicalBody.includes(ctx.clinicalId)) {
  throw new Error(`the routing message does not name the submission: ${clinicalBody}`)
}
```

## 025 — upload a scanned document

The practice's back office receives documents as pixels — an employer group's
registration papers, an insurance letter, a referral — and someone keys the
same handful of fields out of each one. This scenario hands that to a model.

Mint a presigned URL per file and PUT the bytes with a **raw `fetch`**. That
upload goes straight to object storage and is deliberately not an SDK call:
the SDK talks to the platform, and the platform is not in this hop. Keep each
returned `key` — that is how a file is handed to an agent.

```typescript
const { readFileSync } = await import('node:fs')
const { join } = await import('node:path')

const uploadFixture = async (filename: string, contentType: string): Promise<string> => {
  const bytes = readFileSync(
    join(process.env.COOKBOOK_FIXTURES_DIR!, 'primary-care-feedback', filename),
  )
  const { data: link } = await api.datalakes.createUploadLink(tenantSlug, datalakeSlug, {
    filename,
    content_type: contentType,
  })
  const put = await fetch(link.url!, {
    method: 'PUT',
    headers: { 'Content-Type': contentType },
    body: bytes,
  })
  // Read the body before deciding. S3 puts the REASON there and nowhere else:
  // `SignatureDoesNotMatch` and `AccessDenied` are different problems with
  // different fixes, and a bare `403` tells you neither. A presigned PUT is
  // also the one call in this walk that never touches the platform — it goes
  // straight to object storage — so the platform's logs will not have it
  // either.
  if (put.status !== 200) {
    const why = await put.text().catch(() => '(no body)')
    throw new Error(
      `upload of ${filename} failed: ${put.status} ${put.statusText}\n` +
        `  endpoint: ${new URL(link.url!).origin}\n` +
        `  key:      ${link.key}\n` +
        `  body:     ${why.slice(0, 800)}`,
    )
  }
  return link.key!
}

ctx.scanKey = await uploadFixture('memorandum-of-association-01.png', 'image/png')
```

## 026 — the vision agent

Same tool as §021 — one provider adapter serves both agents, because the
adapter is about the provider's wire shape and not about the task. Its
`base_body` already appends `images` to the message, so a multimodal call
needs no second tool.

The agent differs in three ways that matter. Its `input_schema` takes
`instructions` rather than a message, because the caller steers this one per
invocation. Its `llm_response_schema` describes the fields to pull out of the
document, and it is load-bearing: the platform constrains the model to it, so
a document with no company name comes back with `company_name: null` rather
than with the model's apology. And `additionalProperties: false` means a
model that volunteers a field nobody asked for is refused rather than
silently widening the record.

`data_access: 'tokenized'` again. It changes nothing about *this* call — the
input is an uploaded file, not a lake row — and it is set anyway, because the
tier is a property of the agent rather than of the invocation, and an agent
that is later pointed at a workflow inherits whatever it was created with. A
tokenized agent cannot be made raw by the caller; that is the point of
putting the tier on the resource.

```typescript
const extractionAgent = await ctx.ensure(
  'document extraction agent',
  async () => {
    const { data } = await api.aiAgents.list(tenantSlug, datalakeSlug, { page_size: 100 })
    return ctx.firstNamed(data.data, 'Cookbook Document Extraction Agent')
  },
  async () => {
    const { data } = await api.aiAgents.create(tenantSlug, datalakeSlug, {
      name: 'Cookbook Document Extraction Agent',
      description: 'Reads scanned back-office documents and returns structured fields.',
      tool_id: ctx.llmToolId,
      model: 'qwen3-vl:8b-instruct',
      data_access: 'tokenized',
      temperature: 0.0,
      max_tokens: 2048,
      enabled: true,
      input_schema: {
        type: 'object',
        properties: { instructions: { type: 'string' } },
        required: ['instructions'],
      },
      llm_response_schema: {
        type: 'object',
        properties: {
          company_name: { type: ['string', 'null'] },
          registered_address: { type: ['string', 'null'] },
          date_of_formation: { type: ['string', 'null'] },
        },
        required: ['company_name'],
        additionalProperties: false,
      },
      prompt_config: {
        type: 'custom',
        body:
          'You are a document data-extraction specialist. Read the attached ' +
          'document image and return the requested fields as JSON. Use null for ' +
          'anything the document does not state.\n\n{{ instructions }}',
      },
    })
    return data
  },
)
ctx.extractionAgentId = extractionAgent.id!
```

## 027 — invoke it against the scan

`invoke` runs the agent against the uploaded files: `input` satisfies the
`input_schema`, and `files` lists the storage keys plus their content types.
What comes back is the parsed JSON matching `llm_response_schema`, an
`explanation`, and token `usage`.

The assertion is deliberately weak about *content* and strict about *shape*.
A live model reading a scan will not return the same string twice, so
asserting the company's exact name would make this step a coin toss that
fails on a model upgrade rather than on a regression. What must hold is that
the call round-trips: structured output, conforming to the schema the agent
pins, with usage accounted for. That is the contract the platform owns; the
transcription accuracy is the model's.

```typescript
const { data: result } = await api.aiAgents.invoke(tenantSlug, datalakeSlug, ctx.extractionAgentId, {
  input: {
    instructions:
      'Extract the company name, its registered address, and the date of formation ' +
      'from the attached document.',
  },
  files: [{ key: ctx.scanKey, content_type: 'image/png' }],
})

if (!result.output || typeof result.output !== 'object') {
  throw new Error('invoke did not return a structured output object')
}
const output = result.output as Record<string, unknown>
if (!('company_name' in output)) {
  throw new Error(
    `the response does not carry the required company_name key: ${JSON.stringify(output)}`,
  )
}
// additionalProperties: false — nothing outside the pinned schema may appear.
const allowed = new Set(['company_name', 'registered_address', 'date_of_formation'])
const extra = Object.keys(output).filter((k) => !allowed.has(k))
if (extra.length > 0) {
  throw new Error(`the model returned fields outside llm_response_schema: ${extra.join(', ')}`)
}
if (!result.usage || (result.usage as { input_tokens?: number }).input_tokens === undefined) {
  throw new Error('expected token usage on the invoke result')
}
console.log('  extracted →', JSON.stringify(output))
```

## 028 — a malformed input is refused before the model is called

`invoke` validates `input` against the agent's `input_schema` **before**
calling the provider. Omitting the required `instructions` returns a 422, and
no tokens are spent.

That ordering is the point. The schema is not documentation about what the
caller ought to send — it is a gate, and it fails closed at the cheapest
possible moment. An agent whose input gate ran after the model call would
turn every caller's typo into a billed request.

The refusal is caught so the walk proves the gate without aborting, and the
status is checked rather than the message: a 500 here would also throw, and
treating it as a pass would be exactly the kind of test that proves nothing.

```typescript
let rejected = false
try {
  await api.aiAgents.invoke(tenantSlug, datalakeSlug, ctx.extractionAgentId, {
    input: { wrong_field: 'no instructions here' },
    files: [],
  })
} catch (err) {
  const status = (err as { _httpStatus?: number })._httpStatus
  if (status !== 422) throw err
  rejected = true
}
if (!rejected) {
  throw new Error('expected a 422 for input that violates the agent input_schema')
}
```

## 029 — one patient, three lakes, three answers

The last step is the one this whole surface exists for. §016 proved a masked
*array* survives the copy; this proves what masking does to the thing anyone
would actually recognise — a person's name — and it shows all three answers
side by side.

`mode` is a parameter of the read, not a different slug. One lake is
addressed and `mode` selects which copy answers, which works only because
§004 built its client at the `raw` ceiling. A tokenized-ceiling session
asking for `raw` is refused rather than quietly downgraded, and that is how a
support console is kept honest: you give it a key it cannot exceed instead of
trusting every screen to ask nicely.

Read the three and compare. The **name must differ** in both copies — if it
does not, masking is not in force and the agents above have been reading
identities all along. The **appointment id must be identical** in all three —
it is a `none` column, and a masking policy that changed it would have broken
every join in the system while looking like it was working.

The two copies are also different from *each other*, and that is the design
rather than an accident: the tokenized copy carries a stable
`tok_<hash>` — a machine can join on it across tables without ever resolving
it — while the redacted copy carries `xxxx-<tail>`, which is unjoinable and
meant for a human to glance at. One is for a model, the other is for a
person.

```typescript
const identitySql =
  `SELECT appointment_id, patient_name, department ` +
  `FROM ${ctx.appointmentsTableName} WHERE appointment_id = '${ctx.fulfilledId}'`

const readIdentityAs = async (mode: 'raw' | 'tokenized' | 'redacted') => {
  const { data } = await api.datalakes.executeSql(tenantSlug, datalakeSlug, {
    sql: identitySql,
    mode,
  })
  const row = (data.data ?? [])[0]
  if (!row) throw new Error(`the appointment was not readable through the ${mode} lake`)
  const cell = (column: string): unknown => row[data.meta.columns.indexOf(column)]
  return {
    id: String(cell('appointment_id')),
    name: String(cell('patient_name')),
    department: String(cell('department')),
  }
}

const rawRead = await readIdentityAs('raw')
const tokenizedRead = await readIdentityAs('tokenized')
const redactedRead = await readIdentityAs('redacted')

console.log('  raw       →', JSON.stringify(rawRead))
console.log('  tokenized →', JSON.stringify(tokenizedRead))
console.log('  redacted  →', JSON.stringify(redactedRead))

if (rawRead.name !== 'Rosalind Achterberg') {
  throw new Error(`the raw lake should hold the real name, got ${JSON.stringify(rawRead.name)}`)
}
for (const [mode, read] of [['tokenized', tokenizedRead], ['redacted', redactedRead]] as const) {
  if (read.name === rawRead.name) {
    throw new Error(`the ${mode} lake handed back the real patient name — masking is not in force`)
  }
  if (read.id !== rawRead.id || read.department !== rawRead.department) {
    throw new Error(
      `the ${mode} lake changed a 'none' column: id ${read.id} vs ${rawRead.id}, ` +
        `department ${read.department} vs ${rawRead.department}`,
    )
  }
}
// The two copies serve different readers, so they must not be the same value.
if (tokenizedRead.name === redactedRead.name) {
  throw new Error(
    'the tokenized and redacted lakes returned the same masked value — ' +
      'one is meant to be a stable join key and the other a human-readable placeholder',
  )
}
```

## 030 — how far the copy actually got, table by table

Every read above trusted the copy. This step asks it directly.

`derivedLakeTables` answers per table, and the addressing is the first thing
to get right: you name the **primary** lake plus a mode. A derived lake does
have a slug of its own, but that slug is not an address — the lookup resolves
primaries only, and the walk proves it by asking with the tokenized lake's own
slug and catching the refusal.

The state to read is `caught_up`, never the state's *name*. One of the five
states reads as finished without being finished: `finished_copy` means the
first pass is done but the changes made *during* that pass are not applied
yet. A surface that treats it as ready shows a lake as complete while rows
are still missing. The boolean exists so nobody has to pattern-match the word.

`resync_empties` carries a rule worth asserting rather than assuming: it is
`null` for a table that is already caught up — the platform does not spend a
query per table answering a question about tables nobody is going to repair —
and an array otherwise. `null` and `[]` mean different things, and `[]` never
occurs, because a table is always in its own set.

```typescript
const { data: pair } = await api.datalakes.provisionTokenization(tenantSlug, datalakeSlug)
const tokenizedLake = (pair.derived_datalakes ?? []).find((d) => d.type === 'tokenized')
if (!tokenizedLake) throw new Error('no tokenized copy on the lake — §007 should have provisioned it')

const { data: tables } = await api.datalakes.derivedLakeTables(
  tenantSlug, datalakeSlug, 'tokenized',
)
if (tables.data.length === 0) {
  throw new Error('the tokenized copy reported no tables at all')
}
console.log(`  ${tables.data.length} tables in the tokenized copy`)

for (const t of tables.data) {
  if (typeof t.caught_up !== 'boolean') {
    throw new Error(`${t.table}: caught_up is not a boolean, got ${JSON.stringify(t.caught_up)}`)
  }
  if (!t.label || t.label.length === 0) {
    throw new Error(`${t.table}: label is empty — it is what a screen renders`)
  }
  // The agreement that makes the boolean trustworthy: `caught_up` is true for
  // exactly one of the five states. `finished_copy` reading true here would be
  // the defect the boolean exists to prevent. This cannot be asserted against a
  // mock — the schema cannot express a cross-field constraint, so Prism draws
  // the two independently — which is exactly why it is asserted here.
  if (t.caught_up !== (t.state === 'caught_up')) {
    throw new Error(
      `${t.table}: state '${t.state}' disagrees with caught_up=${t.caught_up} — ` +
        `only 'caught_up' means arrived`,
    )
  }
  // The null-vs-array rule, asserted in both directions.
  if (t.caught_up && t.resync_empties !== null && t.resync_empties !== undefined) {
    throw new Error(
      `${t.table} is caught up, so resync_empties should not be computed — got ` +
        JSON.stringify(t.resync_empties),
    )
  }
  if (!t.caught_up) {
    if (!Array.isArray(t.resync_empties) || t.resync_empties.length === 0) {
      throw new Error(`${t.table} is behind, so resync_empties should be a non-empty array`)
    }
    if (t.resync_empties[0] !== t.table) {
      throw new Error(`${t.table}: resync_empties should name the table itself first`)
    }
  }
}
ctx.copyTables = tables.data.map((t) => t.table)

// A derived lake's own slug is not an address. Prove the refusal rather than
// assuming it — a 200 here would mean two ways to name the same thing.
let refusedDerivedSlug = false
try {
  await api.datalakes.derivedLakeTables(tenantSlug, tokenizedLake.slug!, 'tokenized')
} catch (err) {
  const status = (err as { _httpStatus?: number })._httpStatus
  if (status !== 404) throw err
  refusedDerivedSlug = true
}
if (!refusedDerivedSlug) {
  throw new Error("the derived lake's own slug was accepted as an address — it should 404")
}
```

## 031 — empty one table and watch it come back

`resyncTable` is the repair: it empties a table in the copy and copies it
again from the start. Two things about it are easy to get wrong, and both are
asserted here rather than described.

**It empties more than the table you name.** Postgres refuses to truncate any
table with an inbound foreign key — whether or not the referencing tables
hold rows — so repairing a parent takes its whole dependent closure with it,
transitively, in one statement. All of them are re-copied by the same run.
`legal_entities` is chosen deliberately: patient identities were resolved
through it in §008, and `legal_entity_identifications` points at it, so the
`emptied` list here is genuinely longer than one. A surface offering this
action should read `resync_empties` and say the number before the button is
pressed.

**`202` means started, not done.** There is no completion callback. The walk
polls `derivedLakeTables` until the table reads `caught_up` again, which is
also the honest way to prove the repair worked rather than merely dispatched.

The table name is gated against what the copy carries, so a typo is a `422`
and **nothing is emptied**. That gate is the only thing standing between a
mistyped name and an emptied table, so it is proved here too — and it is
proved *first*, because a walk that emptied the wrong table and then checked
the gate would have the order backwards.

```typescript
const TARGET = 'legal_entities'
if (!(ctx.copyTables as string[]).includes(TARGET)) {
  throw new Error(`${TARGET} is not in the copy; it carries: ${(ctx.copyTables as string[]).join(', ')}`)
}

// The gate first — nothing should be emptied by a name the copy does not carry.
let refusedBadTable = false
try {
  await api.datalakes.resyncTable(tenantSlug, datalakeSlug, 'tokenized', {
    table: 'no_such_table_here',
  })
} catch (err) {
  const status = (err as { _httpStatus?: number })._httpStatus
  if (status !== 422) throw err
  refusedBadTable = true
}
if (!refusedBadTable) throw new Error('an unknown table name was accepted — the gate is not closed')

const { data: repair } = await api.datalakes.resyncTable(
  tenantSlug, datalakeSlug, 'tokenized', { table: TARGET },
)
if (repair.table !== TARGET) throw new Error(`repaired ${repair.table}, asked for ${TARGET}`)
if (repair.mode !== 'tokenized') throw new Error(`repaired the ${repair.mode} copy, asked for tokenized`)
if (repair.status !== 'copying') throw new Error(`expected status 'copying', got ${repair.status}`)
if (!repair.emptied.includes(TARGET)) {
  throw new Error(`emptied ${JSON.stringify(repair.emptied)} — the named table is not among them`)
}
console.log(`  emptied ${repair.emptied.length} table(s): ${repair.emptied.join(', ')}`)

// The dependent closure is the point: identifications point at legal entities,
// so a repair of the parent cannot leave the child behind.
if (repair.emptied.length < 2) {
  throw new Error(
    `${TARGET} has dependents, so a repair should empty more than itself — got ` +
      JSON.stringify(repair.emptied),
  )
}

// 202 = started. Wait for the copy to arrive again.
const RESYNC_TIMEOUT_MS = 5 * 60_000
const resyncDeadline = Date.now() + RESYNC_TIMEOUT_MS
let backOnline = false
while (Date.now() < resyncDeadline && !backOnline) {
  const { data: after } = await api.datalakes.derivedLakeTables(
    tenantSlug, datalakeSlug, 'tokenized',
  )
  backOnline = (repair.emptied as string[]).every(
    (name) => after.data.find((t) => t.table === name)?.caught_up === true,
  )
  if (!backOnline) await new Promise((r) => setTimeout(r, 5_000))
}
if (!backOnline) {
  throw new Error(`the repaired tables did not come back within ${RESYNC_TIMEOUT_MS / 1000}s`)
}
console.log('  every emptied table is caught up again')
```

# Outcome

After all thirty-one steps run green, on one tenant and **three datalakes**:

- **A raw lake and its tokenized and redacted copies**, provisioned as a pair
  and each carrying every table the primary has.
- **Two generic tables** — appointments and contact-us submissions — one
  ingested through a contract pair on two sequenced clients, the other
  through the default client the platform provisioned when the table was
  declared.
- **Two workflows**: a review request that reaches only the patient whose
  visit was fulfilled, and a four-way category fan-out driven by an LLM
  agent reading the tokenized projection.
- **Two agents**, one classifying text and one reading a scanned document,
  both pinned to `data_access: 'tokenized'` and both constrained by an
  `llm_response_schema` the platform enforces.
- **The same row, read three ways**, showing what each reader sees.
- **The tokenized copy inspected and repaired** — every table's arrival state
  read back, one table emptied and copied again, and every table the repair
  took with it waited back to `caught_up`.

# Gotchas

1. **A masked jsonb column is `jsonb` in raw and `text` in both copies, and
   the types are supposed to differ.** Logical replication ships a text
   rendering and feeds it to the subscriber's input function, so
   `jsonb → text` always works and `jsonb → text[]` never does. The
   invariant is scalar-ness, not type parity — and `is_array: true`
   describes the JSON document, not the Postgres column — §016.
2. **The tier follows what the prompt can reach.** The triage agent here runs
   `tokenized` because it reads patient messages; the equivalent agent in
   `subscription-saas` runs `raw` because its prompt carries an account
   number and nothing else. Neither is a house style — §021.
3. **The two copies are different from each other on purpose.** `tok_<hash>`
   is a stable join key for a machine; `xxxx-<tail>` is an unjoinable
   placeholder for a person — §029.
4. **A key the contract does not read is not an error.** It renders nothing,
   the column lands `null`, the batch reports green, and the failure appears
   two steps later as an action rejecting an empty recipient. Assert columns
   on the row — §015.
5. **"The copy exists" and "the copy is complete" are different questions.**
   `finished_copy` has not finished — read `caught_up`, not the state's name.
   And a repair is not scoped to the table you name: Postgres will not
   truncate a table anything references, so the dependent closure goes with
   it. Both are why `emptied` is returned rather than assumed — §030, §031.

# See also

- `payments-compliance.md` — the other tokenized surface, and the one that
  shows a masked read against an agent's verdict
- `subscription-saas.md` — the same contract-pair discipline on a raw lake
- `../datalakes.md` — the tier split, and what each copy is for
- `../ai_agents.md` — `data_access`, `input_schema`, `llm_response_schema`