diff --git a/extensions/general/arcim-migration/lib/__tests__/complete-invoice-lines.test.ts b/extensions/general/arcim-migration/lib/__tests__/complete-invoice-lines.test.ts
index f186bd17..41b8379a 100644
--- a/extensions/general/arcim-migration/lib/__tests__/complete-invoice-lines.test.ts
+++ b/extensions/general/arcim-migration/lib/__tests__/complete-invoice-lines.test.ts
@@ -8,7 +8,9 @@ import type { SalesInvoiceDto } from '@/lib/providers/dto'
  * migration's detail hydration is budget-bounded). It must start from OUR
  * row-less invoices, join strictly, hydrate only that subset, write rows only
  * when the provider's total agrees with the stored one, and leave anything it
- * could not reach for the next run rather than guessing.
+ * could not reach for the next run rather than guessing. Every write goes
+ * through the complete_invoice_rows RPC, one call per invoice, which is what
+ * keeps two writers from doubling an invoice's rows.
  */
 
 vi.mock('@/lib/providers/resolve-consent', () => ({
@@ -102,11 +104,33 @@ function hydratedAll(invoices: SalesInvoiceDto[], unhydratedIds: string[] = [])
 
 interface Call { table: string; method: string; args: unknown[] }
 
+/** The complete_invoice_rows payload, as the pass sends it. */
+interface RpcArgs {
+  p_company_id: string
+  p_invoice_id: string
+  p_rows: Record<string, unknown>[]
+  p_header: Record<string, unknown> | null
+}
+
+/** What the real RPC answers a caller that wrote: N rows, header iff one was sent. */
+function rpcWrote(args: RpcArgs) {
+  return {
+    data: { ok: true, wrote: true, rows: args.p_rows.length, header_updated: args.p_header !== null },
+    error: null,
+  }
+}
+
+const rpcAlreadyFilled = { data: { ok: true, wrote: false, rows: 0, header_updated: false }, error: null }
+
 /**
- * Thenable query-builder stand-in. Records every call; resolves with what
- * `respond` returns for the table and the methods used on the chain.
+ * Query-builder and RPC stand-in. Records every call; `respond` answers the
+ * table chains, `respondRpc` answers `.rpc()` (defaults to a successful
+ * write shaped like the real function's reply).
  */
-function makeSupabase(respond: (table: string, methods: string[], calls: Call[]) => unknown) {
+function makeSupabase(
+  respond: (table: string, methods: string[], calls: Call[]) => unknown,
+  respondRpc: (fn: string, args: RpcArgs) => unknown = (_fn, args) => rpcWrote(args),
+) {
   const calls: Call[] = []
   const from = vi.fn((table: string) => {
     const chain: Call[] = []
@@ -125,21 +149,30 @@ function makeSupabase(respond: (table: string, methods: string[], calls: Call[])
         .then(resolve, reject)
     return builder
   })
-  return { supabase: { from } as unknown as SupabaseClient, calls }
+  const rpc = vi.fn((fn: string, args: RpcArgs) => {
+    calls.push({ table: `rpc:${fn}`, method: 'rpc', args: [fn, args] })
+    return Promise.resolve(respondRpc(fn, args))
+  })
+  return { supabase: { from, rpc } as unknown as SupabaseClient, calls }
 }
 
 const ok = { data: [], error: null }
 
-function insertedRows(calls: Call[]): Record<string, unknown>[] {
+function writes(calls: Call[]): RpcArgs[] {
   return calls
-    .filter((c) => c.table === 'invoice_items' && c.method === 'insert')
-    .flatMap((c) => c.args[0] as Record<string, unknown>[])
+    .filter((c) => c.method === 'rpc' && c.args[0] === 'complete_invoice_rows')
+    .map((c) => c.args[1] as RpcArgs)
+}
+
+/** The rows sent, each tagged with the invoice the call named. */
+function insertedRows(calls: Call[]): Record<string, unknown>[] {
+  return writes(calls).flatMap((w) => w.p_rows.map((row) => ({ ...row, invoice_id: w.p_invoice_id })))
 }
 
 function headerUpdates(calls: Call[]): Record<string, unknown>[] {
-  return calls
-    .filter((c) => c.table === 'invoices' && c.method === 'update')
-    .map((c) => c.args[0] as Record<string, unknown>)
+  return writes(calls)
+    .map((w) => w.p_header)
+    .filter((h): h is Record<string, unknown> => h !== null)
 }
 
 describe('completeMigratedInvoiceLines', () => {
@@ -148,7 +181,7 @@ describe('completeMigratedInvoiceLines', () => {
     mResolve.mockResolvedValue({ consent: { provider: 'fortnox' }, accessToken: 'tok', providerCompanyId: undefined })
   })
 
-  it('writes the rows and fills a header that held no VAT evidence', async () => {
+  it('writes the rows and fills a header that held no VAT evidence, through one RPC call per invoice', async () => {
     mFetchAll.mockResolvedValue([storedRow()])
     const dto = providerInvoice()
     mList.mockResolvedValue([dto])
@@ -166,10 +199,14 @@ describe('completeMigratedInvoiceLines', () => {
     expect(mHydrate).toHaveBeenCalledTimes(1)
     expect(mHydrate.mock.calls[0][3]).toEqual([dto])
 
-    const rows = insertedRows(calls)
-    expect(rows).toHaveLength(1)
-    expect(rows[0]).toMatchObject({
-      invoice_id: 'inv-1',
+    // One call, scoped to the company as well as the invoice: the RPC locks
+    // the invoice and stamps invoice_id itself, so the rows carry none.
+    const sent = writes(calls)
+    expect(sent).toHaveLength(1)
+    expect(sent[0]).toMatchObject({ p_company_id: 'co-1', p_invoice_id: 'inv-1' })
+    expect(sent[0].p_rows).toHaveLength(1)
+    expect(sent[0].p_rows[0]).not.toHaveProperty('invoice_id')
+    expect(sent[0].p_rows[0]).toEqual({
       sort_order: 1,
       description: 'Konsulttid',
       quantity: 10,
@@ -178,11 +215,9 @@ describe('completeMigratedInvoiceLines', () => {
       line_total: 1000,
       vat_rate: 25,
       vat_amount: 250,
+      line_type: 'product',
     })
-
-    const headers = headerUpdates(calls)
-    expect(headers).toHaveLength(1)
-    expect(headers[0]).toEqual({
+    expect(sent[0].p_header).toEqual({
       subtotal: 1000,
       subtotal_sek: 1000,
       vat_amount: 250,
@@ -190,14 +225,11 @@ describe('completeMigratedInvoiceLines', () => {
       vat_rate: 25,
       vat_treatment: 'standard_25',
     })
-    // Scoped to the company as well as the id: defense in depth on a
-    // service-role client.
-    const update = calls.find((c) => c.table === 'invoices' && c.method === 'update')!
-    const scope = calls.filter((c) => c.table === 'invoices' && c.method === 'eq' && calls.indexOf(c) > calls.indexOf(update))
-    expect(scope.map((c) => c.args)).toEqual([['id', 'inv-1'], ['company_id', 'co-1']])
     // Never the total, status or payments.
-    expect(Object.keys(headers[0])).not.toContain('total')
-    expect(Object.keys(headers[0])).not.toContain('status')
+    expect(Object.keys(sent[0].p_header!)).not.toContain('total')
+    expect(Object.keys(sent[0].p_header!)).not.toContain('status')
+    // Nothing is written outside the RPC.
+    expect(calls.filter((c) => c.method !== 'rpc')).toHaveLength(0)
   })
 
   it('writes the rows but leaves a header whose split is consistent (momsfri)', async () => {
@@ -211,6 +243,7 @@ describe('completeMigratedInvoiceLines', () => {
 
     expect(result).toMatchObject({ completed: 1, headersUpdated: 0 })
     expect(insertedRows(calls)).toHaveLength(1)
+    expect(writes(calls)[0].p_header).toBeNull()
     expect(headerUpdates(calls)).toHaveLength(0)
   })
 
@@ -256,8 +289,7 @@ describe('completeMigratedInvoiceLines', () => {
     const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' })
 
     expect(result).toMatchObject({ matched: 1, completed: 0, rowsMismatch: 1, remaining: 1 })
-    expect(insertedRows(calls)).toHaveLength(0)
-    expect(headerUpdates(calls)).toHaveLength(0)
+    expect(writes(calls)).toHaveLength(0)
   })
 
   it('tolerates öresavrundning between the rows and the header', async () => {
@@ -288,8 +320,7 @@ describe('completeMigratedInvoiceLines', () => {
     const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' })
 
     expect(result).toMatchObject({ matched: 1, completed: 0, totalMismatch: 1, remaining: 1 })
-    expect(insertedRows(calls)).toHaveLength(0)
-    expect(headerUpdates(calls)).toHaveLength(0)
+    expect(writes(calls)).toHaveLength(0)
   })
 
   it('reverses the rows of a kreditfaktura the way the migration does', async () => {
@@ -318,7 +349,7 @@ describe('completeMigratedInvoiceLines', () => {
     const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' })
 
     expect(result).toMatchObject({ candidates: 2, matched: 2, completed: 1, notHydrated: 1, remaining: 1 })
-    expect(insertedRows(calls).map((r) => r.invoice_id)).toEqual(['inv-1'])
+    expect(writes(calls).map((w) => w.p_invoice_id)).toEqual(['inv-1'])
   })
 
   it('does not join an ambiguous key, and does not hydrate when nothing joined', async () => {
@@ -332,7 +363,7 @@ describe('completeMigratedInvoiceLines', () => {
 
     expect(result).toMatchObject({ candidates: 2, matched: 0, unmatched: 2, completed: 0, remaining: 2 })
     expect(mHydrate).not.toHaveBeenCalled()
-    expect(insertedRows(calls)).toHaveLength(0)
+    expect(writes(calls)).toHaveLength(0)
   })
 
   it('costs one query and no provider call when the company has nothing to complete', async () => {
@@ -347,39 +378,49 @@ describe('completeMigratedInvoiceLines', () => {
     expect(calls).toHaveLength(0)
   })
 
-  it('skips an invoice that gained rows since the candidates were loaded', async () => {
+  it('counts an invoice that gained rows since the candidates were loaded as skipped, not completed', async () => {
+    // The RPC answers wrote = false: another writer (the wizard, an
+    // overlapping cron) got there first. The header in the payload must not
+    // be counted either; header_updated comes from the RPC, not the plan.
     mFetchAll.mockResolvedValue([storedRow()])
     const dto = providerInvoice()
     mList.mockResolvedValue([dto])
     mHydrate.mockResolvedValue(hydratedAll([dto]))
-    const { supabase, calls } = makeSupabase((table, methods) =>
-      table === 'invoice_items' && methods.includes('in') ? { data: [{ invoice_id: 'inv-1' }], error: null } : ok,
-    )
+    const { supabase, calls } = makeSupabase(() => ok, () => rpcAlreadyFilled)
 
     const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' })
 
-    expect(result).toMatchObject({ completed: 0, failed: 0, remaining: 1 })
-    expect(insertedRows(calls)).toHaveLength(0)
-    expect(headerUpdates(calls)).toHaveLength(0)
+    expect(result).toMatchObject({ completed: 0, headersUpdated: 0, failed: 0, remaining: 1 })
+    expect(writes(calls)).toHaveLength(1)
+    expect(writes(calls)[0].p_header).not.toBeNull()
   })
 
-  it('retries per invoice when the batch insert fails, and counts the offender', async () => {
-    mFetchAll.mockResolvedValue([storedRow(), storedRow({ id: 'inv-2', invoice_number: '1002' })])
-    const first = providerInvoice()
-    const second = providerInvoice({ id: '1002', invoiceNumber: '1002' })
-    mList.mockResolvedValue([first, second])
-    mHydrate.mockResolvedValue(hydratedAll([first, second]))
-    const { supabase, calls } = makeSupabase((table, methods, chain) => {
-      if (table !== 'invoice_items' || !methods.includes('insert')) return ok
-      const rows = chain[0].args[0] as { invoice_id: string }[]
-      if (rows.length > 1) return { data: null, error: { message: 'batch rejected' } }
-      return rows[0].invoice_id === 'inv-2' ? { data: null, error: { message: 'check violation' } } : ok
+  it('counts a refused or failed call as failed and goes on with the rest', async () => {
+    mFetchAll.mockResolvedValue([
+      storedRow(),
+      storedRow({ id: 'inv-2', invoice_number: '1002' }),
+      storedRow({ id: 'inv-3', invoice_number: '1003' }),
+    ])
+    const dtos = [
+      providerInvoice(),
+      providerInvoice({ id: '1002', invoiceNumber: '1002' }),
+      providerInvoice({ id: '1003', invoiceNumber: '1003' }),
+    ]
+    mList.mockResolvedValue(dtos)
+    mHydrate.mockResolvedValue(hydratedAll(dtos))
+    const { supabase, calls } = makeSupabase(() => ok, (_fn, args) => {
+      // A Postgres error (a check violation inside the RPC) and a refusal
+      // the function answers with ok = false: both leave the invoice for a
+      // human to look at, neither stops the run.
+      if (args.p_invoice_id === 'inv-2') return { data: null, error: { message: 'check violation' } }
+      if (args.p_invoice_id === 'inv-3') return { data: { ok: false, code: 'INVOICE_NOT_FOUND' }, error: null }
+      return rpcWrote(args)
     })
 
     const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' })
 
-    expect(result).toMatchObject({ completed: 1, failed: 1, headersUpdated: 1, remaining: 1 })
-    expect(headerUpdates(calls)).toHaveLength(1)
+    expect(result).toMatchObject({ completed: 1, failed: 2, headersUpdated: 1, remaining: 2 })
+    expect(writes(calls).map((w) => w.p_invoice_id)).toEqual(['inv-1', 'inv-2', 'inv-3'])
   })
 
   it('dry run: reports the plan and writes nothing', async () => {
diff --git a/extensions/general/arcim-migration/lib/complete-invoice-lines.ts b/extensions/general/arcim-migration/lib/complete-invoice-lines.ts
index 4c531fb3..1459a42e 100644
--- a/extensions/general/arcim-migration/lib/complete-invoice-lines.ts
+++ b/extensions/general/arcim-migration/lib/complete-invoice-lines.ts
@@ -29,13 +29,19 @@
  * momsdeklaration and every report read the ledger, not these columns, so
  * filling them changes what the invoice page shows and nothing that was
  * filed (see the 2026-08-22 verification in DECISIONS.md).
+ *
+ * Both are written by one call to the complete_invoice_rows RPC per invoice
+ * (migration 20260906135730), the write path the migration wizard shares: it
+ * locks the invoice, inserts the rows only when the invoice still has none
+ * and applies the header split in the same transaction, so an invoice's rows
+ * are written at most once whichever writer gets there first, and "rows
+ * landed, header did not" is not a reachable state.
  */
 
 import type { SupabaseClient } from '@supabase/supabase-js'
 import { ISO_DATE_RE } from '@/lib/invariants'
 import { createLogger } from '@/lib/logger'
 import { equalOre, roundOre } from '@/lib/money'
-import { chunk } from '@/lib/utils'
 import type { SalesInvoiceDto } from '@/lib/providers/dto'
 import type { ProviderName } from '@/lib/providers/types'
 import { resolveConsent } from '@/lib/providers/resolve-consent'
@@ -133,9 +139,6 @@ interface CandidateRow {
   invoice_items: { id: string }[] | null
 }
 
-/** Invoices per statement. Small enough that a chunk's rows stay one request. */
-const WRITE_CHUNK_SIZE = 100
-
 /**
  * How far the rows may disagree with the header before the invoice is left
  * alone. Öresavrundning puts up to 0.50 kr between a Fortnox `Total` and the
@@ -219,32 +222,35 @@ async function loadCandidates(supabase: SupabaseClient, companyId: string): Prom
   return rows.filter((row) => (row.invoice_items?.length ?? 0) === 0)
 }
 
-/** Ids among `ids` that gained rows since the candidates were loaded. */
-async function alreadyFilled(supabase: SupabaseClient, ids: string[]): Promise<Set<string>> {
-  const { data, error } = await supabase
-    .from('invoice_items')
-    .select('invoice_id')
-    .in('invoice_id', ids)
-  if (error) throw new Error(`invoice_items lookup failed: ${error.message}`)
-  return new Set(((data ?? []) as { invoice_id: string }[]).map((r) => r.invoice_id))
-}
-
-/** The header VAT split the detail form established, ready to write. */
+/**
+ * The header VAT split the detail form established, as the RPC's p_header:
+ * the six invoice columns it may rewrite, all present or none.
+ */
 interface HeaderFill {
   subtotal: number
-  subtotalSek: number | null
-  vatAmount: number
-  vatAmountSek: number | null
-  vatRate: number | null
-  vatTreatment: string
+  subtotal_sek: number | null
+  vat_amount: number
+  vat_amount_sek: number | null
+  vat_rate: number | null
+  vat_treatment: string
 }
 
 interface PlannedWrite {
   row: CandidateRow
+  /** The invoice_items columns per row; the RPC stamps invoice_id itself. */
   items: Record<string, unknown>[]
   header: HeaderFill | null
 }
 
+/** What complete_invoice_rows returns (migration 20260906135730). */
+interface CompleteRowsOutcome {
+  ok: boolean
+  code?: string
+  wrote?: boolean
+  rows?: number
+  header_updated?: boolean
+}
+
 export async function completeMigratedInvoiceLines(
   options: CompleteInvoiceLinesOptions,
 ): Promise<CompleteInvoiceLinesResult> {
@@ -357,19 +363,15 @@ export async function completeMigratedInvoiceLines(
       const vatAmount = mapped.invoice.vat_amount as number
       header = {
         subtotal,
-        subtotalSek: toRowSek(subtotal, row),
-        vatAmount,
-        vatAmountSek: toRowSek(vatAmount, row),
-        vatRate: mapped.invoice.vat_rate as number | null,
-        vatTreatment: mapped.invoice.vat_treatment as string,
+        subtotal_sek: toRowSek(subtotal, row),
+        vat_amount: vatAmount,
+        vat_amount_sek: toRowSek(vatAmount, row),
+        vat_rate: mapped.invoice.vat_rate as number | null,
+        vat_treatment: mapped.invoice.vat_treatment as string,
       }
     }
 
-    planned.push({
-      row,
-      items: mapped.items.map((item) => ({ ...item, invoice_id: row.id })),
-      header,
-    })
+    planned.push({ row, items: mapped.items, header })
   }
 
   if (dryRun) {
@@ -379,44 +381,36 @@ export async function completeMigratedInvoiceLines(
     return result
   }
 
-  for (const batch of chunk(planned, WRITE_CHUNK_SIZE)) {
-    // A concurrent run (the wizard and the cron, or two crons overlapping)
-    // may have filled some of these since the candidates were loaded. Rows
-    // are appended, never replaced, so a second write would double them.
-    const filled = await alreadyFilled(supabase, batch.map((p) => p.row.id))
-    const todo = batch.filter((p) => !filled.has(p.row.id))
-
-    const written = await insertRows(supabase, todo)
-    for (const plan of todo) {
-      if (!written.has(plan.row.id)) {
-        result.failed++
-        continue
-      }
-      result.completed++
-      if (!plan.header) continue
-      // Written as a literal so the schema guard checks these columns.
-      const { error } = await supabase
-        .from('invoices')
-        .update({
-          subtotal: plan.header.subtotal,
-          subtotal_sek: plan.header.subtotalSek,
-          vat_amount: plan.header.vatAmount,
-          vat_amount_sek: plan.header.vatAmountSek,
-          vat_rate: plan.header.vatRate,
-          vat_treatment: plan.header.vatTreatment,
-        })
-        .eq('id', plan.row.id)
-        .eq('company_id', companyId)
-      if (error) {
-        // The rows landed; only the header split is still the old shape. The
-        // next run will not revisit this invoice (it now has rows), so say so.
-        log.error('header VAT update failed after the rows were written', {
-          companyId, invoiceId: plan.row.id, reason: error.message,
-        })
-        continue
-      }
-      result.headersUpdated++
+  for (const plan of planned) {
+    // One call per invoice, so a bad row set rejects its own invoice and not
+    // the hundred beside it. A concurrent run (the wizard and the cron, or
+    // two crons overlapping) that filled this invoice since the candidates
+    // were loaded leaves this call with wrote = false: rows are appended,
+    // never replaced, and the RPC's invoice lock is what keeps a second
+    // writer from doubling them.
+    const { data, error } = await supabase.rpc('complete_invoice_rows', {
+      p_company_id: companyId,
+      p_invoice_id: plan.row.id,
+      p_rows: plan.items,
+      p_header: plan.header,
+    })
+    const outcome = (data ?? null) as CompleteRowsOutcome | null
+    if (error || !outcome?.ok) {
+      result.failed++
+      log.error('complete_invoice_rows failed', {
+        companyId, invoiceId: plan.row.id, invoiceNumber: plan.row.invoice_number,
+        reason: error?.message ?? outcome?.code ?? 'empty RPC response',
+      })
+      continue
     }
+    if (!outcome.wrote) {
+      log.info('invoice gained rows since the candidates were loaded; left as is', {
+        companyId, invoiceId: plan.row.id,
+      })
+      continue
+    }
+    result.completed++
+    if (outcome.header_updated) result.headersUpdated++
   }
 
   result.remaining = candidates.length - result.completed
@@ -435,33 +429,3 @@ export async function completeMigratedInvoiceLines(
   })
   return result
 }
-
-/**
- * Insert every plan's rows: one statement for the batch, and on failure one
- * statement per invoice so a single bad row rejects its own invoice, not the
- * hundred beside it. An invoice's rows never split across statements: it
- * either has all of them or none.
- */
-async function insertRows(supabase: SupabaseClient, plans: PlannedWrite[]): Promise<Set<string>> {
-  const written = new Set<string>()
-  if (plans.length === 0) return written
-
-  const bulk = await supabase.from('invoice_items').insert(plans.flatMap((p) => p.items))
-  if (!bulk.error) {
-    for (const plan of plans) written.add(plan.row.id)
-    return written
-  }
-
-  log.warn('bulk invoice_items insert failed; retrying per invoice', { reason: bulk.error.message })
-  for (const plan of plans) {
-    const { error } = await supabase.from('invoice_items').insert(plan.items)
-    if (error) {
-      log.error('invoice_items insert failed', {
-        invoiceId: plan.row.id, invoiceNumber: plan.row.invoice_number, reason: error.message,
-      })
-      continue
-    }
-    written.add(plan.row.id)
-  }
-  return written
-}
diff --git a/extensions/general/arcim-migration/lib/migration-orchestrator.ts b/extensions/general/arcim-migration/lib/migration-orchestrator.ts
index 3bab877a..8d383ee7 100644
--- a/extensions/general/arcim-migration/lib/migration-orchestrator.ts
+++ b/extensions/general/arcim-migration/lib/migration-orchestrator.ts
@@ -86,6 +86,8 @@ export interface MigrationOptions {
  * PostgREST's practical size limit while minimising round-trips.
  */
 const INSERT_CHUNK_SIZE = 500
+/** Sales-invoice row writes in flight at once (one complete_invoice_rows call per invoice). */
+const ITEM_RPC_CONCURRENCY = 8
 const ENRICHMENT_CONCURRENCY = 10
 
 function emitProgress(options: MigrationOptions, progress: MigrationProgress) {
@@ -736,13 +738,13 @@ export async function executeMigration(options: MigrationOptions): Promise<Migra
             errorSample ??= outcome.firstError
           }
 
-          const allItems: Record<string, unknown>[] = []
+          const rowsByInvoice: { invoiceId: string; rows: Record<string, unknown>[] }[] = []
           for (let i = 0; i < mappedBatch.length; i++) {
             const insertedRow = outcome.returned[i]
             if (!insertedRow) continue
             const invoiceId = insertedRow.id
-            for (const item of mappedBatch[i].items) {
-              allItems.push({ ...item, invoice_id: invoiceId })
+            if (mappedBatch[i].items.length > 0) {
+              rowsByInvoice.push({ invoiceId: String(invoiceId), rows: mappedBatch[i].items })
             }
             registrationLinkInputs.push({
               invoiceId: String(invoiceId),
@@ -772,13 +774,27 @@ export async function executeMigration(options: MigrationOptions): Promise<Migra
             imported++
           }
 
-          if (allItems.length > 0) {
-            for (const itemBatch of chunk(allItems, INSERT_CHUNK_SIZE)) {
-              const { error: itemErr } = await supabase.from('invoice_items').insert(itemBatch)
-              if (itemErr) {
-                console.error(`[migration] Sales invoice items insert failed (${itemBatch.length}):`, itemErr.message)
+          // One complete_invoice_rows call per invoice (migration
+          // 20260906135730): the write path the row-completion pass uses, so
+          // an invoice's rows are written at most once whichever writer gets
+          // there first, and a bad row set rejects its own invoice rather than
+          // the whole chunk. Small concurrent groups keep the round trips off
+          // the wizard's clock.
+          for (const group of chunk(rowsByInvoice, ITEM_RPC_CONCURRENCY)) {
+            await Promise.all(group.map(async ({ invoiceId, rows }) => {
+              const { data, error: itemErr } = await supabase.rpc('complete_invoice_rows', {
+                p_company_id: companyId,
+                p_invoice_id: invoiceId,
+                p_rows: rows,
+              })
+              const rpcOutcome = (data ?? null) as { ok?: boolean; code?: string } | null
+              if (itemErr || !rpcOutcome?.ok) {
+                console.error(
+                  `[migration] Sales invoice items insert failed for ${invoiceId}:`,
+                  itemErr?.message ?? rpcOutcome?.code ?? 'empty RPC response',
+                )
               }
-            }
+            }))
           }
         }
 
diff --git a/supabase/migrations/20260906135730_complete_invoice_rows_rpc.sql b/supabase/migrations/20260906135730_complete_invoice_rows_rpc.sql
new file mode 100644
index 00000000..f412d956
--- /dev/null
+++ b/supabase/migrations/20260906135730_complete_invoice_rows_rpc.sql
@@ -0,0 +1,190 @@
+-- complete_invoice_rows: write a migrated invoice's rows at most once.
+--
+-- Two writers put invoice_items under MIGRATED sales invoices: the migration
+-- wizard (rows inserted milliseconds after the invoice header) and the hourly
+-- row-completion pass (#2291, extensions/general/arcim-migration/lib/
+-- complete-invoice-lines.ts) that fetches the rows the wizard's hydration
+-- budget did not reach. Both were check-then-insert across two statements
+-- with nothing serializing them per invoice: invoice_items carries only a
+-- non-unique index on invoice_id, so two writers landing on the same invoice
+-- at once both succeeded and doubled its rows (#2313, recorded as an accepted
+-- residual on #2291). The pass also wrote the header VAT split in a third
+-- statement, so "rows landed, header did not" was a reachable state that the
+-- next run could not revisit (the invoice now had rows).
+--
+-- This RPC is the one write path for both. It locks the invoice row
+-- (FOR UPDATE, scoped to the company), inserts the rows only when the invoice
+-- still has none, and applies the optional header split in the same
+-- transaction. A concurrent caller for the same invoice queues on the lock
+-- and, once the first commits, reads the rows and returns wrote = false. The
+-- invariant is "an invoice's rows are written at most once by the completion
+-- writers"; it needs no unique index and therefore no clean-up of the legacy
+-- rows that carry duplicate sort_order values within one invoice.
+--
+-- Column set: exactly what mapSalesInvoiceLine emits. An unknown key is
+-- refused (UNKNOWN_COLUMN) rather than dropped, so a mapper that starts
+-- emitting a column this function does not carry fails loudly instead of
+-- silently losing it. The facts a sales row must state (description,
+-- line_total, vat_rate, vat_amount; ML 17 kap 24 §) are required
+-- (MISSING_REQUIRED) rather than defaulted: the table's DEFAULT 25 on
+-- vat_rate would put a fabricated 25 % on a row whose source said nothing,
+-- and both writers always send all four, so a missing one is a bug to
+-- surface, not a gap to fill. Only sort_order, quantity, unit and line_type
+-- take their table defaults; none of them states a tax fact. The rate is
+-- not restricted to the Swedish set: 0 (omvänd skattskyldighet, export) and
+-- foreign rates (OSS, unionsordningen) are legitimate on a migrated row.
+--
+-- Actor resolution mirrors the sibling definer RPCs: service_role callers
+-- (the cron on createServiceClientNoCookies, auth.uid() NULL) are trusted,
+-- the same trust a direct service-role insert already carries; every other
+-- caller is pinned to auth.uid() and must hold a write role (owner, admin or
+-- member) in p_company_id, the same gate as invoice_items_insert. A caller
+-- with no JWT at all is refused.
+--
+-- pg-test: tests/pg/complete-invoice-rows-rpc.pg.test.ts
+
+CREATE OR REPLACE FUNCTION public.complete_invoice_rows(
+  p_company_id uuid,
+  p_invoice_id uuid,
+  p_rows jsonb,
+  p_header jsonb DEFAULT NULL
+)
+RETURNS jsonb
+LANGUAGE plpgsql
+SECURITY DEFINER
+SET search_path TO 'public'
+AS $$
+DECLARE
+  v_caller uuid;
+  v_locked uuid;
+  v_bad_key text;
+  v_missing text;
+  v_inserted integer := 0;
+  v_header_updated boolean := false;
+BEGIN
+  IF COALESCE(auth.role(), '') <> 'service_role' THEN
+    v_caller := auth.uid();
+    IF v_caller IS NULL THEN
+      RETURN jsonb_build_object('ok', false, 'code', 'FORBIDDEN');
+    END IF;
+    -- SECURITY DEFINER bypasses RLS, so the write-role gate is explicit.
+    IF NOT EXISTS (
+      SELECT 1 FROM public.company_members cm
+      WHERE cm.company_id = p_company_id
+        AND cm.user_id = v_caller
+        AND cm.role IN ('owner', 'admin', 'member')
+    ) THEN
+      RETURN jsonb_build_object('ok', false, 'code', 'FORBIDDEN');
+    END IF;
+  END IF;
+
+  IF p_rows IS NULL OR jsonb_typeof(p_rows) <> 'array' OR jsonb_array_length(p_rows) = 0 THEN
+    RETURN jsonb_build_object('ok', false, 'code', 'NO_ROWS');
+  END IF;
+  IF EXISTS (SELECT 1 FROM jsonb_array_elements(p_rows) AS e WHERE jsonb_typeof(e) <> 'object') THEN
+    RETURN jsonb_build_object('ok', false, 'code', 'INVALID_ROWS');
+  END IF;
+
+  SELECT k INTO v_bad_key
+  FROM jsonb_array_elements(p_rows) AS e, jsonb_object_keys(e) AS k
+  WHERE k NOT IN (
+    'sort_order', 'description', 'quantity', 'unit', 'unit_price',
+    'line_total', 'vat_rate', 'vat_amount', 'line_type'
+  )
+  LIMIT 1;
+  IF v_bad_key IS NOT NULL THEN
+    RETURN jsonb_build_object('ok', false, 'code', 'UNKNOWN_COLUMN',
+      'details', jsonb_build_object('column', v_bad_key));
+  END IF;
+
+  -- Absent or JSON null: either would otherwise fall through to a default.
+  SELECT k INTO v_missing
+  FROM jsonb_array_elements(p_rows) AS e,
+       unnest(ARRAY['description', 'line_total', 'vat_rate', 'vat_amount']) AS k
+  WHERE NOT (e ? k) OR jsonb_typeof(e -> k) = 'null'
+  LIMIT 1;
+  IF v_missing IS NOT NULL THEN
+    RETURN jsonb_build_object('ok', false, 'code', 'MISSING_REQUIRED',
+      'details', jsonb_build_object('column', v_missing));
+  END IF;
+
+  IF p_header IS NOT NULL THEN
+    -- All six or nothing: a partial header would null the columns it omits.
+    IF jsonb_typeof(p_header) <> 'object' OR NOT (p_header ?& ARRAY[
+      'subtotal', 'subtotal_sek', 'vat_amount', 'vat_amount_sek', 'vat_rate', 'vat_treatment'
+    ]) THEN
+      RETURN jsonb_build_object('ok', false, 'code', 'INVALID_HEADER');
+    END IF;
+  END IF;
+
+  -- The per-invoice serialization point. A concurrent writer for the same
+  -- invoice waits here and sees the committed rows below.
+  SELECT i.id INTO v_locked
+  FROM public.invoices i
+  WHERE i.id = p_invoice_id
+    AND i.company_id = p_company_id
+  FOR UPDATE;
+  IF v_locked IS NULL THEN
+    RETURN jsonb_build_object('ok', false, 'code', 'INVOICE_NOT_FOUND');
+  END IF;
+
+  IF EXISTS (SELECT 1 FROM public.invoice_items ii WHERE ii.invoice_id = p_invoice_id) THEN
+    RETURN jsonb_build_object('ok', true, 'wrote', false, 'rows', 0, 'header_updated', false);
+  END IF;
+
+  INSERT INTO public.invoice_items
+    (invoice_id, sort_order, description, quantity, unit, unit_price,
+     line_total, vat_rate, vat_amount, line_type)
+  SELECT
+    p_invoice_id,
+    COALESCE(r.sort_order, 0),
+    r.description,
+    COALESCE(r.quantity, 1),
+    COALESCE(r.unit, 'st'),
+    COALESCE(r.unit_price, 0),
+    r.line_total,
+    r.vat_rate,
+    r.vat_amount,
+    COALESCE(r.line_type, 'product')
+  FROM jsonb_to_recordset(p_rows) AS r(
+    sort_order integer,
+    description text,
+    quantity numeric,
+    unit text,
+    unit_price numeric,
+    line_total numeric,
+    vat_rate numeric,
+    vat_amount numeric,
+    line_type text
+  );
+  GET DIAGNOSTICS v_inserted = ROW_COUNT;
+
+  IF p_header IS NOT NULL THEN
+    UPDATE public.invoices
+    SET subtotal = (p_header ->> 'subtotal')::numeric,
+        subtotal_sek = (p_header ->> 'subtotal_sek')::numeric,
+        vat_amount = (p_header ->> 'vat_amount')::numeric,
+        vat_amount_sek = (p_header ->> 'vat_amount_sek')::numeric,
+        vat_rate = (p_header ->> 'vat_rate')::numeric,
+        vat_treatment = p_header ->> 'vat_treatment'
+    WHERE id = p_invoice_id
+      AND company_id = p_company_id;
+    v_header_updated := true;
+  END IF;
+
+  RETURN jsonb_build_object(
+    'ok', true,
+    'wrote', true,
+    'rows', v_inserted,
+    'header_updated', v_header_updated
+  );
+END;
+$$;
+
+REVOKE ALL ON FUNCTION public.complete_invoice_rows(uuid, uuid, jsonb, jsonb) FROM PUBLIC, anon;
+GRANT EXECUTE ON FUNCTION public.complete_invoice_rows(uuid, uuid, jsonb, jsonb) TO authenticated, service_role;
+
+COMMENT ON FUNCTION public.complete_invoice_rows(uuid, uuid, jsonb, jsonb) IS
+  'Writes a migrated invoice''s rows (and optionally its header VAT split) at most once: locks the invoice, inserts only when it has no rows, returns wrote = false otherwise. The one write path for the migration wizard and the row-completion pass.';
+
+NOTIFY pgrst, 'reload schema';
diff --git a/tests/pg/complete-invoice-rows-rpc.pg.test.ts b/tests/pg/complete-invoice-rows-rpc.pg.test.ts
new file mode 100644
index 00000000..520de3c6
--- /dev/null
+++ b/tests/pg/complete-invoice-rows-rpc.pg.test.ts
@@ -0,0 +1,362 @@
+import { randomUUID } from 'node:crypto'
+import { describe, it, expect } from 'vitest'
+import type { PoolClient } from 'pg'
+import { getPool, getClient, runAsServiceRole, withUserContext } from './setup'
+import { seedCompany, insertAuthUser, insertCompanyMember } from './fixtures'
+
+// pg-real coverage for 20260906135730_complete_invoice_rows_rpc:
+// complete_invoice_rows writes a migrated invoice's rows (and its optional
+// header VAT split) at most once. It locks the invoice, inserts only when the
+// invoice still has no rows, applies the header in the same transaction,
+// refuses non-members and foreign invoices, and serializes concurrent
+// writers so the second one finds the first one's rows (#2313).
+
+type RpcResult = {
+  ok: boolean
+  code?: string
+  wrote?: boolean
+  rows?: number
+  header_updated?: boolean
+  details?: Record<string, unknown>
+}
+
+const SIGNATURE = 'public.complete_invoice_rows(uuid,uuid,jsonb,jsonb)'
+
+/** Two rows the way mapSalesInvoiceLine emits them: 1 000 kr net, 25 %. */
+const ROWS = [
+  { sort_order: 1, description: 'Konsulttid', quantity: 8, unit: 'h', unit_price: 100, line_total: 800, vat_rate: 25, vat_amount: 200, line_type: 'product' },
+  { sort_order: 2, description: 'Resa', quantity: 1, unit: 'st', unit_price: 200, line_total: 200, vat_rate: 25, vat_amount: 50, line_type: 'product' },
+]
+
+/** The header split the detail form established for ROWS. */
+const HEADER = {
+  subtotal: 1000,
+  subtotal_sek: 1000,
+  vat_amount: 250,
+  vat_amount_sek: 250,
+  vat_rate: 25,
+  vat_treatment: 'standard_25',
+}
+
+/**
+ * A migrated invoice the way the pre-#1745 import left it: total right, 25 %
+ * label beside 0 kr VAT and subtotal = total, and no rows.
+ */
+async function insertInvoice(companyId: string, userId: string): Promise<string> {
+  const customerId = randomUUID()
+  await getPool().query(
+    `INSERT INTO public.customers (id, user_id, company_id, name, customer_type)
+     VALUES ($1, $2, $3, 'Kund AB', 'swedish_business')`,
+    [customerId, userId, companyId],
+  )
+  const id = randomUUID()
+  await getPool().query(
+    `INSERT INTO public.invoices
+       (id, user_id, company_id, customer_id, invoice_number, document_type,
+        invoice_date, due_date, currency, subtotal, subtotal_sek, vat_amount, vat_amount_sek,
+        total, total_sek, vat_treatment, vat_rate, status)
+     VALUES ($1, $2, $3, $4, $5, 'invoice',
+             '2026-03-14', '2026-04-13', 'SEK', 1250, 1250, 0, 0,
+             1250, 1250, 'standard_25', 25, 'sent')`,
+    [id, userId, companyId, customerId, `1001-${id.slice(0, 8)}`],
+  )
+  return id
+}
+
+async function callRpc(
+  client: PoolClient,
+  companyId: string,
+  invoiceId: string,
+  rows: unknown,
+  header: unknown = null,
+): Promise<RpcResult> {
+  const { rows: out } = await client.query<{ r: RpcResult }>(
+    `SELECT public.complete_invoice_rows($1, $2, $3::jsonb, $4::jsonb) AS r`,
+    [companyId, invoiceId, JSON.stringify(rows), header === null ? null : JSON.stringify(header)],
+  )
+  return out[0].r
+}
+
+async function beginAsUser(client: PoolClient, userId: string): Promise<void> {
+  await client.query('BEGIN')
+  await client.query(`SELECT set_config('request.jwt.claims', $1, true)`, [
+    JSON.stringify({ sub: userId, role: 'authenticated' }),
+  ])
+  await client.query(`SELECT set_config('request.jwt.claim.sub', $1, true)`, [userId])
+  await client.query('SET LOCAL ROLE authenticated')
+}
+
+/** Like withUserContext but COMMITs, so a later session can observe the result. */
+async function asUser<T>(userId: string, fn: (client: PoolClient) => Promise<T>): Promise<T> {
+  const client = await getClient()
+  try {
+    await beginAsUser(client, userId)
+    const result = await fn(client)
+    await client.query('COMMIT')
+    return result
+  } catch (err) {
+    await client.query('ROLLBACK').catch(() => {})
+    throw err
+  } finally {
+    client.release()
+  }
+}
+
+async function storedRows(invoiceId: string) {
+  const { rows } = await getPool().query<{
+    description: string
+    sort_order: number
+    quantity: string
+    unit: string
+    line_total: string
+    vat_rate: string
+    line_type: string
+  }>(
+    `SELECT description, sort_order, quantity::text, unit, line_total::text, vat_rate::text, line_type
+     FROM public.invoice_items WHERE invoice_id = $1 ORDER BY sort_order, description`,
+    [invoiceId],
+  )
+  return rows
+}
+
+async function storedHeader(invoiceId: string) {
+  const { rows } = await getPool().query<{
+    subtotal: string
+    subtotal_sek: string
+    vat_amount: string
+    vat_amount_sek: string
+    vat_rate: string
+    vat_treatment: string
+    total: string
+  }>(
+    `SELECT subtotal::text, subtotal_sek::text, vat_amount::text, vat_amount_sek::text,
+            vat_rate::text, vat_treatment, total::text
+     FROM public.invoices WHERE id = $1`,
+    [invoiceId],
+  )
+  const h = rows[0]!
+  return {
+    subtotal: Number(h.subtotal),
+    subtotal_sek: Number(h.subtotal_sek),
+    vat_amount: Number(h.vat_amount),
+    vat_amount_sek: Number(h.vat_amount_sek),
+    vat_rate: Number(h.vat_rate),
+    vat_treatment: h.vat_treatment,
+    total: Number(h.total),
+  }
+}
+
+describe('complete_invoice_rows', () => {
+  it('writes the rows and the header split together, and only once', async () => {
+    const { userId, companyId } = await seedCompany()
+    const invoiceId = await insertInvoice(companyId, userId)
+
+    const first = await asUser(userId, (client) => callRpc(client, companyId, invoiceId, ROWS, HEADER))
+    expect(first).toEqual({ ok: true, wrote: true, rows: 2, header_updated: true })
+
+    const rows = await storedRows(invoiceId)
+    expect(rows.map((r) => r.description)).toEqual(['Konsulttid', 'Resa'])
+    expect(rows[0]).toMatchObject({ sort_order: 1, unit: 'h', line_type: 'product' })
+    expect(Number(rows[0].quantity)).toBe(8)
+    expect(Number(rows[0].line_total)).toBe(800)
+    expect(Number(rows[0].vat_rate)).toBe(25)
+
+    const header = await storedHeader(invoiceId)
+    expect(header).toEqual({ ...HEADER, total: 1250 })
+
+    // Same rows again, or different ones: nothing is appended and the header
+    // is not touched, whichever run gets there second.
+    const again = await asUser(userId, (client) => callRpc(client, companyId, invoiceId, ROWS, HEADER))
+    expect(again).toEqual({ ok: true, wrote: false, rows: 0, header_updated: false })
+    const other = await asUser(userId, (client) =>
+      callRpc(
+        client, companyId, invoiceId,
+        [{ ...ROWS[0], description: 'Annat', line_total: 5, vat_amount: 1.25 }],
+        { ...HEADER, subtotal: 5 },
+      ),
+    )
+    expect(other).toEqual({ ok: true, wrote: false, rows: 0, header_updated: false })
+
+    expect((await storedRows(invoiceId)).map((r) => r.description)).toEqual(['Konsulttid', 'Resa'])
+    expect(await storedHeader(invoiceId)).toEqual({ ...HEADER, total: 1250 })
+  })
+
+  it('writes the rows without a header when none is given (the wizard path); only the non-tax columns take table defaults', async () => {
+    const { userId, companyId } = await seedCompany()
+    const invoiceId = await insertInvoice(companyId, userId)
+    const before = await storedHeader(invoiceId)
+
+    const r = await asUser(userId, (client) =>
+      callRpc(client, companyId, invoiceId, [{ description: 'Bara text', line_total: 0, vat_rate: 0, vat_amount: 0 }]),
+    )
+    expect(r).toEqual({ ok: true, wrote: true, rows: 1, header_updated: false })
+
+    const rows = await storedRows(invoiceId)
+    expect(rows).toHaveLength(1)
+    expect(rows[0]).toMatchObject({ description: 'Bara text', sort_order: 0, unit: 'st', line_type: 'product' })
+    expect(Number(rows[0].quantity)).toBe(1)
+    expect(Number(rows[0].line_total)).toBe(0)
+    expect(Number(rows[0].vat_rate)).toBe(0)
+    expect(await storedHeader(invoiceId)).toEqual(before)
+  })
+
+  it('refuses a row that omits a tax fact instead of defaulting it (no fabricated 25 %)', async () => {
+    const { userId, companyId } = await seedCompany()
+    const invoiceId = await insertInvoice(companyId, userId)
+
+    await withUserContext(userId, async (client) => {
+      for (const key of ['vat_rate', 'line_total', 'vat_amount', 'description'] as const) {
+        const { [key]: _omitted, ...without } = ROWS[0]
+        void _omitted
+        expect(await callRpc(client, companyId, invoiceId, [without])).toEqual({
+          ok: false, code: 'MISSING_REQUIRED', details: { column: key },
+        })
+        // JSON null is "absent" too: it must not fall through to a default.
+        expect(await callRpc(client, companyId, invoiceId, [{ ...ROWS[0], [key]: null }])).toEqual({
+          ok: false, code: 'MISSING_REQUIRED', details: { column: key },
+        })
+      }
+      // One bad row refuses the whole set: an invoice's rows land together.
+      expect(await callRpc(client, companyId, invoiceId, [ROWS[0], { ...ROWS[1], vat_rate: undefined }])).toEqual({
+        ok: false, code: 'MISSING_REQUIRED', details: { column: 'vat_rate' },
+      })
+    })
+
+    expect(await storedRows(invoiceId)).toHaveLength(0)
+  })
+
+  it('accepts a stated 0 % (omvänd skattskyldighet, export) and a foreign rate (OSS): the value is not restricted', async () => {
+    const { userId, companyId } = await seedCompany()
+    const invoiceId = await insertInvoice(companyId, userId)
+
+    const r = await asUser(userId, (client) =>
+      callRpc(client, companyId, invoiceId, [
+        { ...ROWS[0], description: 'Konsulttid DE (reverse charge)', vat_rate: 0, vat_amount: 0 },
+        { ...ROWS[1], description: 'Vara DE (OSS)', vat_rate: 19, vat_amount: 38 },
+      ]),
+    )
+    expect(r).toEqual({ ok: true, wrote: true, rows: 2, header_updated: false })
+
+    const rows = await storedRows(invoiceId)
+    expect(rows.map((row) => Number(row.vat_rate))).toEqual([0, 19])
+  })
+
+  it('rolls the rows back when the header update fails: header and rows land together or not at all', async () => {
+    const { userId, companyId } = await seedCompany()
+    const invoiceId = await insertInvoice(companyId, userId)
+    const before = await storedHeader(invoiceId)
+
+    await expect(
+      asUser(userId, (client) => callRpc(client, companyId, invoiceId, ROWS, { ...HEADER, vat_rate: 'tjugofem' })),
+    ).rejects.toThrow(/numeric/)
+
+    expect(await storedRows(invoiceId)).toHaveLength(0)
+    expect(await storedHeader(invoiceId)).toEqual(before)
+  })
+
+  it('refuses a payload it cannot store honestly, before locking anything', async () => {
+    const { userId, companyId } = await seedCompany()
+    const invoiceId = await insertInvoice(companyId, userId)
+
+    await withUserContext(userId, async (client) => {
+      expect(await callRpc(client, companyId, invoiceId, [])).toEqual({ ok: false, code: 'NO_ROWS' })
+      expect(await callRpc(client, companyId, invoiceId, { description: 'x' })).toEqual({ ok: false, code: 'NO_ROWS' })
+      expect(await callRpc(client, companyId, invoiceId, ['x'])).toEqual({ ok: false, code: 'INVALID_ROWS' })
+      // A column the mapper started emitting must fail loudly, not be dropped.
+      expect(await callRpc(client, companyId, invoiceId, [{ ...ROWS[0], invoice_id: invoiceId }])).toEqual({
+        ok: false, code: 'UNKNOWN_COLUMN', details: { column: 'invoice_id' },
+      })
+      // A partial header would null the columns it omits.
+      expect(await callRpc(client, companyId, invoiceId, ROWS, { subtotal: 1000 })).toEqual({
+        ok: false, code: 'INVALID_HEADER',
+      })
+    })
+
+    expect(await storedRows(invoiceId)).toHaveLength(0)
+  })
+
+  it('refuses an invoice outside the company and callers without a write role', async () => {
+    const a = await seedCompany()
+    const b = await seedCompany()
+    const invoiceA = await insertInvoice(a.companyId, a.userId)
+
+    // Owner of B naming their own company with A's invoice, and owner of A
+    // naming company B: both are "not found", never a write.
+    await withUserContext(b.userId, async (client) => {
+      expect(await callRpc(client, b.companyId, invoiceA, ROWS)).toEqual({ ok: false, code: 'INVOICE_NOT_FOUND' })
+      expect(await callRpc(client, a.companyId, invoiceA, ROWS)).toEqual({ ok: false, code: 'FORBIDDEN' })
+    })
+
+    const viewer = await insertAuthUser()
+    await insertCompanyMember({ companyId: a.companyId, userId: viewer, role: 'viewer' })
+    await withUserContext(viewer, async (client) => {
+      expect(await callRpc(client, a.companyId, invoiceA, ROWS)).toEqual({ ok: false, code: 'FORBIDDEN' })
+    })
+
+    // No JWT at all (a plain connection) is refused, not trusted.
+    const client = await getClient()
+    try {
+      expect(await callRpc(client, a.companyId, invoiceA, ROWS)).toEqual({ ok: false, code: 'FORBIDDEN' })
+    } finally {
+      client.release()
+    }
+
+    expect(await storedRows(invoiceA)).toHaveLength(0)
+  })
+
+  it('lets the cron write on the service client, with no session user', async () => {
+    const { userId, companyId } = await seedCompany()
+    const invoiceId = await insertInvoice(companyId, userId)
+
+    const r = await runAsServiceRole((client) => callRpc(client, companyId, invoiceId, ROWS, HEADER))
+    expect(r).toEqual({ ok: true, wrote: true, rows: 2, header_updated: true })
+    expect(await storedRows(invoiceId)).toHaveLength(2)
+    expect(await storedHeader(invoiceId)).toEqual({ ...HEADER, total: 1250 })
+  })
+
+  it('serializes two writers on one invoice: the second waits on the lock and then finds the rows', async () => {
+    const { userId, companyId } = await seedCompany()
+    const invoiceId = await insertInvoice(companyId, userId)
+    const first = await getClient()
+    const second = await getClient()
+    try {
+      await beginAsUser(first, userId)
+      await beginAsUser(second, userId)
+
+      expect(await callRpc(first, companyId, invoiceId, ROWS, HEADER)).toMatchObject({ ok: true, wrote: true, rows: 2 })
+
+      let settled = false
+      const pending = callRpc(second, companyId, invoiceId, [{ ...ROWS[0], description: 'Dubblett' }]).then((r) => {
+        settled = true
+        return r
+      })
+      await new Promise((resolve) => setTimeout(resolve, 300))
+      // Blocked on the first writer's row lock, not running ahead of it.
+      expect(settled).toBe(false)
+
+      await first.query('COMMIT')
+      expect(await pending).toEqual({ ok: true, wrote: false, rows: 0, header_updated: false })
+      await second.query('COMMIT')
+    } catch (err) {
+      await first.query('ROLLBACK').catch(() => {})
+      await second.query('ROLLBACK').catch(() => {})
+      throw err
+    } finally {
+      first.release()
+      second.release()
+    }
+
+    expect((await storedRows(invoiceId)).map((r) => r.description)).toEqual(['Konsulttid', 'Resa'])
+  })
+
+  it('grants: anon and PUBLIC have no EXECUTE; authenticated and service_role do', async () => {
+    const { rows } = await getPool().query<{ anon_can: boolean; public_can: boolean; auth_can: boolean; service_can: boolean }>(
+      `SELECT has_function_privilege('anon', $1, 'EXECUTE') AS anon_can,
+              has_function_privilege('public', $1, 'EXECUTE') AS public_can,
+              has_function_privilege('authenticated', $1, 'EXECUTE') AS auth_can,
+              has_function_privilege('service_role', $1, 'EXECUTE') AS service_can`,
+      [SIGNATURE],
+    )
+    expect(rows[0]).toEqual({ anon_can: false, public_can: false, auth_can: true, service_can: true })
+  })
+})