diff --git a/CHANGELOG.md b/CHANGELOG.md
index ca38dd6..5490b89 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,36 @@ All notable changes to `pdcli` are documented here. Format follows
 [Keep a Changelog](https://keepachangelog.com/); versions follow
 [SemVer](https://semver.org/).
 
+## [0.16.0] - 2026-06-11
+
+### Added
+
+- Idempotent writes — match-or-create that never guesses:
+  - `person upsert`, `org upsert`, `deal upsert` — match a record by `--by`
+    (a built-in key — person email/name/phone, org name, deal title — or a
+    searchable custom field), then **create** it if absent or **PATCH only the
+    changed fields** if exactly one matches. More than one match **refuses with
+    exit 65** rather than writing the wrong record: search `exact_match` is not
+    a unique key, so every candidate is re-verified client-side before the
+    count decides. `--dry-run` previews the action; table prints a one-line
+    summary, `--output json` emits the full action result.
+  - `person import --upsert --match-on <field>` and the same on `org import` —
+    the CSV equivalent: each row is matched on its `--match-on` value, then
+    created or PATCHed. Per-row failures (ambiguous matches, empty match
+    values) are collected without aborting the batch and reported as
+    `created / updated / unchanged` counts; a batch whose failures are all
+    data-validation errors exits 65, otherwise 1. `--dry-run` looks up and
+    reports the counts without writing.
+
+### Changed
+
+- `diffBody` (upsert) compares emails/phones by their value set
+  (case-insensitive, ignoring the `primary`/`label` flags the API echoes) and
+  treats `label_ids` and multi-option custom fields order-insensitively, so
+  re-running an unchanged upsert issues no PATCH.
+- CSV import now rejects duplicate column headers (exit 65) instead of silently
+  keeping only the last cell.
+
 ## [0.15.0] - 2026-06-11
 
 ### Added
diff --git a/README.md b/README.md
index 83c4b6c..c16e001 100644
--- a/README.md
+++ b/README.md
@@ -87,6 +87,18 @@ pdcli person import people.csv --dry-run               # CSV headers map to fiel
 pdcli person import people.csv                         # custom fields by name
 ```
 
+## Idempotent writes
+
+Match-or-create, safe to re-run. More than one match **refuses** (exit 65) —
+never guesses which record to write.
+
+```bash
+pdcli person upsert a@x.com --by email --field "Tier=Gold"   # create or PATCH only what changed
+pdcli deal upsert "Acme expansion" --by title --body '{"value":5000}'
+pdcli org upsert "D-42" --by "External ID" --field "Status=Active" --dry-run  # preview
+pdcli person import contacts.csv --upsert --match-on email   # CSV: per-row create-or-update
+```
+
 ## Analytics & housekeeping
 
 ```bash
diff --git a/docs/commands.md b/docs/commands.md
index ce239c2..7d533f2 100644
--- a/docs/commands.md
+++ b/docs/commands.md
@@ -5,7 +5,7 @@ description: Full command reference for the pdcli command-line interface.
 
 <!-- AUTO-GENERATED from the oclif manifest by scripts/gen-commands.mjs — do not edit by hand. -->
 
-Reference for `pdcli` v0.15.0 (142 commands). Every command also accepts the global flags `--output table|json|yaml|csv`, `--profile`, `--no-color`, `--verbose`, `--no-retry`, `--timeout`, and `--limit`.
+Reference for `pdcli` v0.16.0 (145 commands). Every command also accepts the global flags `--output table|json|yaml|csv`, `--profile`, `--no-color`, `--verbose`, `--no-retry`, `--timeout`, and `--limit`.
 
 ## Top-level
 
@@ -958,6 +958,27 @@ pdcli deal update 42 --status won
 pdcli deal update 42 --field "Deal Size=Large"
 ```
 
+### `pdcli deal upsert`
+
+Idempotent deal upsert: match by --by, then create or PATCH only the changed fields. Refuses (exit 65) if more than one record matches.
+
+```
+pdcli deal upsert <value> [flags]
+```
+
+- `--by <value>` _(required)_ — Match field: title, or a searchable custom field
+- `--field <value>` — Field to set as "Name=Value" (repeatable)
+- `--body <value>` — Raw JSON body to merge
+- `--dry-run` — Preview the action without writing
+
+Examples:
+
+```bash
+pdcli deal upsert "Acme expansion" --by title --field "Stage=Won"
+pdcli deal upsert "D-42" --by "External ID" --body '{"value":5000}'
+pdcli deal upsert "Acme expansion" --by title --field "Stage=Won" --dry-run
+```
+
 ## pdcli field
 
 ### `pdcli field create`
@@ -1798,6 +1819,8 @@ Bulk-create organizations from a CSV (headers map to fields, custom fields by na
 pdcli org import <file> [flags]
 ```
 
+- `--upsert` — Match each row on --match-on, then create or update
+- `--match-on <value>` — Field to match rows on in --upsert mode (e.g. name)
 - `--dry-run` — Validate every row without creating anything
 - `-y, --yes` — Skip the confirmation prompt
 
@@ -1923,6 +1946,27 @@ pdcli org update 7 --owner 9
 pdcli org update 7 --field "Tier=Gold"
 ```
 
+### `pdcli org upsert`
+
+Idempotent organization upsert: match by --by, then create or PATCH only the changed fields. Refuses (exit 65) if more than one record matches.
+
+```
+pdcli org upsert <value> [flags]
+```
+
+- `--by <value>` _(required)_ — Match field: name, or a searchable custom field
+- `--field <value>` — Field to set as "Name=Value" (repeatable)
+- `--body <value>` — Raw JSON body to merge
+- `--dry-run` — Preview the action without writing
+
+Examples:
+
+```bash
+pdcli org upsert Acme --by name --field "Tier=Gold"
+pdcli org upsert "D-42" --by "External ID" --body '{"owner_id":42}'
+pdcli org upsert Acme --by name --field "Tier=Gold" --dry-run
+```
+
 ## pdcli person
 
 ### `pdcli person create`
@@ -2039,6 +2083,8 @@ Bulk-create persons from a CSV (headers map to fields, custom fields by name)
 pdcli person import <file> [flags]
 ```
 
+- `--upsert` — Match each row on --match-on, then create or update
+- `--match-on <value>` — Field to match rows on in --upsert mode (e.g. email)
 - `--dry-run` — Validate every row without creating anything
 - `-y, --yes` — Skip the confirmation prompt
 
@@ -2115,6 +2161,27 @@ pdcli person update 42 --email new@acme.com
 pdcli person update 42 --field "Segment=Enterprise"
 ```
 
+### `pdcli person upsert`
+
+Idempotent person upsert: match by --by, then create or PATCH only the changed fields. Refuses (exit 65) if more than one record matches.
+
+```
+pdcli person upsert <value> [flags]
+```
+
+- `--by <value>` _(required)_ — Match field: email, name, phone, or a searchable custom field
+- `--field <value>` — Field to set as "Name=Value" (repeatable)
+- `--body <value>` — Raw JSON body to merge
+- `--dry-run` — Preview the action without writing
+
+Examples:
+
+```bash
+pdcli person upsert a@x.com --by email --field "Tier=Gold"
+pdcli person upsert "Jane Doe" --by name --body '{"owner_id":42}'
+pdcli person upsert a@x.com --by email --field "Tier=Gold" --dry-run
+```
+
 ## pdcli pipeline
 
 ### `pdcli pipeline get`
diff --git a/package-lock.json b/package-lock.json
index 5096067..bcf3953 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1,12 +1,12 @@
 {
   "name": "@wavyx/pdcli",
-  "version": "0.15.0",
+  "version": "0.16.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@wavyx/pdcli",
-      "version": "0.15.0",
+      "version": "0.16.0",
       "license": "MIT",
       "dependencies": {
         "@inquirer/prompts": "8.5.0",
diff --git a/package.json b/package.json
index 564ce42..90054cb 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@wavyx/pdcli",
-  "version": "0.15.0",
+  "version": "0.16.0",
   "publishConfig": {
     "access": "public"
   },
diff --git a/src/commands/deal/upsert.js b/src/commands/deal/upsert.js
new file mode 100644
index 0000000..c56cf9f
--- /dev/null
+++ b/src/commands/deal/upsert.js
@@ -0,0 +1,54 @@
+import { Args, Flags } from '@oclif/core'
+import BaseCommand from '../../base-command.js'
+import { upsertWithDefs, summarizeUpsert } from '../../lib/upsert.js'
+
+export default class DealUpsertCommand extends BaseCommand {
+  static description =
+    'Idempotent deal upsert: match by --by, then create or PATCH only the ' +
+    'changed fields. Refuses (exit 65) if more than one record matches.'
+
+  static examples = [
+    '<%= config.bin %> deal upsert "Acme expansion" --by title --field "Stage=Won"',
+    '<%= config.bin %> deal upsert "D-42" --by "External ID" --body \'{"value":5000}\'',
+    '<%= config.bin %> deal upsert "Acme expansion" --by title --field "Stage=Won" --dry-run',
+  ]
+
+  static args = {
+    value: Args.string({ required: true, description: 'value to match on' }),
+  }
+
+  static flags = {
+    ...BaseCommand.baseFlags,
+    by: Flags.string({
+      required: true,
+      description: 'Match field: title, or a searchable custom field',
+    }),
+    field: Flags.string({
+      multiple: true,
+      description: 'Field to set as "Name=Value" (repeatable)',
+    }),
+    body: Flags.string({ description: 'Raw JSON body to merge' }),
+    'dry-run': Flags.boolean({
+      description: 'Preview the action without writing',
+    }),
+  }
+
+  async run() {
+    const { args, flags } = await this.parse(DealUpsertCommand)
+    const result = await upsertWithDefs({
+      client: this.apiClient,
+      entity: 'deal',
+      by: flags.by,
+      value: args.value,
+      fields: flags.field,
+      rawBody: flags.body,
+      dryRun: flags['dry-run'],
+    })
+
+    if (this.resolveFormat() !== 'table') {
+      await this.outputResults(result, {})
+      return
+    }
+    this.log(summarizeUpsert(result, 'deal'))
+  }
+}
diff --git a/src/commands/org/import.js b/src/commands/org/import.js
index 8ff24bb..e2c7402 100644
--- a/src/commands/org/import.js
+++ b/src/commands/org/import.js
@@ -6,6 +6,7 @@ import BaseCommand from '../../base-command.js'
 import { parseCsv } from '../../lib/csv-parse.js'
 import { prepareImportBodies } from '../../lib/import.js'
 import { bulkRun } from '../../lib/bulk.js'
+import { bulkUpsertRows } from '../../lib/upsert.js'
 import { getFields } from '../../lib/fields.js'
 import { confirmAction } from '../../lib/confirm.js'
 import { CliError } from '../../lib/errors.js'
@@ -34,6 +35,13 @@ export default class OrgImportCommand extends BaseCommand {
 
   static flags = {
     ...BaseCommand.baseFlags,
+    upsert: Flags.boolean({
+      description: 'Match each row on --match-on, then create or update',
+      default: false,
+    }),
+    'match-on': Flags.string({
+      description: 'Field to match rows on in --upsert mode (e.g. name)',
+    }),
     'dry-run': Flags.boolean({
       description: 'Validate every row without creating anything',
       default: false,
@@ -53,14 +61,39 @@ export default class OrgImportCommand extends BaseCommand {
       throw new CliError('CSV must include a "name" column', { exitCode: 64 })
     }
 
+    let matchIdx
+    if (flags.upsert) {
+      if (!flags['match-on']) {
+        throw new CliError('--upsert requires --match-on <field>', {
+          exitCode: 64,
+        })
+      }
+      matchIdx = headers.findIndex(
+        (h) => h.toLowerCase() === flags['match-on'].toLowerCase(),
+      )
+      if (matchIdx < 0) {
+        throw new CliError(
+          `--match-on "${flags['match-on']}" is not a column in ${args.file}`,
+          { exitCode: 64 },
+        )
+      }
+    }
+
     const needsDefs = headers.some((h) => !(h.toLowerCase() in SPECIAL_COLUMNS))
+    const defs =
+      needsDefs || flags.upsert ? await getFields(this.apiClient, 'org') : []
     const bodies = prepareImportBodies({
       headers,
       rows,
       specialColumns: SPECIAL_COLUMNS,
-      defs: needsDefs ? await getFields(this.apiClient, 'org') : [],
+      defs,
     })
 
+    if (flags.upsert) {
+      await this.upsertRows({ args, flags, rows, bodies, matchIdx, defs })
+      return
+    }
+
     if (flags['dry-run']) {
       this.log(chalk.green(`${bodies.length} rows valid — nothing created`))
       return
@@ -106,4 +139,59 @@ export default class OrgImportCommand extends BaseCommand {
       )
     }
   }
+
+  /** Idempotent CSV path: match each row on --match-on, then create or PATCH. */
+  async upsertRows({ args, flags, rows, bodies, matchIdx, defs }) {
+    const matchOn = flags['match-on']
+    const items = bodies.map((body, i) => ({ body, value: rows[i][matchIdx] }))
+
+    if (!flags['dry-run']) {
+      const ok = await confirmAction(
+        `Upsert ${items.length} organizations from ${args.file} (match on ${matchOn})?`,
+        flags.yes,
+      )
+      if (!ok) {
+        throw new CliError('Aborted', { exitCode: 1 })
+      }
+    }
+
+    const spinner = ora(`Upserting ${items.length} organizations...`).start()
+    let summary
+    try {
+      summary = await bulkUpsertRows({
+        client: this.apiClient,
+        entity: 'org',
+        matchOn,
+        rows: items,
+        defs,
+        dryRun: flags['dry-run'],
+        onProgress: (done, total) => {
+          spinner.text = `Upserting organizations ${done}/${total}`
+        },
+      })
+    } finally {
+      spinner.stop()
+    }
+
+    const { created, updated, unchanged } = summary.counts
+    const prefix = flags['dry-run'] ? '[dry-run] ' : ''
+    this.log(
+      chalk.green(
+        `${prefix}${created} created, ${updated} updated, ${unchanged} unchanged`,
+      ),
+    )
+
+    if (summary.failed.length > 0) {
+      for (const { item, error } of summary.failed) {
+        this.log(chalk.red(`  ✘ ${matchOn}="${item.value}": ${error}`))
+      }
+      // Surface 65 when every failure is a data-validation error (ambiguous
+      // match, empty match value); fall back to 1 for mixed/transport errors.
+      const allData = summary.failed.every((f) => f.exitCode === 65)
+      throw new CliError(
+        `${summary.failed.length} of ${items.length} rows failed`,
+        { exitCode: allData ? 65 : 1 },
+      )
+    }
+  }
 }
diff --git a/src/commands/org/upsert.js b/src/commands/org/upsert.js
new file mode 100644
index 0000000..97bf2f0
--- /dev/null
+++ b/src/commands/org/upsert.js
@@ -0,0 +1,54 @@
+import { Args, Flags } from '@oclif/core'
+import BaseCommand from '../../base-command.js'
+import { upsertWithDefs, summarizeUpsert } from '../../lib/upsert.js'
+
+export default class OrgUpsertCommand extends BaseCommand {
+  static description =
+    'Idempotent organization upsert: match by --by, then create or PATCH only ' +
+    'the changed fields. Refuses (exit 65) if more than one record matches.'
+
+  static examples = [
+    '<%= config.bin %> org upsert Acme --by name --field "Tier=Gold"',
+    '<%= config.bin %> org upsert "D-42" --by "External ID" --body \'{"owner_id":42}\'',
+    '<%= config.bin %> org upsert Acme --by name --field "Tier=Gold" --dry-run',
+  ]
+
+  static args = {
+    value: Args.string({ required: true, description: 'value to match on' }),
+  }
+
+  static flags = {
+    ...BaseCommand.baseFlags,
+    by: Flags.string({
+      required: true,
+      description: 'Match field: name, or a searchable custom field',
+    }),
+    field: Flags.string({
+      multiple: true,
+      description: 'Field to set as "Name=Value" (repeatable)',
+    }),
+    body: Flags.string({ description: 'Raw JSON body to merge' }),
+    'dry-run': Flags.boolean({
+      description: 'Preview the action without writing',
+    }),
+  }
+
+  async run() {
+    const { args, flags } = await this.parse(OrgUpsertCommand)
+    const result = await upsertWithDefs({
+      client: this.apiClient,
+      entity: 'org',
+      by: flags.by,
+      value: args.value,
+      fields: flags.field,
+      rawBody: flags.body,
+      dryRun: flags['dry-run'],
+    })
+
+    if (this.resolveFormat() !== 'table') {
+      await this.outputResults(result, {})
+      return
+    }
+    this.log(summarizeUpsert(result, 'org'))
+  }
+}
diff --git a/src/commands/person/import.js b/src/commands/person/import.js
index a640b33..c9202c3 100644
--- a/src/commands/person/import.js
+++ b/src/commands/person/import.js
@@ -6,6 +6,7 @@ import BaseCommand from '../../base-command.js'
 import { parseCsv } from '../../lib/csv-parse.js'
 import { prepareImportBodies } from '../../lib/import.js'
 import { bulkRun } from '../../lib/bulk.js'
+import { bulkUpsertRows } from '../../lib/upsert.js'
 import { getFields } from '../../lib/fields.js'
 import { confirmAction } from '../../lib/confirm.js'
 import { CliError } from '../../lib/errors.js'
@@ -43,6 +44,13 @@ export default class PersonImportCommand extends BaseCommand {
 
   static flags = {
     ...BaseCommand.baseFlags,
+    upsert: Flags.boolean({
+      description: 'Match each row on --match-on, then create or update',
+      default: false,
+    }),
+    'match-on': Flags.string({
+      description: 'Field to match rows on in --upsert mode (e.g. email)',
+    }),
     'dry-run': Flags.boolean({
       description: 'Validate every row without creating anything',
       default: false,
@@ -62,14 +70,39 @@ export default class PersonImportCommand extends BaseCommand {
       throw new CliError('CSV must include a "name" column', { exitCode: 64 })
     }
 
+    let matchIdx
+    if (flags.upsert) {
+      if (!flags['match-on']) {
+        throw new CliError('--upsert requires --match-on <field>', {
+          exitCode: 64,
+        })
+      }
+      matchIdx = headers.findIndex(
+        (h) => h.toLowerCase() === flags['match-on'].toLowerCase(),
+      )
+      if (matchIdx < 0) {
+        throw new CliError(
+          `--match-on "${flags['match-on']}" is not a column in ${args.file}`,
+          { exitCode: 64 },
+        )
+      }
+    }
+
     const needsDefs = headers.some((h) => !(h.toLowerCase() in SPECIAL_COLUMNS))
+    const defs =
+      needsDefs || flags.upsert ? await getFields(this.apiClient, 'person') : []
     const bodies = prepareImportBodies({
       headers,
       rows,
       specialColumns: SPECIAL_COLUMNS,
-      defs: needsDefs ? await getFields(this.apiClient, 'person') : [],
+      defs,
     })
 
+    if (flags.upsert) {
+      await this.upsertRows({ args, flags, rows, bodies, matchIdx, defs })
+      return
+    }
+
     if (flags['dry-run']) {
       this.log(chalk.green(`${bodies.length} rows valid — nothing created`))
       return
@@ -115,4 +148,59 @@ export default class PersonImportCommand extends BaseCommand {
       )
     }
   }
+
+  /** Idempotent CSV path: match each row on --match-on, then create or PATCH. */
+  async upsertRows({ args, flags, rows, bodies, matchIdx, defs }) {
+    const matchOn = flags['match-on']
+    const items = bodies.map((body, i) => ({ body, value: rows[i][matchIdx] }))
+
+    if (!flags['dry-run']) {
+      const ok = await confirmAction(
+        `Upsert ${items.length} persons from ${args.file} (match on ${matchOn})?`,
+        flags.yes,
+      )
+      if (!ok) {
+        throw new CliError('Aborted', { exitCode: 1 })
+      }
+    }
+
+    const spinner = ora(`Upserting ${items.length} persons...`).start()
+    let summary
+    try {
+      summary = await bulkUpsertRows({
+        client: this.apiClient,
+        entity: 'person',
+        matchOn,
+        rows: items,
+        defs,
+        dryRun: flags['dry-run'],
+        onProgress: (done, total) => {
+          spinner.text = `Upserting persons ${done}/${total}`
+        },
+      })
+    } finally {
+      spinner.stop()
+    }
+
+    const { created, updated, unchanged } = summary.counts
+    const prefix = flags['dry-run'] ? '[dry-run] ' : ''
+    this.log(
+      chalk.green(
+        `${prefix}${created} created, ${updated} updated, ${unchanged} unchanged`,
+      ),
+    )
+
+    if (summary.failed.length > 0) {
+      for (const { item, error } of summary.failed) {
+        this.log(chalk.red(`  ✘ ${matchOn}="${item.value}": ${error}`))
+      }
+      // Surface 65 when every failure is a data-validation error (ambiguous
+      // match, empty match value); fall back to 1 for mixed/transport errors.
+      const allData = summary.failed.every((f) => f.exitCode === 65)
+      throw new CliError(
+        `${summary.failed.length} of ${items.length} rows failed`,
+        { exitCode: allData ? 65 : 1 },
+      )
+    }
+  }
 }
diff --git a/src/commands/person/upsert.js b/src/commands/person/upsert.js
new file mode 100644
index 0000000..4ecae7f
--- /dev/null
+++ b/src/commands/person/upsert.js
@@ -0,0 +1,55 @@
+import { Args, Flags } from '@oclif/core'
+import BaseCommand from '../../base-command.js'
+import { upsertWithDefs, summarizeUpsert } from '../../lib/upsert.js'
+
+export default class PersonUpsertCommand extends BaseCommand {
+  static description =
+    'Idempotent person upsert: match by --by, then create or PATCH only the ' +
+    'changed fields. Refuses (exit 65) if more than one record matches.'
+
+  static examples = [
+    '<%= config.bin %> person upsert a@x.com --by email --field "Tier=Gold"',
+    '<%= config.bin %> person upsert "Jane Doe" --by name --body \'{"owner_id":42}\'',
+    '<%= config.bin %> person upsert a@x.com --by email --field "Tier=Gold" --dry-run',
+  ]
+
+  static args = {
+    value: Args.string({ required: true, description: 'value to match on' }),
+  }
+
+  static flags = {
+    ...BaseCommand.baseFlags,
+    by: Flags.string({
+      required: true,
+      description:
+        'Match field: email, name, phone, or a searchable custom field',
+    }),
+    field: Flags.string({
+      multiple: true,
+      description: 'Field to set as "Name=Value" (repeatable)',
+    }),
+    body: Flags.string({ description: 'Raw JSON body to merge' }),
+    'dry-run': Flags.boolean({
+      description: 'Preview the action without writing',
+    }),
+  }
+
+  async run() {
+    const { args, flags } = await this.parse(PersonUpsertCommand)
+    const result = await upsertWithDefs({
+      client: this.apiClient,
+      entity: 'person',
+      by: flags.by,
+      value: args.value,
+      fields: flags.field,
+      rawBody: flags.body,
+      dryRun: flags['dry-run'],
+    })
+
+    if (this.resolveFormat() !== 'table') {
+      await this.outputResults(result, {})
+      return
+    }
+    this.log(summarizeUpsert(result, 'person'))
+  }
+}
diff --git a/src/lib/backup-diff.js b/src/lib/backup-diff.js
index cd8ed36..eb31926 100644
--- a/src/lib/backup-diff.js
+++ b/src/lib/backup-diff.js
@@ -17,7 +17,7 @@ const FIELDS_FOR = {
 const SCHEMA_RESOURCES = new Set(Object.values(FIELDS_FOR))
 
 /** Recursively sort object keys so equality is key-order insensitive. */
-function canon(value) {
+export function canon(value) {
   if (Array.isArray(value)) return value.map(canon)
   if (value && typeof value === 'object') {
     return Object.fromEntries(
@@ -30,7 +30,7 @@ function canon(value) {
 }
 
 /** Stable equality for scalar/array/object field values (key-order safe). */
-function eq(a, b) {
+export function eq(a, b) {
   return JSON.stringify(canon(a ?? null)) === JSON.stringify(canon(b ?? null))
 }
 
diff --git a/src/lib/bulk.js b/src/lib/bulk.js
index 85c4d26..6987e52 100644
--- a/src/lib/bulk.js
+++ b/src/lib/bulk.js
@@ -97,7 +97,9 @@ export async function bulkRun(
       succeeded.push({ item, result })
     } catch (err) {
       debug('bulk item %o failed: %s', item, err.message)
-      failed.push({ item, error: err.message })
+      // Keep the exit code so the caller can map a batch of data-validation
+      // failures (e.g. ambiguous upsert matches → 65) to the right exit code.
+      failed.push({ item, error: err.message, exitCode: err.exitCode })
     }
     onProgress?.(index + 1, items.length)
   }
diff --git a/src/lib/import.js b/src/lib/import.js
index 36b7648..6b59225 100644
--- a/src/lib/import.js
+++ b/src/lib/import.js
@@ -23,6 +23,17 @@ export function prepareImportBodies({
     Object.entries(specialColumns).map(([k, v]) => [k.toLowerCase(), v]),
   )
 
+  // Duplicate headers silently overwrite each other (last cell wins) — reject
+  // them up front rather than losing a column's data without warning.
+  const seen = new Set()
+  for (const header of headers) {
+    const lower = header.toLowerCase()
+    if (seen.has(lower)) {
+      throw new CliError(`Duplicate CSV column "${header}"`, { exitCode: 65 })
+    }
+    seen.add(lower)
+  }
+
   return rows.map((row, index) => {
     const typed = {}
     const fields = []
diff --git a/src/lib/lookup.js b/src/lib/lookup.js
new file mode 100644
index 0000000..b93235c
--- /dev/null
+++ b/src/lib/lookup.js
@@ -0,0 +1,187 @@
+import { CliError } from './errors.js'
+
+/** v2 custom-field types that the search endpoints can match on. */
+const SEARCHABLE_TYPES = new Set([
+  'address',
+  'varchar',
+  'varchar_auto',
+  'text',
+  'double',
+  'monetary',
+  'phone',
+])
+const NUMERIC_TYPES = new Set(['double', 'monetary'])
+
+/**
+ * Per-entity config: the scoped-search endpoint, the record-fetch base, and
+ * the built-in match fields. Each built-in gives the search `fields` scope,
+ * how to pull the comparable value(s) off the *full* record (NOT the search
+ * item — see lookupByField), and whether the compare is case-insensitive.
+ */
+const SEARCH_CONFIG = {
+  person: {
+    searchPath: '/api/v2/persons/search',
+    recordPath: '/api/v2/persons',
+    builtins: {
+      email: {
+        scope: 'email',
+        extract: (it) => (it.emails ?? []).map((e) => e.value),
+        ci: true,
+      },
+      name: { scope: 'name', extract: (it) => [it.name], ci: false },
+      phone: {
+        scope: 'phone',
+        extract: (it) => (it.phones ?? []).map((p) => p.value),
+        ci: false,
+      },
+    },
+  },
+  org: {
+    searchPath: '/api/v2/organizations/search',
+    recordPath: '/api/v2/organizations',
+    builtins: {
+      name: { scope: 'name', extract: (it) => [it.name], ci: false },
+    },
+  },
+  deal: {
+    searchPath: '/api/v2/deals/search',
+    recordPath: '/api/v2/deals',
+    builtins: {
+      title: { scope: 'title', extract: (it) => [it.title], ci: false },
+    },
+  },
+}
+
+function valuesEqual(actual, expected, ci) {
+  if (actual == null) return false
+  if (typeof expected === 'number') return Number(actual) === expected
+  return ci
+    ? String(actual).toLowerCase() === String(expected).toLowerCase()
+    : String(actual) === String(expected)
+}
+
+/**
+ * Find the record of `entity` whose `field` exactly equals `value`. Routes a
+ * built-in field (person email/name/phone, org name, deal title) or a
+ * searchable custom field (by name) to the scoped /search endpoint to collect
+ * candidate ids across cursor pages, then FETCHES each candidate's full record
+ * and RE-VERIFIES it client-side. The fetch is not optional: the search result
+ * `item` is a lossy projection (emails/phones come back as bare strings,
+ * custom_fields as an unattributed value array) so it can neither be trusted
+ * for verification nor used as the record to diff against. `exact_match` is
+ * also not a unique-key lookup (case-insensitive; a custom-field search scans
+ * every custom field at once), which is the other reason to re-verify against
+ * the authoritative record. The surviving count decides the tri-state:
+ * 0 → create, 1 → update, >1 → caller must refuse.
+ *
+ * @param {{ get: Function }} client
+ * @param {'person'|'org'|'deal'} entity
+ * @param {object[]} [defs] field definitions (needed for a custom --by field)
+ * @param {string} field built-in name or custom-field name
+ * @param {string} value the value to match
+ * @returns {Promise<{ status: 'none' } | { status: 'unique', id: number, record: object } | { status: 'ambiguous', matches: number[] }>}
+ */
+export async function lookupByField({
+  client,
+  entity,
+  defs = [],
+  field,
+  value,
+}) {
+  const cfg = SEARCH_CONFIG[entity]
+  const builtin = cfg.builtins[field.toLowerCase()]
+
+  let scope
+  let extract
+  let ci
+  let compareValue
+
+  if (builtin) {
+    ;({ scope, extract, ci } = builtin)
+    compareValue = value
+  } else {
+    const def = defs.find(
+      (d) =>
+        d.field_name?.toLowerCase() === field.toLowerCase() ||
+        d.field_code === field,
+    )
+    if (!def) {
+      throw new CliError(
+        `Unknown field "${field}" for ${entity}. Run: pdcli field list ${entity}`,
+        { exitCode: 64 },
+      )
+    }
+    if (!SEARCHABLE_TYPES.has(def.field_type)) {
+      throw new CliError(
+        `Field "${field}" (${def.field_type}) is not searchable — --by needs a ` +
+          `built-in key or a searchable custom field (text/number/phone/address)`,
+        { exitCode: 64 },
+      )
+    }
+    scope = 'custom_fields'
+    const key = def.field_code
+    extract = (it) => [it.custom_fields?.[key]]
+    ci = false
+    if (NUMERIC_TYPES.has(def.field_type)) {
+      compareValue = Number(value)
+      // A non-numeric value coerces to NaN, which loses every comparison
+      // (NaN !== NaN) — so a match would silently miss and we'd create or
+      // inject a NaN value. Refuse loudly instead.
+      if (!Number.isFinite(compareValue)) {
+        throw new CliError(
+          `"${value}" is not a valid number for field "${field}"`,
+          { exitCode: 65 },
+        )
+      }
+    } else {
+      compareValue = value
+    }
+  }
+
+  // 1. Collect candidate ids from the scoped exact_match search (the cheap
+  //    finder). The item body is lossy, so we keep only the id.
+  const candidateIds = []
+  const seen = new Set()
+  let cursor
+  do {
+    const body = await client.get(cfg.searchPath, {
+      query: {
+        term: value,
+        exact_match: true,
+        fields: scope,
+        // The per-entity /search endpoints cap limit at 100 (the live API 400s
+        // on more, despite the 500 list cap); cursor paging collects the rest.
+        limit: 100,
+        cursor,
+      },
+    })
+    for (const entry of body.data?.items ?? []) {
+      const id = entry.item?.id
+      if (id != null && !seen.has(id)) {
+        seen.add(id)
+        candidateIds.push(id)
+      }
+    }
+    cursor = body.additional_data?.next_cursor ?? null
+  } while (cursor)
+
+  // 2. Fetch each candidate's full record and re-verify the matched field
+  //    against its authoritative value(s). The surviving records are both the
+  //    ambiguity count and (when unique) the record to diff against.
+  const survivors = []
+  for (const id of candidateIds) {
+    const record = (await client.get(`${cfg.recordPath}/${id}`)).data
+    if (
+      record &&
+      extract(record).some((v) => valuesEqual(v, compareValue, ci))
+    ) {
+      survivors.push(record)
+    }
+  }
+
+  if (survivors.length === 0) return { status: 'none' }
+  if (survivors.length === 1) {
+    return { status: 'unique', id: survivors[0].id, record: survivors[0] }
+  }
+  return { status: 'ambiguous', matches: survivors.map((s) => s.id) }
+}
diff --git a/src/lib/upsert.js b/src/lib/upsert.js
new file mode 100644
index 0000000..1d17a40
--- /dev/null
+++ b/src/lib/upsert.js
@@ -0,0 +1,249 @@
+import { CliError } from './errors.js'
+import { eq } from './backup-diff.js'
+import { lookupByField } from './lookup.js'
+import { getFields } from './fields.js'
+import { buildWriteBody } from './input.js'
+import { bulkRun } from './bulk.js'
+
+/** Entity → v2 write path. */
+const WRITE_PATH = {
+  person: '/api/v2/persons',
+  org: '/api/v2/organizations',
+  deal: '/api/v2/deals',
+}
+
+const NUMERIC_TYPES = new Set(['double', 'monetary'])
+
+/**
+ * Inject the match field+value into a CREATE body (so a created record actually
+ * carries the value it was matched on). `??=` so an explicit body value wins.
+ */
+function injectMatch(body, entity, field, value, defs) {
+  const key = field.toLowerCase()
+  if (entity === 'person' && key === 'email') {
+    body.emails ??= [{ value, primary: true }]
+    return
+  }
+  if (entity === 'person' && key === 'phone') {
+    body.phones ??= [{ value, primary: true }]
+    return
+  }
+  if ((entity === 'person' || entity === 'org') && key === 'name') {
+    body.name ??= value
+    return
+  }
+  if (entity === 'deal' && key === 'title') {
+    body.title ??= value
+    return
+  }
+  // custom field
+  const def = defs.find(
+    (d) => d.field_name?.toLowerCase() === key || d.field_code === field,
+  )
+  body.custom_fields ??= {}
+  body.custom_fields[def.field_code] ??= NUMERIC_TYPES.has(def.field_type)
+    ? Number(value)
+    : value
+}
+
+/** True if every element of an array is a primitive (or null). */
+function isPrimitiveArray(value) {
+  return (
+    Array.isArray(value) &&
+    value.every((v) => v == null || typeof v !== 'object')
+  )
+}
+
+/**
+ * Idempotency-aware field equality for diffBody. Beyond key-order insensitive
+ * `eq`, it treats two classes of field as equal when they carry the same
+ * *content* regardless of incidental shape the API adds back:
+ *   - emails/phones: compared by their set of `value`s (case-insensitive),
+ *     ignoring the `primary`/`label` flags the API echoes — otherwise an
+ *     injected `primary:true` would PATCH on every run.
+ *   - set-like primitive arrays (label_ids, multi-option custom fields):
+ *     compared order-insensitively, since the API may return them sorted.
+ */
+function fieldEq(incoming, existing, key) {
+  if (
+    (key === 'emails' || key === 'phones') &&
+    Array.isArray(incoming) &&
+    Array.isArray(existing)
+  ) {
+    const values = (arr) =>
+      arr.map((e) => String(e?.value ?? '').toLowerCase()).sort()
+    return eq(values(incoming), values(existing))
+  }
+  if (isPrimitiveArray(incoming) && isPrimitiveArray(existing)) {
+    return eq([...incoming].sort(), [...existing].sort())
+  }
+  return eq(incoming, existing)
+}
+
+/**
+ * Field-level diff for an idempotent PATCH: the subset of `incoming` whose
+ * value differs from `existing`. Top-level fields compare directly; the nested
+ * v2 `custom_fields` object is diffed key by key. Equality is key-order
+ * insensitive and set-aware for emails/phones/primitive arrays (see fieldEq),
+ * so re-running an unchanged upsert produces no PATCH. An empty result means
+ * "no change → skip the PATCH".
+ * @param {object} incoming the desired body
+ * @param {object} existing the current record
+ * @returns {object} the changed-only body
+ */
+export function diffBody(incoming, existing) {
+  const changed = {}
+  for (const [key, value] of Object.entries(incoming)) {
+    if (key === 'custom_fields' && value && typeof value === 'object') {
+      const cf = {}
+      for (const [ck, cv] of Object.entries(value)) {
+        if (!fieldEq(cv, existing?.custom_fields?.[ck], ck)) cf[ck] = cv
+      }
+      if (Object.keys(cf).length > 0) changed.custom_fields = cf
+    } else if (!fieldEq(value, existing?.[key], key)) {
+      changed[key] = value
+    }
+  }
+  return changed
+}
+
+/**
+ * Idempotent match-or-create. Looks up `entity` by `by`=`value`:
+ *   - 0 matches → create (the match field is injected into the body)
+ *   - 1 match   → PATCH only the fields that differ (diffBody); none → unchanged
+ *   - >1 match  → refuse (exit 65) — never guess which to write
+ *
+ * @param {{ get, post, patch }} client
+ * @param {'person'|'org'|'deal'} entity
+ * @param {string} by match field (built-in or searchable custom field)
+ * @param {string} value match value
+ * @param {object} body desired write body (from buildWriteBody)
+ * @param {object[]} [defs] field definitions (for custom --by + injection)
+ * @param {boolean} [dryRun]
+ * @returns {Promise<{ action: 'created'|'updated'|'unchanged', id?: number,
+ *   changed?: object, dryRun?: boolean, record?: object }>}
+ */
+export async function runUpsert({
+  client,
+  entity,
+  by,
+  value,
+  body,
+  defs = [],
+  dryRun = false,
+}) {
+  const writePath = WRITE_PATH[entity]
+  const match = await lookupByField({ client, entity, defs, field: by, value })
+
+  if (match.status === 'ambiguous') {
+    throw new CliError(
+      `--by ${by}="${value}" matches ${match.matches.length} ${entity} records ` +
+        `(ids: ${match.matches.join(', ')}) — refusing to guess. Narrow the match value.`,
+      { exitCode: 65 },
+    )
+  }
+
+  if (match.status === 'none') {
+    const createBody = { ...body }
+    injectMatch(createBody, entity, by, value, defs)
+    if (dryRun) return { action: 'created', dryRun: true, body: createBody }
+    const res = await client.post(writePath, { body: createBody })
+    return { action: 'created', id: res.data?.id, record: res.data }
+  }
+
+  // unique → diff-before-PATCH
+  const changed = diffBody(body, match.record)
+  if (Object.keys(changed).length === 0) {
+    return { action: 'unchanged', id: match.id }
+  }
+  if (dryRun) return { action: 'updated', id: match.id, dryRun: true, changed }
+  const res = await client.patch(`${writePath}/${match.id}`, { body: changed })
+  return { action: 'updated', id: match.id, changed, record: res.data }
+}
+
+/**
+ * Command-facing wrapper: fetch field defs, build the write body from --field /
+ * --body, then run the upsert. Keeps the three entity commands trivial.
+ * @param {object} options
+ * @returns {Promise<object>} the runUpsert result
+ */
+export async function upsertWithDefs({
+  client,
+  entity,
+  by,
+  value,
+  fields = [],
+  rawBody,
+  dryRun = false,
+}) {
+  const defs = await getFields(client, entity)
+  const body = buildWriteBody({ fields, rawBody, defs })
+  return runUpsert({ client, entity, by, value, body, defs, dryRun })
+}
+
+/**
+ * Upsert a batch of prepared bodies — one per CSV row — each matched on the
+ * shared `matchOn` field using that row's own `value`. Runs sequentially
+ * through bulkRun's pacing; a per-row failure (empty match value, ambiguous
+ * match, or API error) is collected, never aborting the batch. The surviving
+ * results are tallied by action so the caller can report created/updated/
+ * unchanged counts.
+ *
+ * @param {object} options
+ * @param {{ get, post, patch }} options.client
+ * @param {'person'|'org'|'deal'} options.entity
+ * @param {string} options.matchOn the match field shared by every row
+ * @param {{ body: object, value: string }[]} options.rows
+ * @param {object[]} options.defs field definitions (for a custom match field)
+ * @param {boolean} [options.dryRun]
+ * @param {number} [options.gapMs] pacing gap forwarded to bulkRun
+ * @param {(done: number, total: number) => void} [options.onProgress]
+ * @returns {Promise<{ succeeded, failed, counts: { created: number,
+ *   updated: number, unchanged: number } }>}
+ */
+export async function bulkUpsertRows({
+  client,
+  entity,
+  matchOn,
+  rows,
+  defs = [],
+  dryRun = false,
+  gapMs,
+  onProgress,
+}) {
+  const summary = await bulkRun(
+    rows,
+    ({ body, value }) => {
+      if (value == null || value === '') {
+        throw new CliError(`empty "${matchOn}" value — cannot match a row`, {
+          exitCode: 65,
+        })
+      }
+      return runUpsert({
+        client,
+        entity,
+        by: matchOn,
+        value,
+        body,
+        defs,
+        dryRun,
+      })
+    },
+    { gapMs, onProgress },
+  )
+
+  const counts = { created: 0, updated: 0, unchanged: 0 }
+  for (const { result } of summary.succeeded) counts[result.action] += 1
+  return { ...summary, counts }
+}
+
+/** One-line human summary of an upsert result for table output. */
+export function summarizeUpsert(result, entity) {
+  const prefix = result.dryRun ? '[dry-run] would ' : ''
+  if (result.action === 'unchanged') return `${entity} #${result.id} unchanged`
+  if (result.action === 'created') {
+    return `${prefix}create ${entity}${result.id != null ? ` #${result.id}` : ''}`
+  }
+  const n = result.changed ? Object.keys(result.changed).length : 0
+  return `${prefix}update ${entity} #${result.id} (${n} field${n === 1 ? '' : 's'})`
+}
diff --git a/test/commands/deal/upsert.test.js b/test/commands/deal/upsert.test.js
new file mode 100644
index 0000000..dfbe3b6
--- /dev/null
+++ b/test/commands/deal/upsert.test.js
@@ -0,0 +1,108 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
+import nock from 'nock'
+
+const mockResolveCredentials = vi.fn()
+vi.mock('../../../src/lib/auth.js', async (importOriginal) => {
+  const actual = await importOriginal()
+  return { ...actual, resolveCredentials: mockResolveCredentials }
+})
+
+vi.mock('../../../src/lib/config.js', () => ({
+  loadConfig: vi.fn().mockReturnValue({ activeProfile: 'default' }),
+}))
+
+const { default: DealUpsertCommand } =
+  await import('../../../src/commands/deal/upsert.js')
+const { clearFieldsCache } = await import('../../../src/lib/fields.js')
+import { runCmd, mockApi } from '../../helpers.js'
+
+function mockFields(defs = []) {
+  mockApi().get('/api/v2/dealFields').reply(200, { success: true, data: defs })
+}
+
+function mockSearch(records = []) {
+  mockApi()
+    .get('/api/v2/deals/search')
+    .query(true)
+    .reply(200, {
+      success: true,
+      data: { items: records.map((r) => ({ item: { id: r.id } })) },
+      additional_data: { next_cursor: null },
+    })
+  for (const r of records) {
+    mockApi()
+      .get(`/api/v2/deals/${r.id}`)
+      .reply(200, { success: true, data: r })
+  }
+}
+
+describe('deal upsert', () => {
+  beforeEach(() => {
+    nock.cleanAll()
+    clearFieldsCache()
+    mockResolveCredentials.mockResolvedValue({
+      companyDomain: 'acme',
+      token: 'tok',
+      source: 'profile',
+    })
+  })
+
+  afterEach(() => {
+    nock.cleanAll()
+  })
+
+  it('creates when no deal matches, injecting the title (json)', async () => {
+    mockFields()
+    mockSearch([])
+    mockApi()
+      .post('/api/v2/deals', { title: 'Acme expansion' })
+      .reply(201, { success: true, data: { id: 3, title: 'Acme expansion' } })
+
+    const stdout = await runCmd(DealUpsertCommand, [
+      'Acme expansion',
+      '--by',
+      'title',
+      '--output',
+      'json',
+    ])
+    const out = JSON.parse(stdout)
+    expect(out.action).toBe('created')
+    expect(out.id).toBe(3)
+  })
+
+  it('PATCHes one field and prints a singular summary (table)', async () => {
+    mockFields()
+    mockSearch([{ id: 7, title: 'Acme expansion', value: 100 }])
+    mockApi()
+      .patch('/api/v2/deals/7', { value: 200 })
+      .reply(200, { success: true, data: { id: 7, value: 200 } })
+
+    const stdout = await runCmd(DealUpsertCommand, [
+      'Acme expansion',
+      '--by',
+      'title',
+      '--body',
+      '{"value":200}',
+      '--output',
+      'table',
+    ])
+    expect(stdout).toBe('update deal #7 (1 field)')
+  })
+
+  it('previews an update without writing, pluralizing the count (dry-run table)', async () => {
+    mockFields()
+    mockSearch([{ id: 3, title: 'Acme', value: 1, owner_id: 1 }])
+
+    const stdout = await runCmd(DealUpsertCommand, [
+      'Acme',
+      '--by',
+      'title',
+      '--body',
+      '{"value":2,"owner_id":9}',
+      '--dry-run',
+      '--output',
+      'table',
+    ])
+    expect(stdout).toBe('[dry-run] would update deal #3 (2 fields)')
+  })
+})
diff --git a/test/commands/org/import.test.js b/test/commands/org/import.test.js
index 7fa7708..5f525f5 100644
--- a/test/commands/org/import.test.js
+++ b/test/commands/org/import.test.js
@@ -184,3 +184,192 @@ describe('org import unnamed failures', () => {
     )
   })
 })
