diff --git a/supabase/drift-manifest.json b/supabase/drift-manifest.json
index 30fc86da3..8c975b2bd 100644
--- a/supabase/drift-manifest.json
+++ b/supabase/drift-manifest.json
@@ -1,9 +1,9 @@
 {
-  "generated_at": "2026-07-06T18:30:39.629Z",
+  "generated_at": "2026-07-08T06:45:01.039Z",
   "generator": "scripts/generate-drift-manifest.ts",
   "postgres_image": "supabase/postgres:17.6.1.127",
-  "schema_sha256": "8a88db3fcd974fae74807a91da118e8b0f9f4bcbf02df65b8b5670b22c32e24c",
-  "replay_seconds": 15,
+  "schema_sha256": "6174f3657e00b3a9ec11258cc599b956b6bdad805d7e37472710d0d34f4f1591",
+  "replay_seconds": 13,
   "snapshot": {
     "views": [
       {
@@ -5658,7 +5658,7 @@
           "postgres=X/postgres",
           "service_role=X/postgres"
         ],
-        "def_hash": "7ae76db3475c957f6af5199d59fa05ff",
+        "def_hash": "4ee148dcab28d6bbe10cfa542213f0f6",
         "signature": "public.cleanup_abandoned_document_index_generations(uuid,integer,boolean)"
       },
       {
@@ -5674,8 +5674,8 @@
           "postgres=X/postgres",
           "service_role=X/postgres"
         ],
-        "def_hash": "de9b9a4cc8ee9398b23488ca5a4749b9",
-        "signature": "public.complete_ingestion_job(uuid,uuid,uuid,text)"
+        "def_hash": "6406ade971963fa9b6d952765a697bd9",
+        "signature": "public.complete_ingestion_job(uuid,uuid,uuid,text,text)"
       },
       {
         "acl": [
@@ -5701,6 +5701,14 @@
         "def_hash": "e263f3819b05177abcf2eedd3f468991",
         "signature": "public.consume_api_subject_rate_limit(text,text,integer,integer)"
       },
+      {
+        "acl": [
+          "postgres=X/postgres",
+          "service_role=X/postgres"
+        ],
+        "def_hash": "c270be1b09209a63fd55f924748f40a2",
+        "signature": "public.corpus_topic_term_stats(text[],uuid)"
+      },
       {
         "acl": [
           "postgres=X/postgres",
@@ -5746,8 +5754,8 @@
           "postgres=X/postgres",
           "service_role=X/postgres"
         ],
-        "def_hash": "bdc7c13da9993cd550a10202b2c725e5",
-        "signature": "public.fail_or_retry_ingestion_job(uuid,uuid,uuid,boolean,text,text,text,timestamp with time zone)"
+        "def_hash": "0bbcb829db375283a92627aba610e6df",
+        "signature": "public.fail_or_retry_ingestion_job(uuid,uuid,uuid,boolean,text,text,text,timestamp with time zone,text)"
       },
       {
         "acl": [
@@ -5898,7 +5906,7 @@
           "postgres=X/postgres",
           "service_role=X/postgres"
         ],
-        "def_hash": "ded3c523452e1483097d15a7ac8a4639",
+        "def_hash": "843e3e00240d7fcbf6465bdfd3801525",
         "signature": "public.refresh_import_batch_status(uuid)"
       },
       {
diff --git a/supabase/migrations/20260708130000_ingestion_concurrency_rpc_hardening.sql b/supabase/migrations/20260708130000_ingestion_concurrency_rpc_hardening.sql
new file mode 100644
index 000000000..1664e326c
--- /dev/null
+++ b/supabase/migrations/20260708130000_ingestion_concurrency_rpc_hardening.sql
@@ -0,0 +1,431 @@
+-- Ingestion concurrency RPC hardening (audit R1/R2, R7, R9, R23).
+--
+-- NOT YET APPLIED TO LIVE. Apply through the linked migration workflow with
+-- operator approval, then run npm run check:drift + check:indexing (per
+-- docs/supabase-migration-reconciliation.md). schema.sql is reconciled to match
+-- these definitions and supabase/drift-manifest.json is regenerated in the same
+-- change. Backward compatible: p_worker_id defaults null, so the 4-/8-arg calls
+-- the current worker makes keep resolving to these functions until the worker is
+-- redeployed to pass its worker id.
+--
+-- Companion state model + verified violations: docs/ingestion-state-machine.md.
+-- Related shipped app-code fixes (PR #369): R11 janitor guard, R1 lease
+-- heartbeat (worker refreshes locked_at). These RPC fences are the residual
+-- defense-in-depth once a worker is genuinely reclaimed.
+
+set search_path = public, extensions, pg_temp;
+
+-- R9: serialize concurrent batch-status refreshes by locking the batch row
+-- before counting, so the last two jobs of a batch cannot both compute
+-- 'processing' from pre-commit snapshots and pin it forever.
+create or replace function public.refresh_import_batch_status(p_batch_id uuid)
+returns jsonb
+language plpgsql
+set search_path = public, extensions, pg_temp
+as $$
+declare
+  queued_count integer := 0;
+  processing_count integer := 0;
+  failed_count integer := 0;
+  next_status text;
+begin
+  if p_batch_id is null then
+    return jsonb_build_object('ok', false, 'reason', 'missing_batch_id');
+  end if;
+
+  perform 1 from public.import_batches where id = p_batch_id for update;
+
+  select
+    count(*) filter (where status = 'pending'),
+    count(*) filter (where status = 'processing'),
+    count(*) filter (where status = 'failed')
+  into queued_count, processing_count, failed_count
+  from public.ingestion_jobs
+  where batch_id = p_batch_id;
+
+  next_status := case
+    when queued_count > 0 or processing_count > 0 then 'processing'
+    when failed_count > 0 then 'completed_with_errors'
+    else 'completed'
+  end;
+
+  update public.import_batches
+  set
+    status = next_status,
+    failed_files = failed_count,
+    completed_at = case when next_status = 'processing' then null else now() end
+  where id = p_batch_id;
+
+  return jsonb_build_object(
+    'ok', true,
+    'status', next_status,
+    'queued', queued_count,
+    'processing', processing_count,
+    'failed', failed_count
+  );
+end;
+$$;
+
+-- R1/R2: complete_ingestion_job gains an optional p_worker_id lease fence.
+-- Signature change (added trailing param) → drop the old signature first so
+-- create does not leave a stale overload.
+drop function if exists public.complete_ingestion_job(uuid, uuid, uuid, text);
+
+create or replace function public.complete_ingestion_job(
+  p_job_id uuid,
+  p_document_id uuid,
+  p_batch_id uuid default null,
+  p_stage text default 'indexed',
+  p_worker_id text default null
+)
+returns jsonb
+language plpgsql
+set search_path = public, extensions, pg_temp
+as $$
+declare
+  v_matched integer;
+begin
+  update public.ingestion_jobs
+  set
+    status = 'completed',
+    stage = p_stage,
+    progress = 100,
+    error_message = null,
+    locked_at = null,
+    locked_by = null,
+    completed_at = now()
+  where id = p_job_id
+    and document_id = p_document_id
+    and (p_worker_id is null or locked_by = p_worker_id);
+  get diagnostics v_matched = row_count;
+
+  if p_worker_id is not null and v_matched = 0 then
+    return jsonb_build_object('ok', false, 'reason', 'lease_lost', 'job_id', p_job_id, 'document_id', p_document_id);
+  end if;
+
+  update public.ingestion_jobs
+  set
+    status = 'completed',
+    stage = 'superseded by successful index',
+    progress = 100,
+    error_message = null,
+    locked_at = null,
+    locked_by = null,
+    completed_at = now()
+  where document_id = p_document_id
+    and id <> p_job_id
+    and status in ('pending', 'processing', 'failed');
+
+  if p_batch_id is not null then
+    perform public.refresh_import_batch_status(p_batch_id);
+  end if;
+
+  return jsonb_build_object('ok', true, 'job_id', p_job_id, 'document_id', p_document_id);
+end;
+$$;
+
+-- R1/R2 lease fence + R7 attempt-exhaustion strand guard.
+drop function if exists public.fail_or_retry_ingestion_job(uuid, uuid, uuid, boolean, text, text, text, timestamptz);
+
+create or replace function public.fail_or_retry_ingestion_job(
+  p_job_id uuid,
+  p_document_id uuid,
+  p_batch_id uuid default null,
+  p_retry boolean default false,
+  p_document_status text default 'failed',
+  p_stage text default 'failed',
+  p_error_message text default null,
+  p_next_run_at timestamptz default null,
+  p_worker_id text default null
+)
+returns jsonb
+language plpgsql
+set search_path = public, extensions, pg_temp
+as $$
+declare
+  v_job public.ingestion_jobs%rowtype;
+  v_retry boolean;
+begin
+  select * into v_job
+  from public.ingestion_jobs
+  where id = p_job_id and document_id = p_document_id
+  for update;
+
+  if p_worker_id is not null and (v_job.id is null or v_job.locked_by is distinct from p_worker_id) then
+    return jsonb_build_object('ok', false, 'reason', 'lease_lost', 'job_id', p_job_id, 'document_id', p_document_id, 'retry', false);
+  end if;
+
+  v_retry := p_retry and coalesce(v_job.attempt_count, 0) < coalesce(v_job.max_attempts, 0);
+
+  update public.documents
+  set
+    status = p_document_status,
+    error_message = p_error_message
+  where id = p_document_id;
+
+  update public.ingestion_jobs
+  set
+    status = case when v_retry then 'pending' else 'failed' end,
+    stage = p_stage,
+    progress = case when v_retry then 0 else 100 end,
+    error_message = p_error_message,
+    locked_at = null,
+    locked_by = null,
+    next_run_at = coalesce(p_next_run_at, next_run_at),
+    completed_at = case when v_retry then null else now() end
+  where id = p_job_id
+    and document_id = p_document_id;
+
+  if p_batch_id is not null then
+    perform public.refresh_import_batch_status(p_batch_id);
+  end if;
+
+  return jsonb_build_object('ok', true, 'job_id', p_job_id, 'document_id', p_document_id, 'retry', v_retry);
+end;
+$$;
+
+-- R23: re-assert the open-job guard immediately before the destructive deletes
+-- (it previously lived only in the candidate CTE, before the seven counts).
+create or replace function public.cleanup_abandoned_document_index_generations(
+  p_document_id uuid default null,
+  p_limit integer default 100,
+  p_dry_run boolean default true
+)
+returns jsonb
+language plpgsql
+set search_path = public, extensions, pg_temp
+as $$
+declare
+  target_document_ids uuid[] := '{}'::uuid[];
+  chunk_count integer := 0;
+  image_count integer := 0;
+  table_fact_count integer := 0;
+  embedding_field_count integer := 0;
+  index_unit_count integer := 0;
+  memory_card_count integer := 0;
+  section_count integer := 0;
+begin
+  perform set_config('statement_timeout', '180000', true);
+
+  with candidate_documents as (
+    select distinct document_id
+    from (
+      select c.document_id
+      from public.document_chunks c
+      join public.documents d on d.id = c.document_id
+      where (p_document_id is null or c.document_id = p_document_id)
+        and c.index_generation_id is not null
+        and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '')
+        and not exists (
+          select 1 from public.ingestion_jobs j
+          where j.document_id = c.document_id
+            and j.status in ('pending', 'processing')
+        )
+      union all
+      select a.document_id
+      from public.document_images a
+      join public.documents d on d.id = a.document_id
+      where (p_document_id is null or a.document_id = p_document_id)
+        and a.index_generation_id is not null
+        and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '')
+        and not exists (
+          select 1 from public.ingestion_jobs j
+          where j.document_id = a.document_id
+            and j.status in ('pending', 'processing')
+        )
+      union all
+      select a.document_id
+      from public.document_table_facts a
+      join public.documents d on d.id = a.document_id
+      where (p_document_id is null or a.document_id = p_document_id)
+        and a.index_generation_id is not null
+        and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '')
+        and not exists (
+          select 1 from public.ingestion_jobs j
+          where j.document_id = a.document_id
+            and j.status in ('pending', 'processing')
+        )
+      union all
+      select a.document_id
+      from public.document_embedding_fields a
+      join public.documents d on d.id = a.document_id
+      where (p_document_id is null or a.document_id = p_document_id)
+        and a.index_generation_id is not null
+        and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '')
+        and not exists (
+          select 1 from public.ingestion_jobs j
+          where j.document_id = a.document_id
+            and j.status in ('pending', 'processing')
+        )
+      union all
+      select a.document_id
+      from public.document_index_units a
+      join public.documents d on d.id = a.document_id
+      where (p_document_id is null or a.document_id = p_document_id)
+        and a.index_generation_id is not null
+        and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '')
+        and not exists (
+          select 1 from public.ingestion_jobs j
+          where j.document_id = a.document_id
+            and j.status in ('pending', 'processing')
+        )
+      union all
+      select a.document_id
+      from public.document_memory_cards a
+      join public.documents d on d.id = a.document_id
+      where (p_document_id is null or a.document_id = p_document_id)
+        and a.index_generation_id is not null
+        and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '')
+        and not exists (
+          select 1 from public.ingestion_jobs j
+          where j.document_id = a.document_id
+            and j.status in ('pending', 'processing')
+        )
+      union all
+      select a.document_id
+      from public.document_sections a
+      join public.documents d on d.id = a.document_id
+      where (p_document_id is null or a.document_id = p_document_id)
+        and a.index_generation_id is not null
+        and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '')
+        and not exists (
+          select 1 from public.ingestion_jobs j
+          where j.document_id = a.document_id
+            and j.status in ('pending', 'processing')
+        )
+    ) candidates
+    limit least(greatest(coalesce(p_limit, 100), 1), 1000)
+  )
+  select coalesce(array_agg(document_id), '{}'::uuid[])
+  into target_document_ids
+  from candidate_documents;
+
+  select count(*) into chunk_count
+  from public.document_chunks c
+  join public.documents d on d.id = c.document_id
+  where c.document_id = any(target_document_ids)
+    and c.index_generation_id is not null
+    and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+  select count(*) into image_count
+  from public.document_images a
+  join public.documents d on d.id = a.document_id
+  where a.document_id = any(target_document_ids)
+    and a.index_generation_id is not null
+    and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+  select count(*) into table_fact_count
+  from public.document_table_facts a
+  join public.documents d on d.id = a.document_id
+  where a.document_id = any(target_document_ids)
+    and a.index_generation_id is not null
+    and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+  select count(*) into embedding_field_count
+  from public.document_embedding_fields a
+  join public.documents d on d.id = a.document_id
+  where a.document_id = any(target_document_ids)
+    and a.index_generation_id is not null
+    and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+  select count(*) into index_unit_count
+  from public.document_index_units a
+  join public.documents d on d.id = a.document_id
+  where a.document_id = any(target_document_ids)
+    and a.index_generation_id is not null
+    and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+  select count(*) into memory_card_count
+  from public.document_memory_cards a
+  join public.documents d on d.id = a.document_id
+  where a.document_id = any(target_document_ids)
+    and a.index_generation_id is not null
+    and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+  select count(*) into section_count
+  from public.document_sections a
+  join public.documents d on d.id = a.document_id
+  where a.document_id = any(target_document_ids)
+    and a.index_generation_id is not null
+    and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+  if not coalesce(p_dry_run, true) then
+    select coalesce(array_agg(doc_id), '{}'::uuid[])
+    into target_document_ids
+    from unnest(target_document_ids) as doc_id
+    where not exists (
+      select 1 from public.ingestion_jobs j
+      where j.document_id = doc_id
+        and j.status in ('pending', 'processing')
+    );
+
+    delete from public.document_chunks c
+    using public.documents d
+    where d.id = c.document_id
+      and c.document_id = any(target_document_ids)
+      and c.index_generation_id is not null
+      and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+    delete from public.document_images a
+    using public.documents d
+    where d.id = a.document_id
+      and a.document_id = any(target_document_ids)
+      and a.index_generation_id is not null
+      and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+    delete from public.document_table_facts a
+    using public.documents d
+    where d.id = a.document_id
+      and a.document_id = any(target_document_ids)
+      and a.index_generation_id is not null
+      and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+    delete from public.document_embedding_fields a
+    using public.documents d
+    where d.id = a.document_id
+      and a.document_id = any(target_document_ids)
+      and a.index_generation_id is not null
+      and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+    delete from public.document_index_units a
+    using public.documents d
+    where d.id = a.document_id
+      and a.document_id = any(target_document_ids)
+      and a.index_generation_id is not null
+      and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+    delete from public.document_memory_cards a
+    using public.documents d
+    where d.id = a.document_id
+      and a.document_id = any(target_document_ids)
+      and a.index_generation_id is not null
+      and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+
+    delete from public.document_sections a
+    using public.documents d
+    where d.id = a.document_id
+      and a.document_id = any(target_document_ids)
+      and a.index_generation_id is not null
+      and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
+  end if;
+
+  return jsonb_build_object(
+    'ok', true,
+    'dry_run', coalesce(p_dry_run, true),
+    'document_count', coalesce(array_length(target_document_ids, 1), 0),
+    'document_ids', to_jsonb(target_document_ids),
+    'counts', jsonb_build_object(
+      'document_chunks', chunk_count,
+      'document_images', image_count,
+      'document_table_facts', table_fact_count,
+      'document_embedding_fields', embedding_field_count,
+      'document_index_units', index_unit_count,
+      'document_memory_cards', memory_card_count,
+      'document_sections', section_count
+    )
+  );
+end;
+$$;
+
+revoke execute on function public.cleanup_abandoned_document_index_generations(uuid, integer, boolean) from public, anon, authenticated;
+grant execute on function public.cleanup_abandoned_document_index_generations(uuid, integer, boolean) to service_role;
diff --git a/supabase/schema.sql b/supabase/schema.sql
index 02bbe5a34..b5481e9d5 100644
--- a/supabase/schema.sql
+++ b/supabase/schema.sql
@@ -1642,6 +1642,22 @@ begin
     and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '');
 
   if not coalesce(p_dry_run, true) then