+
+describe('org import --upsert', () => {
+  beforeEach(() => {
+    nock.cleanAll()
+    clearFieldsCache()
+    mockConfirmAction.mockReset()
+    mockConfirmAction.mockResolvedValue(true)
+    mockReadFileSync.mockReset()
+    mockResolveCredentials.mockResolvedValue({
+      mode: 'token',
+      companyDomain: 'acme',
+      token: 'tok',
+      source: 'profile',
+    })
+  })
+
+  afterEach(() => nock.cleanAll())
+
+  function mockFields() {
+    mockApi()
+      .get('/api/v2/organizationFields')
+      .reply(200, { success: true, data: [] })
+  }
+
+  it('creates new rows and patches matched ones, reporting counts', async () => {
+    mockFields()
+    mockReadFileSync.mockReturnValue('name,owner_id\nAcme,42\nGlobex,5\n')
+    mockApi()
+      .get('/api/v2/organizations/search')
+      .query((q) => q.term === 'Acme')
+      .reply(200, {
+        success: true,
+        data: { items: [{ item: { id: 7 } }] },
+        additional_data: { next_cursor: null },
+      })
+    mockApi()
+      .get('/api/v2/organizations/7')
+      .reply(200, {
+        success: true,
+        data: { id: 7, name: 'Acme', owner_id: 1 },
+      })
+    mockApi()
+      .get('/api/v2/organizations/search')
+      .query((q) => q.term === 'Globex')
+      .reply(200, {
+        success: true,
+        data: { items: [] },
+        additional_data: { next_cursor: null },
+      })
+    mockApi()
+      .patch('/api/v2/organizations/7')
+      .reply(200, { success: true, data: { id: 7 } })
+    mockApi()
+      .post('/api/v2/organizations')
+      .reply(201, { success: true, data: { id: 8 } })
+
+    const stdout = await runCmd(OrgImportCommand, [
+      'o.csv',
+      '--upsert',
+      '--match-on',
+      'name',
+      '--yes',
+    ])
+    expect(stdout).toContain('1 created')
+    expect(stdout).toContain('1 updated')
+  })
+
+  it('requires --match-on', async () => {
+    mockReadFileSync.mockReturnValue('name\nAcme\n')
+    nock.disableNetConnect()
+    try {
+      await expect(
+        OrgImportCommand.run(['o.csv', '--upsert', '--yes']),
+      ).rejects.toThrow(/match-on/i)
+    } finally {
+      nock.enableNetConnect()
+    }
+  })
+
+  it('rejects a --match-on column that is not in the CSV', async () => {
+    mockReadFileSync.mockReturnValue('name\nAcme\n')
+    nock.disableNetConnect()
+    try {
+      await expect(
+        OrgImportCommand.run([
+          'o.csv',
+          '--upsert',
+          '--match-on',
+          'industry',
+          '--yes',
+        ]),
+      ).rejects.toThrow(/column/i)
+    } finally {
+      nock.enableNetConnect()
+    }
+  })
+
+  it('--dry-run looks up but writes nothing', async () => {
+    mockFields()
+    mockReadFileSync.mockReturnValue('name\nAcme\n')
+    mockApi()
+      .get('/api/v2/organizations/search')
+      .query((q) => q.term === 'Acme')
+      .reply(200, {
+        success: true,
+        data: { items: [] },
+        additional_data: { next_cursor: null },
+      })
+
+    const stdout = await runCmd(OrgImportCommand, [
+      'o.csv',
+      '--upsert',
+      '--match-on',
+      'name',
+      '--dry-run',
+    ])
+    expect(stdout).toContain('[dry-run]')
+    expect(stdout).toContain('1 created')
+    expect(mockConfirmAction).not.toHaveBeenCalled()
+  })
+
+  it('exits 65 when an ambiguous row is the only failure', async () => {
+    mockFields()
+    mockReadFileSync.mockReturnValue('name\ndup\n')
+    mockApi()
+      .get('/api/v2/organizations/search')
+      .query((q) => q.term === 'dup')
+      .reply(200, {
+        success: true,
+        data: { items: [{ item: { id: 1 } }, { item: { id: 2 } }] },
+        additional_data: { next_cursor: null },
+      })
+    mockApi()
+      .get('/api/v2/organizations/1')
+      .reply(200, { success: true, data: { id: 1, name: 'dup' } })
+    mockApi()
+      .get('/api/v2/organizations/2')
+      .reply(200, { success: true, data: { id: 2, name: 'dup' } })
+
+    const err = await OrgImportCommand.run([
+      'o.csv',
+      '--upsert',
+      '--match-on',
+      'name',
+      '--yes',
+    ]).catch((e) => e)
+    expect(err.message).toMatch(/1 of 1/i)
+    expect(err.exitCode ?? err.oclif?.exit).toBe(65)
+  })
+
+  it('exits 1 when a row fails on a non-validation API error', async () => {
+    mockFields()
+    mockReadFileSync.mockReturnValue('name,owner_id\nAcme,42\n')
+    mockApi()
+      .get('/api/v2/organizations/search')
+      .query((q) => q.term === 'Acme')
+      .reply(200, {
+        success: true,
+        data: { items: [{ item: { id: 7 } }] },
+        additional_data: { next_cursor: null },
+      })
+    mockApi()
+      .get('/api/v2/organizations/7')
+      .reply(200, { success: true, data: { id: 7, name: 'Acme', owner_id: 1 } })
+    mockApi()
+      .patch('/api/v2/organizations/7')
+      .reply(401, { success: false, error: 'unauthorized' })
+
+    const err = await OrgImportCommand.run([
+      'o.csv',
+      '--upsert',
+      '--match-on',
+      'name',
+      '--no-retry',
+      '--yes',
+    ]).catch((e) => e)
+    expect(err.message).toMatch(/1 of 1/i)
+    expect(err.exitCode ?? err.oclif?.exit).toBe(1)
+  })
+
+  it('aborts when the upsert confirmation is declined', async () => {
+    mockFields()
+    mockReadFileSync.mockReturnValue('name\nAcme\n')
+    mockConfirmAction.mockResolvedValue(false)
+    await expect(
+      OrgImportCommand.run(['o.csv', '--upsert', '--match-on', 'name']),
+    ).rejects.toThrow(/abort/i)
+  })
+})
diff --git a/test/commands/org/upsert.test.js b/test/commands/org/upsert.test.js
new file mode 100644
index 0000000..bb297b3
--- /dev/null
+++ b/test/commands/org/upsert.test.js
@@ -0,0 +1,105 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
+import nock from 'nock'
+
+const mockResolveCredentials = vi.fn()
+vi.mock('../../../src/lib/auth.js', async (importOriginal) => {
+  const actual = await importOriginal()
+  return { ...actual, resolveCredentials: mockResolveCredentials }
+})
+
+vi.mock('../../../src/lib/config.js', () => ({
+  loadConfig: vi.fn().mockReturnValue({ activeProfile: 'default' }),
+}))
+
+const { default: OrgUpsertCommand } =
+  await import('../../../src/commands/org/upsert.js')
+const { clearFieldsCache } = await import('../../../src/lib/fields.js')
+import { runCmd, mockApi } from '../../helpers.js'
+
+function mockFields(defs = []) {
+  mockApi()
+    .get('/api/v2/organizationFields')
+    .reply(200, { success: true, data: defs })
+}
+
+function mockSearch(records = []) {
+  mockApi()
+    .get('/api/v2/organizations/search')
+    .query(true)
+    .reply(200, {
+      success: true,
+      data: { items: records.map((r) => ({ item: { id: r.id } })) },
+      additional_data: { next_cursor: null },
+    })
+  for (const r of records) {
+    mockApi()
+      .get(`/api/v2/organizations/${r.id}`)
+      .reply(200, { success: true, data: r })
+  }
+}
+
+describe('org upsert', () => {
+  beforeEach(() => {
+    nock.cleanAll()
+    clearFieldsCache()
+    mockResolveCredentials.mockResolvedValue({
+      companyDomain: 'acme',
+      token: 'tok',
+      source: 'profile',
+    })
+  })
+
+  afterEach(() => {
+    nock.cleanAll()
+  })
+
+  it('creates when no org matches, injecting the name (json)', async () => {
+    mockFields()
+    mockSearch([])
+    mockApi()
+      .post('/api/v2/organizations', { name: 'Acme' })
+      .reply(201, { success: true, data: { id: 5, name: 'Acme' } })
+
+    const stdout = await runCmd(OrgUpsertCommand, [
+      'Acme',
+      '--by',
+      'name',
+      '--output',
+      'json',
+    ])
+    const out = JSON.parse(stdout)
+    expect(out.action).toBe('created')
+    expect(out.id).toBe(5)
+  })
+
+  it('reports unchanged when the matched org already matches (table)', async () => {
+    mockFields()
+    mockSearch([{ id: 7, name: 'Acme', owner_id: 1 }])
+
+    const stdout = await runCmd(OrgUpsertCommand, [
+      'Acme',
+      '--by',
+      'name',
+      '--body',
+      '{"owner_id":1}',
+      '--output',
+      'table',
+    ])
+    expect(stdout).toBe('org #7 unchanged')
+  })
+
+  it('previews a create without writing (dry-run table)', async () => {
+    mockFields()
+    mockSearch([])
+
+    const stdout = await runCmd(OrgUpsertCommand, [
+      'Acme',
+      '--by',
+      'name',
+      '--dry-run',
+      '--output',
+      'table',
+    ])
+    expect(stdout).toBe('[dry-run] would create org')
+  })
+})
diff --git a/test/commands/person/import.test.js b/test/commands/person/import.test.js
index 1d78864..8857fcf 100644
--- a/test/commands/person/import.test.js
+++ b/test/commands/person/import.test.js
@@ -207,3 +207,210 @@ describe('person import custom-field headers and unnamed failures', () => {
     )
   })
 })
+
+describe('person import --upsert', () => {
+  beforeEach(() => {
+    nock.cleanAll()
+    mockConfirmAction.mockReset()
+    mockConfirmAction.mockResolvedValue(true)
+    mockReadFileSync.mockReset()
+    mockResolveCredentials.mockResolvedValue({
+      mode: 'token',
+      companyDomain: 'acme',
+      token: 'tok',
+      source: 'profile',
+    })
+  })
+
+  afterEach(() => nock.cleanAll())
+
+  async function withEmptyFields(fn) {
+    const { clearFieldsCache } = await import('../../../src/lib/fields.js')
+    clearFieldsCache()
+    mockApi()
+      .get('/api/v2/personFields')
+      .reply(200, { success: true, data: [] })
+    return fn()
+  }
+
+  it('creates new rows and patches matched ones, reporting counts', async () => {
+    await withEmptyFields(async () => {
+      mockReadFileSync.mockReturnValue(
+        'name,email,owner_id\nJane,a@x.com,42\nBob,b@x.com,5\n',
+      )
+      mockApi()
+        .get('/api/v2/persons/search')
+        .query((q) => q.term === 'a@x.com')
+        .reply(200, {
+          success: true,
+          data: { items: [{ item: { id: 7 } }] },
+          additional_data: { next_cursor: null },
+        })
+      mockApi()
+        .get('/api/v2/persons/7')
+        .reply(200, {
+          success: true,
+          data: { id: 7, emails: [{ value: 'a@x.com' }], owner_id: 1 },
+        })
+      mockApi()
+        .get('/api/v2/persons/search')
+        .query((q) => q.term === 'b@x.com')
+        .reply(200, {
+          success: true,
+          data: { items: [] },
+          additional_data: { next_cursor: null },
+        })
+      mockApi()
+        .patch('/api/v2/persons/7')
+        .reply(200, { success: true, data: { id: 7 } })
+      mockApi()
+        .post('/api/v2/persons')
+        .reply(201, { success: true, data: { id: 8 } })
+
+      const stdout = await runCmd(PersonImportCommand, [
+        'p.csv',
+        '--upsert',
+        '--match-on',
+        'email',
+        '--yes',
+      ])
+      expect(stdout).toContain('1 created')
+      expect(stdout).toContain('1 updated')
+    })
+  })
+
+  it('requires --match-on', async () => {
+    mockReadFileSync.mockReturnValue('name,email\nJane,a@x.com\n')
+    nock.disableNetConnect()
+    try {
+      await expect(
+        PersonImportCommand.run(['p.csv', '--upsert', '--yes']),
+      ).rejects.toThrow(/match-on/i)
+    } finally {
+      nock.enableNetConnect()
+    }
+  })
+
+  it('rejects a --match-on column that is not in the CSV', async () => {
+    mockReadFileSync.mockReturnValue('name,email\nJane,a@x.com\n')
+    nock.disableNetConnect()
+    try {
+      await expect(
+        PersonImportCommand.run([
+          'p.csv',
+          '--upsert',
+          '--match-on',
+          'phone',
+          '--yes',
+        ]),
+      ).rejects.toThrow(/column/i)
+    } finally {
+      nock.enableNetConnect()
+    }
+  })
+
+  it('--dry-run looks up but writes nothing', async () => {
+    await withEmptyFields(async () => {
+      mockReadFileSync.mockReturnValue('name,email\nJane,a@x.com\n')
+      mockApi()
+        .get('/api/v2/persons/search')
+        .query((q) => q.term === 'a@x.com')
+        .reply(200, {
+          success: true,
+          data: { items: [] },
+          additional_data: { next_cursor: null },
+        })
+
+      const stdout = await runCmd(PersonImportCommand, [
+        'p.csv',
+        '--upsert',
+        '--match-on',
+        'email',
+        '--dry-run',
+      ])
+      expect(stdout).toContain('[dry-run]')
+      expect(stdout).toContain('1 created')
+      expect(mockConfirmAction).not.toHaveBeenCalled()
+    })
+  })
+
+  it('exits 65 when an ambiguous row is the only failure', async () => {
+    await withEmptyFields(async () => {
+      mockReadFileSync.mockReturnValue('name,email\nJane,dup@x.com\n')
+      mockApi()
+        .get('/api/v2/persons/search')
+        .query((q) => q.term === 'dup@x.com')
+        .reply(200, {
+          success: true,
+          data: { items: [{ item: { id: 1 } }, { item: { id: 2 } }] },
+          additional_data: { next_cursor: null },
+        })
+      mockApi()
+        .get('/api/v2/persons/1')
+        .reply(200, {
+          success: true,
+          data: { id: 1, emails: [{ value: 'dup@x.com' }] },
+        })
+      mockApi()
+        .get('/api/v2/persons/2')
+        .reply(200, {
+          success: true,
+          data: { id: 2, emails: [{ value: 'dup@x.com' }] },
+        })
+
+      const err = await PersonImportCommand.run([
+        'p.csv',
+        '--upsert',
+        '--match-on',
+        'email',
+        '--yes',
+      ]).catch((e) => e)
+      expect(err.message).toMatch(/1 of 1/i)
+      expect(err.exitCode ?? err.oclif?.exit).toBe(65)
+    })
+  })
+
+  it('exits 1 when a row fails on a non-validation API error', async () => {
+    await withEmptyFields(async () => {
+      mockReadFileSync.mockReturnValue('name,email,owner_id\nJane,a@x.com,42\n')
+      mockApi()
+        .get('/api/v2/persons/search')
+        .query((q) => q.term === 'a@x.com')
+        .reply(200, {
+          success: true,
+          data: { items: [{ item: { id: 7 } }] },
+          additional_data: { next_cursor: null },
+        })
+      mockApi()
+        .get('/api/v2/persons/7')
+        .reply(200, {
+          success: true,
+          data: { id: 7, emails: [{ value: 'a@x.com' }], owner_id: 1 },
+        })
+      mockApi()
+        .patch('/api/v2/persons/7')
+        .reply(401, { success: false, error: 'unauthorized' })
+
+      const err = await PersonImportCommand.run([
+        'p.csv',
+        '--upsert',
+        '--match-on',
+        'email',
+        '--no-retry',
+        '--yes',
+      ]).catch((e) => e)
+      expect(err.message).toMatch(/1 of 1/i)
+      expect(err.exitCode ?? err.oclif?.exit).toBe(1)
+    })
+  })
+
+  it('aborts when the upsert confirmation is declined', async () => {
+    await withEmptyFields(async () => {
+      mockReadFileSync.mockReturnValue('name,email\nJane,a@x.com\n')
+      mockConfirmAction.mockResolvedValue(false)
+      await expect(
+        PersonImportCommand.run(['p.csv', '--upsert', '--match-on', 'email']),
+      ).rejects.toThrow(/abort/i)
+    })
+  })
+})
diff --git a/test/commands/person/upsert.test.js b/test/commands/person/upsert.test.js
new file mode 100644
index 0000000..9c57a7f
--- /dev/null
+++ b/test/commands/person/upsert.test.js
@@ -0,0 +1,136 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
+import nock from 'nock'
+
+const mockResolveCredentials = vi.fn()
+vi.mock('../../../src/lib/auth.js', async (importOriginal) => {
+  const actual = await importOriginal()
+  return { ...actual, resolveCredentials: mockResolveCredentials }
+})
+
+vi.mock('../../../src/lib/config.js', () => ({
+  loadConfig: vi.fn().mockReturnValue({ activeProfile: 'default' }),
+}))
+
+const { default: PersonUpsertCommand } =
+  await import('../../../src/commands/person/upsert.js')
+const { clearFieldsCache } = await import('../../../src/lib/fields.js')
+import { runCmd, mockApi } from '../../helpers.js'
+
+/** Mock the field-defs fetch (getFields always runs first). */
+function mockFields(defs = []) {
+  mockApi()
+    .get('/api/v2/personFields')
+    .reply(200, { success: true, data: defs })
+}
+
+/**
+ * Mock the scoped search as a candidate-id finder, then a record fetch per
+ * candidate (lookup verifies against the full record, not the search item).
+ */
+function mockSearch(records = []) {
+  mockApi()
+    .get('/api/v2/persons/search')
+    .query(true)
+    .reply(200, {
+      success: true,
+      data: { items: records.map((r) => ({ item: { id: r.id } })) },
+      additional_data: { next_cursor: null },
+    })
+  for (const r of records) {
+    mockApi()
+      .get(`/api/v2/persons/${r.id}`)
+      .reply(200, { success: true, data: r })
+  }
+}
+
+describe('person upsert', () => {
+  beforeEach(() => {
+    nock.cleanAll()
+    clearFieldsCache()
+    mockResolveCredentials.mockResolvedValue({
+      companyDomain: 'acme',
+      token: 'tok',
+      source: 'profile',
+    })
+  })
+
+  afterEach(() => {
+    nock.cleanAll()
+  })
+
+  it('creates when no match, injecting the match field (json)', async () => {
+    mockFields()
+    mockSearch([])
+    mockApi()
+      .post('/api/v2/persons', {
+        emails: [{ value: 'a@x.com', primary: true }],
+      })
+      .reply(201, { success: true, data: { id: 99, name: 'A' } })
+
+    const stdout = await runCmd(PersonUpsertCommand, [
+      'a@x.com',
+      '--by',
+      'email',
+      '--output',
+      'json',
+    ])
+    const out = JSON.parse(stdout)
+    expect(out.action).toBe('created')
+    expect(out.id).toBe(99)
+  })
+
+  it('PATCHes only the changed fields when exactly one matches (json)', async () => {
+    mockFields()
+    mockSearch([{ id: 7, emails: [{ value: 'a@x.com' }], owner_id: 1 }])
+    mockApi()
+      .patch('/api/v2/persons/7', { owner_id: 42 })
+      .reply(200, { success: true, data: { id: 7, owner_id: 42 } })
+
+    const stdout = await runCmd(PersonUpsertCommand, [
+      'a@x.com',
+      '--by',
+      'email',
+      '--body',
+      '{"owner_id":42}',
+      '--output',
+      'json',
+    ])
+    const out = JSON.parse(stdout)
+    expect(out.action).toBe('updated')
+    expect(out.id).toBe(7)
+    expect(out.changed).toEqual({ owner_id: 42 })
+  })
+
+  it('refuses an ambiguous match with exit 65', async () => {
+    mockFields()
+    mockSearch([
+      { id: 7, emails: [{ value: 'a@x.com' }] },
+      { id: 8, emails: [{ value: 'a@x.com' }] },
+    ])
+    const err = await PersonUpsertCommand.run([
+      'a@x.com',
+      '--by',
+      'email',
+      '--output',
+      'json',
+    ]).catch((e) => e)
+    expect(err.exitCode ?? err.oclif?.exit).toBe(65)
+  })
+
+  it('prints a one-line summary for the created record (table)', async () => {
+    mockFields()
+    mockSearch([])
+    mockApi()
+      .post('/api/v2/persons')
+      .reply(201, { success: true, data: { id: 99 } })
+
+    const stdout = await runCmd(PersonUpsertCommand, [
+      'a@x.com',
+      '--by',
+      'email',
+      '--output',
+      'table',
+    ])
+    expect(stdout).toBe('create person #99')
+  })
+})
diff --git a/test/lib/bulk.test.js b/test/lib/bulk.test.js
index 4d57da9..cbe98b7 100644
--- a/test/lib/bulk.test.js
+++ b/test/lib/bulk.test.js
@@ -108,6 +108,18 @@ describe('bulkRun', () => {
     expect(result.failed).toEqual([{ item: 2, error: 'boom on 2' }])
   })
 