+    -- Audit R23: the pending/processing-job guard was only evaluated during
+    -- candidate selection, before the seven count statements above. A reindex
+    -- enqueued and claimed in that window would have its freshly-staged (still
+    -- uncommitted, generation-mismatched) rows deleted mid-build. Re-filter to
+    -- documents that STILL have no open job immediately before deleting, so the
+    -- exposure window shrinks to this block instead of spanning candidate
+    -- selection plus the counts.
+    select coalesce(array_agg(doc_id), '{}'::uuid[])
+    into target_document_ids
+    from unnest(target_document_ids) as doc_id
+    where not exists (
+      select 1 from public.ingestion_jobs j
+      where j.document_id = doc_id
+        and j.status in ('pending', 'processing')
+    );
+
     delete from public.document_chunks c
     using public.documents d
     where d.id = c.document_id
@@ -1725,6 +1741,14 @@ begin
     return jsonb_build_object('ok', false, 'reason', 'missing_batch_id');
   end if;
 
+  -- Audit R9: two jobs of the same batch finishing in overlapping transactions
+  -- each counted the batch before the other committed, so both computed
+  -- 'processing' and the second write pinned the batch as processing forever.
+  -- Locking the import_batches row first serializes concurrent refreshes: the
+  -- second caller blocks here until the first commits, then its count below
+  -- sees the first's committed job state.
+  perform 1 from public.import_batches where id = p_batch_id for update;
+
   select
     count(*) filter (where status = 'pending'),
     count(*) filter (where status = 'processing'),