+  it('preserves a failed item exit code for the caller', async () => {
+    const { CliError } = await import('../../src/lib/errors.js')
+    const result = await bulkRun(
+      [1],
+      async () => {
+        throw new CliError('ambiguous', { exitCode: 65 })
+      },
+      { gapMs: 0 },
+    )
+    expect(result.failed[0].exitCode).toBe(65)
+  })
+
   it('reports progress per item', async () => {
     const progress = []
     await bulkRun([1, 2], async () => ({}), {
diff --git a/test/lib/import.test.js b/test/lib/import.test.js
index 876b1f9..104ef2d 100644
--- a/test/lib/import.test.js
+++ b/test/lib/import.test.js
@@ -29,6 +29,34 @@ const PERSON_SPECIALS = {
   },
 }
 
+describe('prepareImportBodies duplicate headers', () => {
+  it('rejects a duplicate column header with exit 65 instead of losing data', () => {
+    expect(() =>
+      prepareImportBodies({
+        headers: ['name', 'email', 'email'],
+        rows: [['Jane', 'first@a.com', 'second@a.com']],
+        specialColumns: PERSON_SPECIALS,
+        defs: DEFS,
+      }),
+    ).toThrowError(/duplicate/i)
+  })
+
+  it('rejects duplicate headers case-insensitively', () => {
+    let caught
+    try {
+      prepareImportBodies({
+        headers: ['Name', 'name'],
+        rows: [['Jane', 'Jane']],
+        specialColumns: PERSON_SPECIALS,
+        defs: DEFS,
+      })
+    } catch (e) {
+      caught = e
+    }
+    expect(caught.exitCode).toBe(65)
+  })
+})
+
 describe('prepareImportBodies', () => {
   it('maps special columns and resolves the rest via field defs', () => {
     const bodies = prepareImportBodies({
diff --git a/test/lib/lookup.test.js b/test/lib/lookup.test.js
new file mode 100644
index 0000000..0d9e175
--- /dev/null
+++ b/test/lib/lookup.test.js
@@ -0,0 +1,376 @@
+import { describe, it, expect } from 'vitest'
+import { lookupByField } from '../../src/lib/lookup.js'
+
+/**
+ * Fake client modelling the two-step lookup: a `/search` GET returns lossy
+ * candidate items (id only is used), and a record-fetch GET
+ * (`/api/v2/<entity>/<id>`) returns the authoritative full record. `records`
+ * maps id → full record; `searchPages` is the queued list of search pages.
+ */
+function fakeClient({ searchPages = [], records = {} } = {}) {
+  let i = 0
+  const calls = []
+  return {
+    calls,
+    async get(path, opts) {
+      calls.push({ path, query: opts?.query })
+      if (path.endsWith('/search')) {
+        return searchPages[i++] ?? { data: { items: [] } }
+      }
+      const id = Number(path.split('/').pop())
+      return { data: records[id] ?? null }
+    },
+  }
+}
+
+/** A search page advertising candidate ids; full bodies live in `records`. */
+const page = (ids, next = null) => ({
+  data: { items: ids.map((id) => ({ item: { id } })) },
+  additional_data: { next_cursor: next },
+})
+
+describe('lookupByField', () => {
+  it('returns none when nothing matches', async () => {
+    const client = fakeClient({ searchPages: [page([])] })
+    const r = await lookupByField({
+      client,
+      entity: 'person',
+      field: 'email',
+      value: 'a@x.com',
+    })
+    expect(r.status).toBe('none')
+  })
+
+  it('returns unique with the id for exactly one verified match', async () => {
+    const client = fakeClient({
+      searchPages: [page([7])],
+      records: { 7: { id: 7, emails: [{ value: 'a@x.com' }] } },
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'person',
+      field: 'email',
+      value: 'a@x.com',
+    })
+    expect(r).toMatchObject({ status: 'unique', id: 7 })
+    expect(r.record).toEqual({ id: 7, emails: [{ value: 'a@x.com' }] })
+  })
+
+  it('refuses (ambiguous) on more than one verified match', async () => {
+    const client = fakeClient({
+      searchPages: [page([7, 8])],
+      records: {
+        7: { id: 7, emails: [{ value: 'a@x.com' }] },
+        8: { id: 8, emails: [{ value: 'a@x.com' }] },
+      },
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'person',
+      field: 'email',
+      value: 'a@x.com',
+    })
+    expect(r.status).toBe('ambiguous')
+    expect(r.matches).toEqual([7, 8])
+  })
+
+  it('matches email case-insensitively', async () => {
+    const client = fakeClient({
+      searchPages: [page([7])],
+      records: { 7: { id: 7, emails: [{ value: 'A@X.com' }] } },
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'person',
+      field: 'email',
+      value: 'a@x.com',
+    })
+    expect(r).toMatchObject({ status: 'unique', id: 7 })
+  })
+
+  it('re-verifies the full record, discarding a search over-match', async () => {
+    // search leaked a non-matching person (exact_match is not a unique key);
+    // the fetched record proves its email differs.
+    const client = fakeClient({
+      searchPages: [page([9])],
+      records: { 9: { id: 9, emails: [{ value: 'other@x.com' }] } },
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'person',
+      field: 'email',
+      value: 'a@x.com',
+    })
+    expect(r.status).toBe('none')
+  })
+
+  it('sends exact_match and the scoped fields/limit, then fetches the record', async () => {
+    const client = fakeClient({
+      searchPages: [page([5])],
+      records: { 5: { id: 5, name: 'Acme' } },
+    })
+    await lookupByField({ client, entity: 'org', field: 'name', value: 'Acme' })
+    expect(client.calls[0]).toMatchObject({
+      path: '/api/v2/organizations/search',
+      // Per-entity search caps limit at 100 (NOT the 500 list cap) — the live
+      // API 400s on limit > 100.
+      query: { term: 'Acme', exact_match: true, fields: 'name', limit: 100 },
+    })
+    expect(client.calls[1].path).toBe('/api/v2/organizations/5')
+  })
+
+  it('pages through next_cursor before fetching candidates', async () => {
+    const client = fakeClient({
+      searchPages: [page([1], 'CUR'), page([2])],
+      records: { 1: { id: 1, name: 'Acme' }, 2: { id: 2, name: 'Acme' } },
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'org',
+      field: 'name',
+      value: 'Acme',
+    })
+    expect(r.status).toBe('ambiguous')
+    const searches = client.calls.filter((c) => c.path.endsWith('/search'))
+    expect(searches).toHaveLength(2)
+    expect(searches[1].query.cursor).toBe('CUR')
+  })
+
+  it('de-duplicates a candidate id that the search repeats across pages', async () => {
+    const client = fakeClient({
+      searchPages: [page([7], 'CUR'), page([7])],
+      records: { 7: { id: 7, name: 'Acme' } },
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'org',
+      field: 'name',
+      value: 'Acme',
+    })
+    expect(r).toMatchObject({ status: 'unique', id: 7 })
+    const fetches = client.calls.filter(
+      (c) => c.path === '/api/v2/organizations/7',
+    )
+    expect(fetches).toHaveLength(1)
+  })
+
+  it('matches a searchable custom field by name, verifying the hash key', async () => {
+    const defs = [
+      {
+        field_name: 'External ID',
+        field_code: 'abc123',
+        field_type: 'varchar',
+      },
+    ]
+    const client = fakeClient({
+      searchPages: [page([5, 6])],
+      records: {
+        5: { id: 5, custom_fields: { abc123: 'D-42' } },
+        6: { id: 6, custom_fields: { abc123: 'D-99' } }, // verify drops it
+      },
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'deal',
+      defs,
+      field: 'External ID',
+      value: 'D-42',
+    })
+    expect(r).toMatchObject({ status: 'unique', id: 5 })
+    expect(client.calls[0].query.fields).toBe('custom_fields')
+  })
+
+  it('coerces numeric custom-field values for comparison', async () => {
+    const defs = [
+      { field_name: 'Score', field_code: 'sc', field_type: 'double' },
+    ]
+    const client = fakeClient({
+      searchPages: [page([5])],
+      records: { 5: { id: 5, custom_fields: { sc: 42 } } },
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'deal',
+      defs,
+      field: 'Score',
+      value: '42',
+    })
+    expect(r).toMatchObject({ status: 'unique', id: 5 })
+  })
+
+  it('matches a deal by title', async () => {
+    const client = fakeClient({
+      searchPages: [page([3])],
+      records: { 3: { id: 3, title: 'Acme expansion' } },
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'deal',
+      field: 'title',
+      value: 'Acme expansion',
+    })
+    expect(r).toMatchObject({ status: 'unique', id: 3 })
+    expect(client.calls[0].query.fields).toBe('title')
+  })
+
+  it('matches a person by name or phone (with empty-array tolerance)', async () => {
+    const byName = await lookupByField({
+      client: fakeClient({
+        searchPages: [page([1])],
+        records: { 1: { id: 1, name: 'Jane Doe', phones: [] } },
+      }),
+      entity: 'person',
+      field: 'name',
+      value: 'Jane Doe',
+    })
+    expect(byName).toMatchObject({ status: 'unique', id: 1 })
+
+    const byPhone = await lookupByField({
+      client: fakeClient({
+        searchPages: [page([2])],
+        records: { 2: { id: 2, phones: [{ value: '+15551234' }], emails: [] } },
+      }),
+      entity: 'person',
+      field: 'phone',
+      value: '+15551234',
+    })
+    expect(byPhone).toMatchObject({ status: 'unique', id: 2 })
+  })
+
+  it('accepts a custom field referenced by its hash code', async () => {
+    const defs = [
+      {
+        field_name: 'External ID',
+        field_code: 'abc123',
+        field_type: 'varchar',
+      },
+    ]
+    const client = fakeClient({
+      searchPages: [page([5])],
+      records: { 5: { id: 5, custom_fields: { abc123: 'D-42' } } },
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'deal',
+      defs,
+      field: 'abc123',
+      value: 'D-42',
+    })
+    expect(r).toMatchObject({ status: 'unique', id: 5 })
+  })
+
+  it('tolerates a malformed search response (no data/cursor)', async () => {
+    const client = fakeClient({ searchPages: [{}] })
+    const r = await lookupByField({
+      client,
+      entity: 'org',
+      field: 'name',
+      value: 'Acme',
+    })
+    expect(r.status).toBe('none')
+  })
+
+  it('tolerates a candidate whose fetched record is null', async () => {
+    const client = fakeClient({ searchPages: [page([1])], records: {} })
+    const r = await lookupByField({
+      client,
+      entity: 'org',
+      field: 'name',
+      value: 'Acme',
+    })
+    expect(r.status).toBe('none')
+  })
+
+  it('skips a search item without an id', async () => {
+    const client = fakeClient({
+      searchPages: [{ data: { items: [{ item: {} }] } }],
+    })
+    const r = await lookupByField({
+      client,
+      entity: 'org',
+      field: 'name',
+      value: 'Acme',
+    })
+    expect(r.status).toBe('none')
+  })
+
+  it('tolerates fetched records missing the matched field', async () => {
+    const noEmail = await lookupByField({
+      client: fakeClient({
+        searchPages: [page([1])],
+        records: { 1: { id: 1 } },
+      }),
+      entity: 'person',
+      field: 'email',
+      value: 'a@x.com',
+    })
+    expect(noEmail.status).toBe('none')
+
+    const noCustom = await lookupByField({
+      client: fakeClient({
+        searchPages: [page([2])],
+        records: { 2: { id: 2 } },
+      }),
+      entity: 'deal',
+      defs: [
+        { field_name: 'External ID', field_code: 'k', field_type: 'varchar' },
+      ],
+      field: 'External ID',
+      value: 'D-1',
+    })
+    expect(noCustom.status).toBe('none')
+
+    const noPhone = await lookupByField({
+      client: fakeClient({
+        searchPages: [page([3])],
+        records: { 3: { id: 3 } },
+      }),
+      entity: 'person',
+      field: 'phone',
+      value: '+1555',
+    })
+    expect(noPhone.status).toBe('none')
+  })
+
+  it('rejects a non-numeric value for a numeric custom field with exit 65', async () => {
+    const defs = [
+      { field_name: 'Score', field_code: 'sc', field_type: 'double' },
+    ]
+    const err = await lookupByField({
+      client: fakeClient({}),
+      entity: 'deal',
+      defs,
+      field: 'Score',
+      value: 'notanumber',
+    }).catch((e) => e)
+    expect(err.exitCode).toBe(65)
+    expect(err.message).toMatch(/number/i)
+  })
+
+  it('rejects a non-searchable field type with exit 64', async () => {
+    const defs = [
+      { field_name: 'Stage Color', field_code: 'cc', field_type: 'enum' },
+    ]
+    const err = await lookupByField({
+      client: fakeClient({}),
+      entity: 'deal',
+      defs,
+      field: 'Stage Color',
+      value: 'Red',
+    }).catch((e) => e)
+    expect(err.exitCode).toBe(64)
+    expect(err.message).toMatch(/not searchable/i)
+  })
+
+  it('rejects an unknown field with exit 64', async () => {
+    const err = await lookupByField({
+      client: fakeClient({}),
+      entity: 'deal',
+      defs: [],
+      field: 'Nonexistent',
+      value: 'x',
+    }).catch((e) => e)
+    expect(err.exitCode).toBe(64)
+    expect(err.message).toMatch(/unknown field/i)
+  })
+})
diff --git a/test/lib/upsert.test.js b/test/lib/upsert.test.js
new file mode 100644
index 0000000..f2402c8
--- /dev/null
+++ b/test/lib/upsert.test.js
@@ -0,0 +1,500 @@
+import { describe, it, expect } from 'vitest'
+import {
+  diffBody,
+  runUpsert,
+  summarizeUpsert,
+  bulkUpsertRows,
+} from '../../src/lib/upsert.js'
+
+describe('diffBody', () => {
+  it('returns only the changed top-level fields', () => {
+    expect(
+      diffBody({ title: 'B', value: 100 }, { title: 'A', value: 100 }),
+    ).toEqual({
+      title: 'B',
+    })
+  })
+
+  it('diffs nested custom_fields key by key', () => {
+    expect(
+      diffBody(
+        { custom_fields: { a: 1, b: 2 } },
+        { custom_fields: { a: 1, b: 9 } },
+      ),
+    ).toEqual({ custom_fields: { b: 2 } })
+  })
+
+  it('includes a field the existing record lacks', () => {
+    expect(diffBody({ value: 5 }, {})).toEqual({ value: 5 })
+  })
+
+  it('returns empty when nothing changed (key-order insensitive)', () => {
+    expect(
+      diffBody({ meta: { x: 1, y: 2 } }, { meta: { y: 2, x: 1 } }),
+    ).toEqual({})
+  })
+
+  it('omits custom_fields entirely when no custom value changed', () => {
+    expect(
+      diffBody({ custom_fields: { a: 1 } }, { custom_fields: { a: 1 } }),
+    ).toEqual({})
+  })
+
+  it('treats emails as unchanged when the value set matches (ignores primary, case)', () => {
+    expect(
+      diffBody(
+        { emails: [{ value: 'A@x.com', primary: true }] },
+        { emails: [{ value: 'a@x.com', label: 'work' }] },
+      ),
+    ).toEqual({})
+  })
+
+  it('detects a genuinely changed email set', () => {
+    expect(
+      diffBody(
+        { emails: [{ value: 'b@x.com', primary: true }] },
+        { emails: [{ value: 'a@x.com' }] },
+      ),
+    ).toEqual({ emails: [{ value: 'b@x.com', primary: true }] })
+  })
+
+  it('treats phones as unchanged when the value set matches', () => {
+    expect(
+      diffBody(
+        { phones: [{ value: '+15551234', primary: true }] },
+        { phones: [{ value: '+15551234' }] },
+      ),
+    ).toEqual({})
+  })
+
+  it('emits emails when the existing record has none', () => {
+    expect(diffBody({ emails: [{ value: 'a@x.com' }] }, {})).toEqual({
+      emails: [{ value: 'a@x.com' }],
+    })
+  })
+
+  it('treats a reordered primitive array (label_ids) as unchanged', () => {
+    expect(
+      diffBody({ label_ids: [2, 1, 3] }, { label_ids: [1, 2, 3] }),
+    ).toEqual({})
+  })
+
+  it('detects an added id in a primitive array', () => {
+    expect(diffBody({ label_ids: [1, 2, 3] }, { label_ids: [1, 2] })).toEqual({
+      label_ids: [1, 2, 3],
+    })
+  })
+
+  it('treats a reordered multi-option custom field as unchanged', () => {
+    expect(
+      diffBody(
+        { custom_fields: { k: [2, 1] } },
+        { custom_fields: { k: [1, 2] } },
+      ),
+    ).toEqual({})
+  })
+
+  it('tolerates null / value-less email entries when comparing sets', () => {
+    expect(
+      diffBody({ emails: [{ value: 'a@x.com' }] }, { emails: [null, {}] }),
+    ).toEqual({ emails: [{ value: 'a@x.com' }] })
+  })
+})
+
+/**
+ * Fake client modelling lookup's two-step flow: `/search` advertises candidate
+ * ids, then a record-fetch returns the full record. `items` are full records.
+ */
+function fakeClient({ items = [] } = {}) {
+  const calls = { post: [], patch: [] }
+  const byId = new Map(items.map((it) => [it.id, it]))
+  return {
+    calls,
+    async get(path) {
+      if (path.endsWith('/search')) {
+        return {
+          data: { items: items.map((it) => ({ item: { id: it.id } })) },
+          additional_data: { next_cursor: null },
+        }
+      }
+      const id = Number(path.split('/').pop())
+      return { data: byId.get(id) ?? null }
+    },
+    async post(path, opts) {
+      calls.post.push({ path, body: opts.body })
+      return { data: { id: 99, ...opts.body } }
+    },
+    async patch(path, opts) {
+      calls.patch.push({ path, body: opts.body })
+      return { data: { id: 7, ...opts.body } }
+    },
+  }
+}
+
+describe('runUpsert', () => {
+  it('creates when no match, injecting the match field into the body', async () => {
+    const client = fakeClient({ items: [] })
+    const r = await runUpsert({
+      client,
+      entity: 'person',
+      by: 'email',
+      value: 'a@x.com',
+      body: {},
+    })
+    expect(r).toMatchObject({ action: 'created', id: 99 })
+    expect(client.calls.post[0]).toMatchObject({
+      path: '/api/v2/persons',
+      body: { emails: [{ value: 'a@x.com', primary: true }] },
+    })
+  })
+
+  it('updates only the changed fields when exactly one matches', async () => {
+    const client = fakeClient({
+      items: [
+        { id: 7, emails: [{ value: 'a@x.com' }], custom_fields: { k: 6 } },
+      ],
+    })
+    const r = await runUpsert({
+      client,
+      entity: 'person',
+      by: 'email',
+      value: 'a@x.com',
+      body: { custom_fields: { k: 5 } },
+    })
+    expect(r).toMatchObject({ action: 'updated', id: 7 })
+    expect(client.calls.patch[0]).toMatchObject({
+      path: '/api/v2/persons/7',
+      body: { custom_fields: { k: 5 } },
+    })
+  })
+
+  it('reports unchanged and issues no PATCH when nothing differs', async () => {
+    const client = fakeClient({
+      items: [
+        { id: 7, emails: [{ value: 'a@x.com' }], custom_fields: { k: 5 } },
+      ],
+    })
+    const r = await runUpsert({
+      client,
+      entity: 'person',
+      by: 'email',
+      value: 'a@x.com',
+      body: { custom_fields: { k: 5 } },
+    })
+    expect(r).toMatchObject({ action: 'unchanged', id: 7 })
+    expect(client.calls.patch).toHaveLength(0)
+  })
+
+  it('refuses an ambiguous match with exit 65', async () => {
+    const client = fakeClient({
+      items: [
+        { id: 7, emails: [{ value: 'a@x.com' }] },
+        { id: 8, emails: [{ value: 'a@x.com' }] },
+      ],
+    })
+    const err = await runUpsert({
+      client,
+      entity: 'person',
+      by: 'email',
+      value: 'a@x.com',
+      body: {},
+    }).catch((e) => e)
+    expect(err.exitCode).toBe(65)
+    expect(err.message).toMatch(/7.*8|8.*7/)
+  })
+
+  it('dry-run create writes nothing', async () => {
+    const client = fakeClient({ items: [] })
+    const r = await runUpsert({
+      client,
+      entity: 'org',
+      by: 'name',
+      value: 'Acme',
+      body: {},
+      dryRun: true,
+    })
+    expect(r).toMatchObject({ action: 'created', dryRun: true })
+    expect(client.calls.post).toHaveLength(0)
+  })
+
+  it('dry-run update writes nothing but reports the change set', async () => {
+    const client = fakeClient({ items: [{ id: 3, title: 'Acme', value: 1 }] })
+    const r = await runUpsert({
+      client,
+      entity: 'deal',
+      by: 'title',
+      value: 'Acme',
+      body: { value: 2 },
+      dryRun: true,
+    })
+    expect(r).toMatchObject({ action: 'updated', id: 3, dryRun: true })
+    expect(r.changed).toEqual({ value: 2 })
+    expect(client.calls.patch).toHaveLength(0)
+  })
+
+  it('injects a custom match field into the create body', async () => {
+    const defs = [
+      { field_name: 'External ID', field_code: 'ext', field_type: 'varchar' },
+    ]
+    const client = fakeClient({ items: [] })
+    await runUpsert({
+      client,
+      entity: 'deal',
+      by: 'External ID',
+      value: 'D-42',
+      body: {},
+      defs,
+    })
+    expect(client.calls.post[0].body.custom_fields).toEqual({ ext: 'D-42' })
+  })
+
+  it('injects a phone match field when creating a person by phone', async () => {
+    const client = fakeClient({ items: [] })
+    await runUpsert({
+      client,
+      entity: 'person',
+      by: 'phone',
+      value: '+15551234',
+      body: {},
+    })
+    expect(client.calls.post[0].body.phones).toEqual([
+      { value: '+15551234', primary: true },
+    ])
+  })
+
+  it('injects a title match field when creating a deal by title', async () => {
+    const client = fakeClient({ items: [] })
+    await runUpsert({
+      client,
+      entity: 'deal',
+      by: 'title',
+      value: 'Acme expansion',
+      body: {},
+    })
+    expect(client.calls.post[0].body.title).toBe('Acme expansion')
+  })
+
+  it('injects a numeric custom match field, preserving other custom fields', async () => {
+    const defs = [
+      { field_name: 'Score', field_code: 'sc', field_type: 'double' },
+      { field_name: 'Region', field_code: 'rg', field_type: 'varchar' },
+    ]
+    const client = fakeClient({ items: [] })
+    await runUpsert({
+      client,
+      entity: 'deal',
+      by: 'Score',
+      value: '42',
+      body: { custom_fields: { rg: 'EMEA' } },
+      defs,
+    })
+    expect(client.calls.post[0].body.custom_fields).toEqual({
+      rg: 'EMEA',
+      sc: 42,
+    })
+  })
+
+  it('does not clobber an explicit body value with the injected match field', async () => {
+    const client = fakeClient({ items: [] })
+    await runUpsert({
+      client,
+      entity: 'org',
+      by: 'name',
+      value: 'Acme',
+      body: { name: 'Acme Corporation' }, // explicit wins
+    })
+    expect(client.calls.post[0].body.name).toBe('Acme Corporation')
+  })
+
+  it('matches a custom field referenced by hash code (no field_name)', async () => {
+    const defs = [{ field_code: 'ext', field_type: 'varchar' }] // no field_name
+    const client = fakeClient({ items: [] })
+    await runUpsert({
+      client,
+      entity: 'deal',
+      by: 'ext',
+      value: 'D-1',
+      body: {},
+      defs,
+    })
+    expect(client.calls.post[0].body.custom_fields).toEqual({ ext: 'D-1' })
+  })
+
+  it('does not clobber an explicit custom match value', async () => {
+    const defs = [
+      { field_name: 'External ID', field_code: 'ext', field_type: 'varchar' },
+    ]
+    const client = fakeClient({ items: [] })
+    await runUpsert({
+      client,
+      entity: 'deal',
+      by: 'External ID',
+      value: 'D-42',
+      body: { custom_fields: { ext: 'EXPLICIT' } },
+      defs,
+    })
+    expect(client.calls.post[0].body.custom_fields.ext).toBe('EXPLICIT')
+  })
+})
+
+describe('summarizeUpsert', () => {
+  it('counts changed fields, pluralizing past one', () => {
+    expect(
+      summarizeUpsert({ action: 'updated', id: 7, changed: { a: 1 } }, 'deal'),
+    ).toBe('update deal #7 (1 field)')
+    expect(
+      summarizeUpsert(
+        { action: 'updated', id: 7, changed: { a: 1, b: 2 } },
+        'deal',
+      ),
+    ).toBe('update deal #7 (2 fields)')
+  })
+
+  it('treats an updated result with no change set as zero fields', () => {
+    expect(summarizeUpsert({ action: 'updated', id: 7 }, 'deal')).toBe(
+      'update deal #7 (0 fields)',
+    )
+  })
+
+  it('prefixes dry-run and omits the id when a created record has none', () => {
+    expect(summarizeUpsert({ action: 'created', dryRun: true }, 'org')).toBe(
+      '[dry-run] would create org',
+    )
+    expect(summarizeUpsert({ action: 'created', id: 5 }, 'org')).toBe(
+      'create org #5',
+    )
+  })
+
+  it('reports an unchanged record', () => {
+    expect(summarizeUpsert({ action: 'unchanged', id: 7 }, 'person')).toBe(
+      'person #7 unchanged',
+    )
+  })
+})
+
+/**
+ * Fake client whose search candidates vary by the search `term`; a record
+ * fetch then returns the full record by id (lookup's two-step flow).
+ */
+function termFakeClient(byTerm = {}) {
+  const calls = { post: [], patch: [], terms: [] }
+  const byId = new Map()
+  for (const recs of Object.values(byTerm)) {
+    for (const r of recs) byId.set(r.id, r)
+  }
+  return {
+    calls,
+    async get(path, opts) {
+      if (path.endsWith('/search')) {
+        calls.terms.push(opts.query.term)
+        const items = byTerm[opts.query.term] ?? []
+        return {
+          data: { items: items.map((r) => ({ item: { id: r.id } })) },
+          additional_data: { next_cursor: null },
+        }
+      }
+      const id = Number(path.split('/').pop())
+      return { data: byId.get(id) ?? null }
+    },
+    async post(path, opts) {
+      calls.post.push({ path, body: opts.body })
+      return { data: { id: 99, ...opts.body } }
+    },
+    async patch(path, opts) {
+      calls.patch.push({ path, body: opts.body })
+      return { data: { ...opts.body } }
+    },
+  }
+}
+
+describe('bulkUpsertRows', () => {
+  it('creates new rows, patches existing ones, and tallies actions', async () => {
+    const client = termFakeClient({
+      'a@x.com': [{ id: 7, emails: [{ value: 'a@x.com' }], owner_id: 1 }],
+      'b@x.com': [],
+    })
+    const r = await bulkUpsertRows({
+      client,
+      entity: 'person',
+      matchOn: 'email',
+      rows: [
+        { body: { owner_id: 42 }, value: 'a@x.com' },
+        { body: { name: 'Bob' }, value: 'b@x.com' },
+      ],
+      defs: [],
+      gapMs: 0,
+    })
+    expect(r.counts).toEqual({ created: 1, updated: 1, unchanged: 0 })
+    expect(client.calls.patch).toHaveLength(1)
+    expect(client.calls.post).toHaveLength(1)
+  })
+
+  it('counts an already-matching row as unchanged', async () => {
+    const client = termFakeClient({
+      'a@x.com': [{ id: 7, emails: [{ value: 'a@x.com' }], owner_id: 1 }],
+    })
+    const r = await bulkUpsertRows({
+      client,
+      entity: 'person',
+      matchOn: 'email',
+      rows: [{ body: { owner_id: 1 }, value: 'a@x.com' }],
+      defs: [],
+      gapMs: 0,
+    })
+    expect(r.counts).toEqual({ created: 0, updated: 0, unchanged: 1 })
+    expect(client.calls.patch).toHaveLength(0)
+  })
+
+  it('collects an empty match value as a failed row', async () => {
+    const client = termFakeClient({})
+    const r = await bulkUpsertRows({
+      client,
+      entity: 'person',
+      matchOn: 'email',
+      rows: [{ body: {}, value: '' }],
+      defs: [],
+      gapMs: 0,
+    })
+    expect(r.failed).toHaveLength(1)
+    expect(r.failed[0].error).toMatch(/empty|match/i)
+    expect(client.calls.post).toHaveLength(0)
+  })
+
+  it('collects an ambiguous match as a failed row without aborting', async () => {
+    const client = termFakeClient({
+      dup: [
+        { id: 1, name: 'dup' },
+        { id: 2, name: 'dup' },
+      ],
+      'ok@x.com': [],
+    })
+    const r = await bulkUpsertRows({
+      client,
+      entity: 'org',
+      matchOn: 'name',
+      rows: [
+        { body: {}, value: 'dup' },
+        { body: { name: 'ok@x.com' }, value: 'ok@x.com' },
+      ],
+      defs: [],
+      gapMs: 0,
+    })
+    expect(r.failed).toHaveLength(1)
+    expect(r.counts.created).toBe(1)
+  })
+
+  it('looks up but never writes under dry-run', async () => {
+    const client = termFakeClient({ 'a@x.com': [] })
+    const r = await bulkUpsertRows({
+      client,
+      entity: 'person',
+      matchOn: 'email',
+      rows: [{ body: {}, value: 'a@x.com' }],
+      defs: [],
+      dryRun: true,
+      gapMs: 0,
+    })
+    expect(r.counts.created).toBe(1)
+    expect(client.calls.post).toHaveLength(0)
+  })
+})
diff --git a/website/src/content/docs/guides/bulk.mdx b/website/src/content/docs/guides/bulk.mdx
index 074814c..ea5c18a 100644
--- a/website/src/content/docs/guides/bulk.mdx
+++ b/website/src/content/docs/guides/bulk.mdx
@@ -101,6 +101,8 @@ each deal succeeds or fails on its own.
 
 `person import <file>` and `org import <file>` bulk-create records from a CSV. The first
 row is the header; **a `name` column is required** (case-insensitive), or it's exit 64.
+Duplicate column headers are rejected (exit 65) rather than silently keeping only the last
+cell.
 
 Each header maps to a field. Recognized special columns build standard values directly;
 every other header is resolved through the entity's field definitions — by human name,
@@ -145,3 +147,77 @@ Imported 23/24 persons
   ✘ John Roe: Pipedrive API 422: email already in use
 Error: 1 of 24 rows failed
 ```
+
+## Idempotent upsert (match-or-create)
+
+`person upsert`, `org upsert`, and `deal upsert` make a write **safe to re-run**: match an
+existing record, then **create** it if absent or **PATCH only the fields that changed** if
+exactly one matches. Re-running the same command a second time is a no-op.
+
+```bash
+pdcli person upsert a@x.com --by email --field "Tier=Gold"
+pdcli deal upsert "Acme expansion" --by title --body '{"value":5000}'
+pdcli org upsert "D-42" --by "External ID" --field "Status=Active"
+```
+
+The match value is the positional argument; `--by` names the field to match on — a built-in
+key or a **searchable** custom field (`address`, `varchar`, `text`, `double`, `monetary`,
+`phone`):
+
+| Entity | Built-in `--by` keys  |
+| ------ | --------------------- |
+| person | `email`, `name`, `phone` |
+| org    | `name`                |
+| deal   | `title`               |
+
+### Why it refuses instead of guessing
+
+Pipedrive's `exact_match` search is **not** a unique-key lookup — it's case-insensitive, and
+a custom-field search scans every custom field at once. So pdcli re-verifies each candidate
+client-side and counts the survivors:
+
+- **0 matches** → create (the match value is injected into the new record).
+- **1 match** → PATCH only the differing fields; if nothing differs, it reports `unchanged`
+  and issues no write.
+- **more than 1** → **refuse with exit 65**, listing the colliding IDs. It never picks one.
+
+```text
+Error: --by email="a@x.com" matches 2 person records (ids: 7, 12) — refusing to guess. Narrow the match value.
+```
+
+`--dry-run` previews the action without writing. Table output prints a one-line summary
+(`update person #7 (2 fields)`); `--output json` emits the full action result
+(`action`, `id`, `changed`).
+
+### Upserting a whole CSV
+
+`person import` and `org import` gain `--upsert --match-on <field>`: each row is matched on
+its value in the `--match-on` column, then created or PATCHed like the single-record upsert.
+
+```bash
+pdcli person import contacts.csv --upsert --match-on email --dry-run   # preview the counts
+pdcli person import contacts.csv --upsert --match-on email             # apply
+```
+
+`--match-on` is required with `--upsert` and must name a column present in the CSV (else exit
+64). Rows are paced like the other bulk flows; per-row failures — an ambiguous match or an
+empty match cell — are collected without aborting the batch and summarized by action:
+
+```text
+18 created, 5 updated, 1 unchanged
+  ✘ email="dup@x.com": --by email="dup@x.com" matches 2 person records (ids: 7, 12) — refusing to guess. Narrow the match value.
+Error: 1 of 24 rows failed
+```
+
+When every failed row is a data-validation error (ambiguous match, empty match value) the
+command exits **65**; a mix that includes a transport/API error exits **1**.
+
+:::caution
+Matching relies on Pipedrive's **search index, which is eventually consistent** — a
+just-created record is not searchable for a short window. So upserting the _same_ match
+value twice in quick succession (the same key on two CSV rows, or an upsert moments after a
+create) can still create a duplicate, because the second lookup runs before the first write
+is indexed. For repeatable sync, key on a stable external ID and let the index settle
+between runs; a periodic `backup diff` or a search will surface any duplicates that slip
+through.
+:::
diff --git a/website/src/content/docs/reference/commands.mdx b/website/src/content/docs/reference/commands.mdx
index fcfd476..57ec8b8 100644
--- a/website/src/content/docs/reference/commands.mdx
+++ b/website/src/content/docs/reference/commands.mdx
@@ -5,7 +5,7 @@ description: Every pdcli command, flag, and example — generated from the CLI m
 
 {/* AUTO-GENERATED from the oclif manifest by scripts/gen-commands.mjs — do not edit by hand. */}
 
-All 142 commands in `pdcli` v0.15.0. Every command also
+All 145 commands in `pdcli` v0.16.0. Every command also
 accepts the [global flags](/pdcli/reference/config/) `--output`, `--jq`,
 `--fields`, `--profile`, `--limit`, `--no-color`, `--verbose`,
 `--no-retry`, and `--timeout`. Run `pdcli <command> --help` for the live version.
@@ -929,6 +929,27 @@ pdcli deal update 42 --status won
 pdcli deal update 42 --field "Deal Size=Large"
 ```
 
+### `pdcli deal upsert`
+
+Idempotent deal upsert: match by --by, then create or PATCH only the changed fields. Refuses (exit 65) if more than one record matches.
+
+```text
+pdcli deal upsert <value> [flags]
+```
+
+| Flag | Description |
+| --- | --- |
+| `--by` &lt;value&gt; | Match field: title, or a searchable custom field |
+| `--field` &lt;value&gt; | Field to set as "Name=Value" (repeatable) |
+| `--body` &lt;value&gt; | Raw JSON body to merge |
+| `--dry-run` | Preview the action without writing |
+
+```bash
+pdcli deal upsert "Acme expansion" --by title --field "Stage=Won"
+pdcli deal upsert "D-42" --by "External ID" --body '{"value":5000}'
+pdcli deal upsert "Acme expansion" --by title --field "Stage=Won" --dry-run
+```
+
 ## pdcli field
 
 ### `pdcli field create`
@@ -1749,6 +1770,8 @@ pdcli org import <file> [flags]
 
 | Flag | Description |
 | --- | --- |
+| `--upsert` | Match each row on --match-on, then create or update |
+| `--match-on` &lt;value&gt; | Field to match rows on in --upsert mode (e.g. name) |
 | `--dry-run` | Validate every row without creating anything |
 | `-y, --yes` | Skip the confirmation prompt |
 
@@ -1872,6 +1895,27 @@ pdcli org update 7 --owner 9
 pdcli org update 7 --field "Tier=Gold"
 ```
 
+### `pdcli org upsert`
+
+Idempotent organization upsert: match by --by, then create or PATCH only the changed fields. Refuses (exit 65) if more than one record matches.
+
+```text
+pdcli org upsert <value> [flags]
+```
+
+| Flag | Description |
+| --- | --- |
+| `--by` &lt;value&gt; | Match field: name, or a searchable custom field |
+| `--field` &lt;value&gt; | Field to set as "Name=Value" (repeatable) |
+| `--body` &lt;value&gt; | Raw JSON body to merge |
+| `--dry-run` | Preview the action without writing |
+
+```bash
+pdcli org upsert Acme --by name --field "Tier=Gold"
+pdcli org upsert "D-42" --by "External ID" --body '{"owner_id":42}'
+pdcli org upsert Acme --by name --field "Tier=Gold" --dry-run
+```
+
 ## pdcli person
 
 ### `pdcli person create`
@@ -1986,6 +2030,8 @@ pdcli person import <file> [flags]
 
 | Flag | Description |
 | --- | --- |
+| `--upsert` | Match each row on --match-on, then create or update |
+| `--match-on` &lt;value&gt; | Field to match rows on in --upsert mode (e.g. email) |
 | `--dry-run` | Validate every row without creating anything |
 | `-y, --yes` | Skip the confirmation prompt |
 
@@ -2060,6 +2106,27 @@ pdcli person update 42 --email new@acme.com
 pdcli person update 42 --field "Segment=Enterprise"
 ```
 
+### `pdcli person upsert`
+
+Idempotent person upsert: match by --by, then create or PATCH only the changed fields. Refuses (exit 65) if more than one record matches.
+
+```text
+pdcli person upsert <value> [flags]
+```
+
+| Flag | Description |
+| --- | --- |
+| `--by` &lt;value&gt; | Match field: email, name, phone, or a searchable custom field |
+| `--field` &lt;value&gt; | Field to set as "Name=Value" (repeatable) |
+| `--body` &lt;value&gt; | Raw JSON body to merge |
+| `--dry-run` | Preview the action without writing |
+
+```bash
+pdcli person upsert a@x.com --by email --field "Tier=Gold"
+pdcli person upsert "Jane Doe" --by name --body '{"owner_id":42}'
+pdcli person upsert a@x.com --by email --field "Tier=Gold" --dry-run
+```
+
 ## pdcli pipeline
 
 ### `pdcli pipeline get`
diff --git a/website/src/content/docs/start/agents.mdx b/website/src/content/docs/start/agents.mdx
index 94b7a34..97187ee 100644
--- a/website/src/content/docs/start/agents.mdx
+++ b/website/src/content/docs/start/agents.mdx
@@ -92,6 +92,13 @@ an exit-code-gated anomaly poller that fires only on findings new since the last
 (`pdcli watch || notify`), `sync warehouse` for an incremental NDJSON export with per-entity
 high-water marks, and `--updated-since` on the list commands for incremental polling.
 
+For writes that must be safe to retry, `person`/`org`/`deal upsert` match a record by `--by`
+(a built-in key or a searchable custom field) and then create or PATCH only what changed —
+and **refuse with exit 65** when more than one record matches, so an agent never silently
+writes the wrong one. `person import`/`org import --upsert --match-on <field>` apply the same
+match-or-create per CSV row, reporting created/updated/unchanged counts. Pair upsert with an
+external key (a custom field carrying your system's ID) for clean, repeatable sync.
+
 ## The host-locked `api` escape hatch
 
 When no dedicated command exists, call any endpoint directly. The request is
diff --git a/website/src/data/cli-stats.json b/website/src/data/cli-stats.json
index 247efc2..c868b3c 100644
--- a/website/src/data/cli-stats.json
+++ b/website/src/data/cli-stats.json
@@ -1,6 +1,6 @@
 {
-  "version": "0.15.0",
-  "commands": 142,
+  "version": "0.16.0",
+  "commands": 145,
   "topics": 26,
   "formats": 4
 }