@@ -1760,13 +1784,21 @@ create or replace function public.complete_ingestion_job(
   p_job_id uuid,
   p_document_id uuid,
   p_batch_id uuid default null,
-  p_stage text default 'indexed'
+  p_stage text default 'indexed',
+  p_worker_id text default null
 )
 returns jsonb
 language plpgsql
 set search_path = public, extensions, pg_temp
 as $$
+declare
+  v_matched integer;
 begin
+  -- Audit R1/R2: when the caller passes its worker id, fence the completion on
+  -- `locked_by = p_worker_id`. A worker that lost its lease to a stale reclaim
+  -- (or a resumed zombie) then matches 0 rows and returns lease_lost WITHOUT
+  -- superseding siblings or touching the batch — the reclaimer owns the outcome.
+  -- p_worker_id null preserves the pre-fence behavior for older callers.
   update public.ingestion_jobs
   set
     status = 'completed',
@@ -1777,7 +1809,13 @@ begin
     locked_by = null,
     completed_at = now()
   where id = p_job_id
-    and document_id = p_document_id;
+    and document_id = p_document_id
+    and (p_worker_id is null or locked_by = p_worker_id);
+  get diagnostics v_matched = row_count;
+
+  if p_worker_id is not null and v_matched = 0 then
+    return jsonb_build_object('ok', false, 'reason', 'lease_lost', 'job_id', p_job_id, 'document_id', p_document_id);
+  end if;
 
   update public.ingestion_jobs
   set
@@ -1808,13 +1846,36 @@ create or replace function public.fail_or_retry_ingestion_job(
   p_document_status text default 'failed',
   p_stage text default 'failed',
   p_error_message text default null,
-  p_next_run_at timestamptz default null
+  p_next_run_at timestamptz default null,
+  p_worker_id text default null
 )
 returns jsonb
 language plpgsql
 set search_path = public, extensions, pg_temp
 as $$
+declare
+  v_job public.ingestion_jobs%rowtype;
+  v_retry boolean;
 begin
+  -- Lock the job row so the ownership check and the write are atomic.
+  select * into v_job
+  from public.ingestion_jobs
+  where id = p_job_id and document_id = p_document_id
+  for update;
+
+  -- Audit R1/R2: a fenced caller (p_worker_id set) must still hold the lease,
+  -- otherwise a resumed zombie could demote a document the reclaimer already
+  -- indexed. Return lease_lost without touching the document or job.
+  if p_worker_id is not null and (v_job.id is null or v_job.locked_by is distinct from p_worker_id) then
+    return jsonb_build_object('ok', false, 'reason', 'lease_lost', 'job_id', p_job_id, 'document_id', p_document_id, 'retry', false);
+  end if;
+
+  -- Audit R7: re-pending a job whose attempt_count already reached max_attempts
+  -- would strand it as permanently-unclaimable 'pending' (claim requires
+  -- attempt_count < max_attempts), pinning its batch as 'processing' forever.
+  -- Downgrade such a retry to terminal 'failed' so the queue can drain.
+  v_retry := p_retry and coalesce(v_job.attempt_count, 0) < coalesce(v_job.max_attempts, 0);
+
   update public.documents
   set
     status = p_document_status,
@@ -1823,14 +1884,14 @@ begin
 
   update public.ingestion_jobs
   set
-    status = case when p_retry then 'pending' else 'failed' end,
+    status = case when v_retry then 'pending' else 'failed' end,
     stage = p_stage,
-    progress = case when p_retry then 0 else 100 end,
+    progress = case when v_retry then 0 else 100 end,
     error_message = p_error_message,
     locked_at = null,
     locked_by = null,
     next_run_at = coalesce(p_next_run_at, next_run_at),
-    completed_at = case when p_retry then null else now() end
+    completed_at = case when v_retry then null else now() end
   where id = p_job_id
     and document_id = p_document_id;
 
@@ -1838,7 +1899,7 @@ begin
     perform public.refresh_import_batch_status(p_batch_id);
   end if;
 
-  return jsonb_build_object('ok', true, 'job_id', p_job_id, 'document_id', p_document_id, 'retry', p_retry);
+  return jsonb_build_object('ok', true, 'job_id', p_job_id, 'document_id', p_document_id, 'retry', v_retry);
 end;
 $$;
 
diff --git a/worker/main.ts b/worker/main.ts
index 60e24d548..55c339601 100644
--- a/worker/main.ts
+++ b/worker/main.ts
@@ -229,15 +229,23 @@ async function updateBatch(batchId: string | null) {
 }
 
 async function completeJob(job: JobRow, stage: string) {
-  const { error } = await supabase.rpc("complete_ingestion_job", {
+  const { data, error } = await supabase.rpc("complete_ingestion_job", {
     p_job_id: job.id,
     p_document_id: job.document_id,
     // SQL default for p_batch_id is null, so omitting the key when batch_id
     // is null sends the same value the explicit null did.
     p_batch_id: job.batch_id ?? undefined,
     p_stage: stage,
+    p_worker_id: workerId,
   });
   if (!error) {
+    // Audit R1: the RPC returns ok:false when this worker no longer holds the
+    // lease (a stale reclaim took it). The reclaiming worker owns the outcome —
+    // do not fall back or clobber its state.
+    if ((data as { ok?: boolean } | null)?.ok === false) {
+      console.warn("Ingestion completion skipped; lease lost to a reclaim", safeIngestionJobLog(job.id));
+      return;
+    }
     invalidateRagCachesForDocumentMutation(job.documents.owner_id ?? "anonymous");
     return;
   }
@@ -297,7 +305,7 @@ async function failOrRetryJob(args: {
   errorMessage: string;
   nextRunAt?: string;
 }) {
-  const { error } = await supabase.rpc("fail_or_retry_ingestion_job", {
+  const { data, error } = await supabase.rpc("fail_or_retry_ingestion_job", {
     p_job_id: args.job.id,
     p_document_id: args.job.document_id,
     p_batch_id: args.job.batch_id ?? undefined,
@@ -306,8 +314,17 @@ async function failOrRetryJob(args: {
     p_stage: args.stage,
     p_error_message: args.errorMessage,
     p_next_run_at: args.nextRunAt ?? undefined,
+    p_worker_id: workerId,
   });
-  if (!error) return;
+  if (!error) {
+    // Audit R1: ok:false means this worker lost the lease; the reclaimer owns
+    // the document/job state, so do not fall back and demote it.
+    if ((data as { ok?: boolean } | null)?.ok === false) {
+      console.warn("Ingestion fail/retry skipped; lease lost to a reclaim", safeIngestionJobLog(args.job.id));
+      return;
+    }
+    return;
+  }
   if (!isMissingSchemaError(error)) throw supabaseStageError("fail or retry ingestion job", error);
 
   await updateDocument(args.job.document_id, { status: args.documentStatus, error_message: args.errorMessage });