diff --git a/.impeccable/hook.cache.json b/.impeccable/hook.cache.json
new file mode 100644
index 0000000000..a2d4e16581
--- /dev/null
+++ b/.impeccable/hook.cache.json
@@ -0,0 +1 @@
+{"version":1,"sessions":{"70c96a73-49a6-4422-a310-d0294a45dc49":{"updatedAt":1782977742215,"files":{"C:\\Users\\joshs\\.copilot\\repos\\copilot-worktrees\\Database\\bigsimmo-bookish-barnacle\\supabase\\functions\\indexing-v3-agent\\index.ts":{"editCount":1,"findings":[]}}}}}
\ No newline at end of file
diff --git a/supabase/functions/indexing-v3-agent/index.ts b/supabase/functions/indexing-v3-agent/index.ts
index 1a6025edef..aa66e93543 100644
--- a/supabase/functions/indexing-v3-agent/index.ts
+++ b/supabase/functions/indexing-v3-agent/index.ts
@@ -123,6 +123,8 @@ type ChunkSectionSource = {
   content: string;
 };
 
+type AgentJobStatus = "pending" | "completed" | "failed" | "needs_enrichment_artifacts";
+
 const GENERATED_BY = "indexing-v3-agent";
 const AGENT_SECRET = Deno.env.get("INDEXING_V3_AGENT_SECRET") ?? Deno.env.get("CRON_SECRET") ?? "";
 const EXPECTED_EMBED_DIM = 1536;
@@ -1874,6 +1876,26 @@ async function needsVisualArtifacts(job: ClaimedJob): Promise<boolean> {
   return shouldRunVisualArtifacts(row);
 }
 
+async function updateAgentJobStatus(
+  job: ClaimedJob,
+  status: AgentJobStatus,
+  error: string | null = null,
+  nextRunAt: string | null = null,
+): Promise<void> {
+  const rows = await sql<Array<{ ok: boolean }>>`
+    select *
+    from public.update_indexing_v3_agent_job_status(
+      ${job.document_id}::uuid,
+      ${status}::text,
+      ${error}::text,
+      ${nextRunAt}::timestamptz
+    )
+  `;
+  if (!rows[0]?.ok) {
+    throw new Error(`Failed to update indexing_v3_agent_jobs status to ${status} for document ${job.document_id}`);
+  }
+}
+
 function logCompletionGate(job: ClaimedJob, gate: CompletionGate): void {
   console.log(
     JSON.stringify({
@@ -1921,6 +1943,12 @@ async function deferJob(job: ClaimedJob, gate: CompletionGate): Promise<void> {
       updated_at = now()
     where id = ${job.document_id}::uuid
   `;
+  await updateAgentJobStatus(
+    job,
+    decision.status === "needs_enrichment_artifacts" ? "needs_enrichment_artifacts" : "pending",
+    null,
+    decision.status === "needs_enrichment_artifacts" ? null : decision.next_run_at,
+  );
 }
 
 async function completeJob(job: ClaimedJob): Promise<void> {
@@ -1943,6 +1971,7 @@ async function completeJob(job: ClaimedJob): Promise<void> {
       })}`,
     );
   }
+  await updateAgentJobStatus(job, "completed");
 }
 
 async function markJobFailure(job: ClaimedJob, message: string): Promise<boolean> {
@@ -1971,6 +2000,7 @@ async function markJobFailure(job: ClaimedJob, message: string): Promise<boolean
       updated_at = now()
     where id = ${job.document_id}::uuid
   `;
+  await updateAgentJobStatus(job, shouldRetry ? "pending" : "failed", message, nextRunAt);
   return shouldRetry;
 }
 
diff --git a/supabase/migrations/20260702100000_add_claim_ingestion_jobs_comment.sql b/supabase/migrations/20260702100000_add_claim_ingestion_jobs_comment.sql
new file mode 100644
index 0000000000..4acb013f9c
--- /dev/null
+++ b/supabase/migrations/20260702100000_add_claim_ingestion_jobs_comment.sql
@@ -0,0 +1,11 @@
+-- Fix #12: Add explanatory comment to claim_ingestion_jobs explaining the
+-- dual-lock (FOR UPDATE OF j, d SKIP LOCKED) pattern. No behaviour change.
+
+comment on function public.claim_ingestion_jobs(text, integer, integer) is
+  'Claims up to p_limit pending/failed ingestion jobs for the given worker_id.
+   Uses "FOR UPDATE OF j, d SKIP LOCKED" to lock both the ingestion_job row (j)
+   and the parent document row (d) in a single CTE scan. Locking the document
+   prevents two concurrent workers from racing on the same document even when
+   separate ingestion jobs reference it (e.g. a retry and a re-queue arriving
+   simultaneously). SKIP LOCKED ensures a busy document is silently bypassed
+   rather than causing a block, giving other workers fair access.';
diff --git a/supabase/migrations/20260702110000_drop_redundant_indexes.sql b/supabase/migrations/20260702110000_drop_redundant_indexes.sql
new file mode 100644
index 0000000000..d2fa7660f9
--- /dev/null
+++ b/supabase/migrations/20260702110000_drop_redundant_indexes.sql
@@ -0,0 +1,27 @@
+-- Fix #9: Drop redundant indexes.
+--
+-- 1. documents_owner_hash_idx (owner_id, content_hash) is a plain non-unique
+--    index whose column set is a strict subset of the UNIQUE partial index
+--    documents_owner_content_hash_unique_idx (owner_id, content_hash WHERE
+--    content_hash IS NOT NULL). The unique index is used for duplicate detection
+--    (ON CONFLICT) and also satisfies all equality lookups on (owner_id,
+--    content_hash). The plain index adds write overhead with no read benefit.
+--
+-- NOTE: DROP INDEX CONCURRENTLY cannot run inside a transaction block.
+-- Supabase migrations are wrapped in a transaction by default, which means
+-- CONCURRENTLY is not available here. We use a plain DROP INDEX instead;
+-- the table is small relative to write load and this is a one-time maintenance
+-- operation. If you prefer zero-impact removal, run this statement manually in
+-- the Supabase SQL editor outside a transaction:
+--   DROP INDEX CONCURRENTLY IF EXISTS public.documents_owner_hash_idx;
+
+drop index if exists public.documents_owner_hash_idx;
+
+-- 2. ingestion_jobs_claim_idx covers (status, next_run_at, created_at) WHERE
+--    status IN ('pending','processing'). The superset index
+--    ingestion_jobs_status_next_run_idx covers the same columns with WHERE
+--    status IN ('pending','processing','failed'). PostgreSQL can use the
+--    superset index for any query the subset index would satisfy, so the subset
+--    index is fully redundant once the superset exists.
+
+drop index if exists public.ingestion_jobs_claim_idx;
diff --git a/supabase/migrations/20260702120000_rag_retrieval_logs_retention.sql b/supabase/migrations/20260702120000_rag_retrieval_logs_retention.sql
new file mode 100644
index 0000000000..53760b850b
--- /dev/null
+++ b/supabase/migrations/20260702120000_rag_retrieval_logs_retention.sql
@@ -0,0 +1,28 @@
+-- Fix #6: Add retention policy for rag_retrieval_logs and document audit_logs intent.
+--
+-- rag_queries already has a purge cron (20260629100000). rag_retrieval_logs has
+-- no TTL. We add a matching cron to purge rows older than 90 days.
+-- audit_logs is intentionally kept indefinitely (compliance requirement); add a
+-- comment so this is self-documenting and not mistaken for an oversight.
+
+comment on table public.audit_logs is
+  'Append-only audit trail. Rows are retained indefinitely for compliance.
+   Writes are best-effort (fire-and-forget) from the application layer; a write
+   failure is swallowed and does not affect the calling request.
+   Do NOT add an automatic purge to this table without a compliance review.';
+
+comment on table public.rag_retrieval_logs is
+  'Per-request retrieval telemetry. Rows older than 90 days are purged nightly
+   by the pg_cron job "purge-rag-retrieval-logs". Adjust the retention window
+   by changing the interval in that cron job definition.';
+
+-- Register the nightly purge cron job.
+-- cron.schedule is idempotent on the job name; re-running this migration is safe.
+select cron.schedule(
+  'purge-rag-retrieval-logs',
+  '0 3 * * *',   -- 03:00 UTC daily
+  $$
+    delete from public.rag_retrieval_logs
+    where created_at < now() - interval '90 days';
+  $$
+);
diff --git a/supabase/migrations/20260702130000_storage_cleanup_jobs_document_fk.sql b/supabase/migrations/20260702130000_storage_cleanup_jobs_document_fk.sql
new file mode 100644
index 0000000000..4c043c2af4
--- /dev/null
+++ b/supabase/migrations/20260702130000_storage_cleanup_jobs_document_fk.sql
@@ -0,0 +1,21 @@
+-- Fix #3: Add foreign key from storage_cleanup_jobs.document_id to documents(id).
+--
+-- The column was declared as "uuid" with no referential constraint, meaning
+-- orphaned rows (pointing to deleted documents) would accumulate silently.
+-- We first delete any orphaned rows, then add ON DELETE SET NULL so future
+-- document deletions leave the cleanup job record in place (the worker should
+-- still attempt storage cleanup for any paths recorded before deletion).
+
+-- Step 1: remove orphans (document_id is non-null but no matching document exists).
+delete from public.storage_cleanup_jobs
+where document_id is not null
+  and not exists (
+    select 1 from public.documents d where d.id = storage_cleanup_jobs.document_id
+  );
+
+-- Step 2: add the FK constraint.
+alter table public.storage_cleanup_jobs
+  add constraint storage_cleanup_jobs_document_id_fkey
+  foreign key (document_id)
+  references public.documents(id)
+  on delete set null;
diff --git a/supabase/migrations/20260702140000_fix_reset_document_index_duplicate.sql b/supabase/migrations/20260702140000_fix_reset_document_index_duplicate.sql
new file mode 100644
index 0000000000..41ed97acb0
--- /dev/null
+++ b/supabase/migrations/20260702140000_fix_reset_document_index_duplicate.sql
@@ -0,0 +1,39 @@
+-- Fix #2: Remove duplicate/incomplete reset_document_index definition.
+--
+-- schema.sql contained two definitions of reset_document_index. The first
+-- (lines 1075-1091) did NOT delete from document_index_units. The second
+-- (later in the file) does delete from document_index_units and is the
+-- authoritative live version. Because CREATE OR REPLACE is last-write-wins,
+-- the second definition is what the database actually runs. This migration
+-- is a no-op CREATE OR REPLACE of the correct complete definition, making
+-- migration history authoritative and preventing any future schema replay
+-- from accidentally deploying the incomplete version first.
+
+create or replace function public.reset_document_index(p_document_id uuid)
+returns void
+language plpgsql
+set search_path = public, extensions, pg_temp
+as $$
+begin
+  perform set_config('statement_timeout', '180000', true);
+  -- document_index_units must be deleted first (references document_chunks via FK).
+  delete from public.document_index_units where document_id = p_document_id;
+  delete from public.document_memory_cards where document_id = p_document_id;
+  delete from public.document_sections where document_id = p_document_id;
+  delete from public.document_table_facts where document_id = p_document_id;
+  delete from public.document_embedding_fields where document_id = p_document_id;
+  delete from public.document_index_quality where document_id = p_document_id;
+  delete from public.document_chunks where document_id = p_document_id;
+  delete from public.document_images where document_id = p_document_id;
+  delete from public.document_pages where document_id = p_document_id;
+end;
+$$;
+
+-- Validation: confirm the function exists with the correct signature.
+do $$
+begin
+  if to_regprocedure('public.reset_document_index(uuid)') is null then
+    raise exception 'reset_document_index(uuid) not found after migration';
+  end if;
+end;
+$$;
diff --git a/supabase/migrations/20260702150000_documents_owner_covering_index.sql b/supabase/migrations/20260702150000_documents_owner_covering_index.sql
new file mode 100644
index 0000000000..e8388222c5
--- /dev/null
+++ b/supabase/migrations/20260702150000_documents_owner_covering_index.sql
@@ -0,0 +1,21 @@
+-- Fix #5: Add covering index for RLS correlated subquery on documents.
+--
+-- Several tables (document_chunks, document_sections, document_memory_cards,
+-- document_index_units, etc.) have RLS policies of the form:
+--
+--   EXISTS (SELECT 1 FROM documents WHERE id = document_id AND owner_id = auth.uid())
+--
+-- The existing documents_owner_idx covers (owner_id) only, so PostgreSQL must
+-- re-fetch the heap to confirm id = document_id. A composite index on
+-- (owner_id, id) allows an index-only scan, eliminating the heap fetch.
+-- The index is small (two uuid columns) and benefits every authenticated read
+-- on the child tables.
+--
+-- NOTE: CONCURRENTLY cannot run inside a transaction.
+-- If you need a zero-lock creation on a loaded system, run this statement
+-- manually outside a transaction:
+--   CREATE INDEX CONCURRENTLY IF NOT EXISTS documents_owner_id_covering_idx
+--     ON public.documents(owner_id, id);
+
+create index if not exists documents_owner_id_covering_idx
+  on public.documents(owner_id, id);
diff --git a/supabase/migrations/20260702160000_fix_invoke_agent_url_to_guc.sql b/supabase/migrations/20260702160000_fix_invoke_agent_url_to_guc.sql
new file mode 100644
index 0000000000..8ef2c37b95
--- /dev/null
+++ b/supabase/migrations/20260702160000_fix_invoke_agent_url_to_guc.sql
@@ -0,0 +1,57 @@
+-- Fix #8: Replace hardcoded project URL in invoke_indexing_v3_agent with a
+-- GUC-based setting, so the function works across all environments (staging,
+-- production, local) without a code change.
+--
+-- We store the base URL in a database-level GUC (app.indexing_v3_agent_base_url).
+-- current_setting('app.indexing_v3_agent_base_url', true) returns NULL if the
+-- GUC is not set, so the function retains the current project URL as its fallback,
+-- meaning this change is fully backwards-compatible.
+
+-- Set the default base URL for the current (production) project.
+-- This value must be changed for staging/dev environments via:
+--   ALTER DATABASE postgres SET app.indexing_v3_agent_base_url = '...';
+alter database postgres
+  set app.indexing_v3_agent_base_url = 'https://sjrfecxgysukkwxsowpy.supabase.co';
+
+create or replace function public.invoke_indexing_v3_agent(p_limit integer default 1)
+returns bigint
+language plpgsql
+security definer
+set search_path = public, extensions, vault, pg_temp
+as $$
+declare
+  v_request_id bigint;
+  v_secret     text;
+  v_base_url   text;
+begin
+  select decrypted_secret
+    into v_secret
+  from vault.decrypted_secrets
+  where name = 'indexing_v3_agent_secret'
+  limit 1;
+
+  if nullif(v_secret, '') is null then
+    raise exception 'indexing_v3_agent_secret is missing from Supabase Vault';
+  end if;
+
+  -- Prefer the GUC; fall back to the hardcoded production URL so that
+  -- existing deployments that have not yet set the GUC continue to work.
+  v_base_url := coalesce(
+    nullif(current_setting('app.indexing_v3_agent_base_url', true), ''),
+    'https://sjrfecxgysukkwxsowpy.supabase.co'
+  );
+
+  select net.http_post(
+    url := v_base_url || '/functions/v1/indexing-v3-agent?limit='
+           || greatest(1, least(coalesce(p_limit, 1), 10))::text,
+    headers := jsonb_build_object(
+      'Content-Type', 'application/json',
+      'x-indexing-agent-secret', v_secret
+    ),
+    body := jsonb_build_object('source', 'pg_cron', 'worker', 'v3-indexing-worker', 'ts', now()),
+    timeout_milliseconds := 60000
+  ) into v_request_id;
+
+  return v_request_id;
+end;
+$$;
diff --git a/supabase/migrations/20260702170000_fix_match_chunks_text_n1.sql b/supabase/migrations/20260702170000_fix_match_chunks_text_n1.sql
new file mode 100644
index 0000000000..b8ed4e1430
--- /dev/null
+++ b/supabase/migrations/20260702170000_fix_match_chunks_text_n1.sql
@@ -0,0 +1,144 @@
+-- Fix #7: Eliminate N+1 scalar function calls in match_document_chunks_text.
+--
+-- The original outer SELECT called document_label_metadata(document_id) and
+-- document_summary_text(document_id) once per row returned from the `ranked`
+-- CTE. With match_count=12 and duplicate document_ids, this could fire 24+
+-- independent SELECTs against document_labels and document_summaries.
+--
+-- Fix: add two CTEs that batch-fetch the data for all distinct document_ids
+-- in the result set, then LEFT JOIN them back. chunk_image_metadata() accepts
+-- a uuid[] so it is already a single call per row (no N+1); it is unchanged.
+
+create or replace function public.match_document_chunks_text(
+  query_text text,
+  match_count integer default 12,
+  document_filters uuid[] default null,
+  owner_filter uuid default null
+)
+returns table (
+  id uuid,
+  document_id uuid,
+  title text,
+  file_name text,
+  page_number integer,
+  chunk_index integer,
+  section_heading text,
+  content text,
+  retrieval_synopsis text,
+  image_ids uuid[],
+  source_metadata jsonb,
+  document_labels jsonb,
+  document_summary text,
+  similarity double precision,
+  text_rank double precision,
+  hybrid_score double precision,
+  lexical_score double precision,
+  images jsonb
+)
+language sql
+stable
+set search_path = public, extensions, pg_temp
+as $$
+  with query as (
+    select websearch_to_tsquery('english', coalesce(query_text, '')) as tsq
+  ),
+  ranked as (
+    select
+      c.id,
+      c.document_id,
+      d.title,
+      d.file_name,
+      c.page_number,
+      c.chunk_index,
+      c.section_heading,
+      c.content,
+      c.retrieval_synopsis,
+      c.image_ids,
+      d.metadata as source_metadata,
+      (
+        ts_rank_cd(c.search_tsv, query.tsq) +
+        (ts_rank_cd(d.title_search_tsv, query.tsq) * 3.0)
+      )::double precision as text_rank
+    from public.document_chunks c
+    join public.documents d on d.id = c.document_id
+    cross join query
+    where (document_filters is null or c.document_id = any(document_filters))
+      and (owner_filter is null or d.owner_id = owner_filter)
+      and d.status = 'indexed'
+      and public.is_committed_document_generation(c.index_generation_id, d.metadata)
+      and (c.search_tsv @@ query.tsq or d.title_search_tsv @@ query.tsq)
+    order by (
+      ts_rank_cd(c.search_tsv, query.tsq) +
+      (ts_rank_cd(d.title_search_tsv, query.tsq) * 3.0)
+    ) desc
+    limit least(greatest(match_count * 2, 24), 96)
+  ),
+  -- Batch-fetch label metadata for all distinct document_ids in the result set.
+  -- One query replaces N per-row calls to document_label_metadata().
+  doc_labels as (
+    select
+      l.document_id,
+      coalesce(
+        jsonb_agg(
+          jsonb_build_object(
+            'id',          l.id,
+            'document_id', l.document_id,
+            'owner_id',    l.owner_id,
+            'label',       l.label,
+            'label_type',  l.label_type,
+            'source',      l.source,
+            'confidence',  l.confidence,
+            'metadata',    l.metadata,
+            'created_at',  l.created_at,
+            'updated_at',  l.updated_at
+          )
+          order by l.confidence desc, l.label
+        ),
+        '[]'::jsonb
+      ) as labels
+    from public.document_labels l
+    where l.document_id in (select distinct ranked.document_id from ranked)
+    group by l.document_id
+  ),
+  -- Batch-fetch summary text for all distinct document_ids in the result set.
+  -- One query replaces N per-row calls to document_summary_text().
+  doc_summaries as (
+    select distinct on (s.document_id)
+      s.document_id,
+      s.summary
+    from public.document_summaries s
+    where s.document_id in (select distinct ranked.document_id from ranked)
+    order by s.document_id
+  )
+  select
+    ranked.id,
+    ranked.document_id,
+    ranked.title,
+    ranked.file_name,
+    ranked.page_number,
+    ranked.chunk_index,
+    ranked.section_heading,
+    ranked.content,
+    ranked.retrieval_synopsis,
+    ranked.image_ids,
+    ranked.source_metadata,
+    coalesce(doc_labels.labels,   '[]'::jsonb) as document_labels,
+    doc_summaries.summary                       as document_summary,
+    -- Text-only fallback has NO vector cosine similarity. Do not fabricate one:
+    -- a synthetic value here was read downstream as a real semantic score and
+    -- could label a pure keyword hit as "strong"/"moderate" evidence (>=0.64).
+    -- Leave similarity at 0; the lexical signal lives in lexical_score.
+    0::double precision                                                              as similarity,
+    ranked.text_rank,
+    -- Cap hybrid_score well below the 0.64 "moderate" threshold so a lexical-only
+    -- row can order amongst its peers but can never masquerade as a moderate/strong
+    -- cosine match when merged with vector results.
+    least(0.5,  0.18 + (least(ranked.text_rank, 1) * 0.3))::double precision       as hybrid_score,
+    least(0.99, 0.4  + (least(ranked.text_rank, 1) * 0.59))::double precision      as lexical_score,
+    public.chunk_image_metadata(ranked.image_ids)                                   as images
+  from ranked
+  left join doc_labels    on doc_labels.document_id    = ranked.document_id
+  left join doc_summaries on doc_summaries.document_id = ranked.document_id
+  order by lexical_score desc, text_rank desc
+  limit match_count;
+$$;
diff --git a/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql b/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql
new file mode 100644
index 0000000000..bef99a6d74
--- /dev/null
+++ b/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql
@@ -0,0 +1,618 @@
+-- Fix #4: Promote index_generation_id from JSONB metadata to typed UUID columns
+-- in the 6 artifact tables that lag behind document_chunks.
+--
+-- document_chunks already carries a typed index_generation_id uuid column, allowing
+-- fast index-only scans during commit and cleanup. The other 6 artifact tables
+-- (document_images, document_table_facts, document_embedding_fields,
+--  document_index_units, document_memory_cards, document_sections) still fish the
+-- value out of a JSONB blob, forcing a full table scan for commit/cleanup DELETEs.
+--
+-- What this migration does:
+--   1. ADD COLUMN index_generation_id uuid to each of the 6 tables
+--   2. Backfill from metadata->>'index_generation_id'
+--   3. Add partial indexes (document_id, index_generation_id) WHERE index_generation_id IS NOT NULL
+--   4. Add an overloaded is_committed_artifact_generation(uuid, jsonb) helper
+--      matching the existing is_committed_document_generation(uuid, jsonb) pattern
+--   5. Rewrite commit_document_index_generation DELETEs to use the typed column
+--   6. Rewrite cleanup_abandoned_document_index_generations to use the typed column
+--
+-- BACKWARD COMPATIBILITY: The original JSONB-based
+--   is_committed_artifact_generation(jsonb, jsonb) overload is preserved.
+--   Query functions that call it continue to work unchanged. They can be
+--   migrated to the new (uuid, jsonb) overload in a follow-up migration.
+
+-- -------------------------------------------------------------------------
+-- Step 1: Add typed columns
+-- -------------------------------------------------------------------------
+
+alter table public.document_images
+  add column if not exists index_generation_id uuid;
+
+alter table public.document_table_facts
+  add column if not exists index_generation_id uuid;
+
+alter table public.document_embedding_fields
+  add column if not exists index_generation_id uuid;
+
+alter table public.document_index_units
+  add column if not exists index_generation_id uuid;
+
+alter table public.document_memory_cards
+  add column if not exists index_generation_id uuid;
+
+alter table public.document_sections
+  add column if not exists index_generation_id uuid;
+
+-- -------------------------------------------------------------------------
+-- Step 2: Backfill from existing JSONB metadata (NULL-safe cast)
+-- -------------------------------------------------------------------------
+
+update public.document_images
+set index_generation_id = (metadata->>'index_generation_id')::uuid
+where index_generation_id is null
+  and nullif(metadata->>'index_generation_id', '') is not null;
+
+update public.document_table_facts
+set index_generation_id = (metadata->>'index_generation_id')::uuid
+where index_generation_id is null
+  and nullif(metadata->>'index_generation_id', '') is not null;
+
+update public.document_embedding_fields
+set index_generation_id = (metadata->>'index_generation_id')::uuid
+where index_generation_id is null
+  and nullif(metadata->>'index_generation_id', '') is not null;
+
+update public.document_index_units
+set index_generation_id = (metadata->>'index_generation_id')::uuid
+where index_generation_id is null
+  and nullif(metadata->>'index_generation_id', '') is not null;
+
+update public.document_memory_cards
+set index_generation_id = (metadata->>'index_generation_id')::uuid
+where index_generation_id is null
+  and nullif(metadata->>'index_generation_id', '') is not null;
+
+update public.document_sections
+set index_generation_id = (metadata->>'index_generation_id')::uuid
+where index_generation_id is null
+  and nullif(metadata->>'index_generation_id', '') is not null;
+
+-- -------------------------------------------------------------------------
+-- Step 3: Partial indexes on (document_id, index_generation_id)
+--         WHERE index_generation_id IS NOT NULL
+--         Mirrors document_chunks_document_generation_chunk_idx pattern.
+-- -------------------------------------------------------------------------
+
+create index if not exists document_images_document_generation_idx
+  on public.document_images(document_id, index_generation_id)
+  where index_generation_id is not null;
+
+create index if not exists document_table_facts_document_generation_idx
+  on public.document_table_facts(document_id, index_generation_id)
+  where index_generation_id is not null;
+
+create index if not exists document_embedding_fields_document_generation_idx
+  on public.document_embedding_fields(document_id, index_generation_id)
+  where index_generation_id is not null;
+
+create index if not exists document_index_units_document_generation_idx
+  on public.document_index_units(document_id, index_generation_id)
+  where index_generation_id is not null;
+
+create index if not exists document_memory_cards_document_generation_idx
+  on public.document_memory_cards(document_id, index_generation_id)
+  where index_generation_id is not null;
+
+create index if not exists document_sections_document_generation_idx
+  on public.document_sections(document_id, index_generation_id)
+  where index_generation_id is not null;
+
+-- -------------------------------------------------------------------------
+-- Step 4: New overload is_committed_artifact_generation(uuid, jsonb)
+--         Mirrors is_committed_document_generation(uuid, jsonb).
+--         Returns true if artifact_generation_id is null (uncommitted/legacy)
+--         OR if it matches the document's committed generation.
+-- -------------------------------------------------------------------------
+
+create or replace function public.is_committed_artifact_generation(
+  artifact_generation_id uuid,
+  document_metadata jsonb
+)
+returns boolean
+language sql
+stable
+set search_path = public, extensions, pg_temp
+as $$
+  select artifact_generation_id is null
+    or artifact_generation_id::text =
+       nullif(coalesce(document_metadata, '{}'::jsonb)->>'index_generation_id', '');
+$$;
+
+revoke execute on function public.is_committed_artifact_generation(uuid, jsonb) from public, anon, authenticated;
+grant execute on function public.is_committed_artifact_generation(uuid, jsonb) to service_role;
+
+-- -------------------------------------------------------------------------
+-- Step 5: Update commit_document_index_generation to use typed columns
+-- -------------------------------------------------------------------------
+
+create or replace function public.commit_document_index_generation(
+  p_document_id uuid,
+  p_index_generation_id uuid,
+  p_status text default 'indexed',
+  p_page_count integer default 0,
+  p_chunk_count integer default 0,
+  p_image_count integer default 0,
+  p_metadata jsonb default '{}'::jsonb,
+  p_pages jsonb default null,
+  p_quality jsonb default null
+)
+returns jsonb
+language plpgsql
+set search_path = public, extensions, pg_temp
+as $$
+begin
+  perform set_config('statement_timeout', '180000', true);
+
+  update public.documents
+  set
+    status = p_status,
+    page_count = p_page_count,
+    chunk_count = p_chunk_count,
+    image_count = p_image_count,
+    error_message = null,
+    metadata = coalesce(p_metadata, '{}'::jsonb) || jsonb_build_object('index_generation_id', p_index_generation_id),
+    updated_at = now()
+  where id = p_document_id;
+
+  if p_pages is not null then
+    delete from public.document_pages
+    where document_id = p_document_id;
+
+    insert into public.document_pages (document_id, page_number, text, ocr_used, metadata)
+    select
+      p_document_id,
+      page_row.page_number,
+      coalesce(page_row.text, ''),
+      coalesce(page_row.ocr_used, false),
+      coalesce(page_row.metadata, '{}'::jsonb)
+    from jsonb_to_recordset(coalesce(p_pages, '[]'::jsonb)) as page_row(
+      page_number integer,
+      text text,
+      ocr_used boolean,
+      metadata jsonb
+    )
+    where page_row.page_number is not null;
+  end if;
+
+  if p_quality is not null then
+    insert into public.document_index_quality (
+      document_id,
+      owner_id,
+      quality_score,
+      extraction_quality,
+      metrics,
+      issues,
+      updated_at
+    )
+    values (
+      p_document_id,
+      nullif(p_quality->>'owner_id', '')::uuid,
+      coalesce((p_quality->>'quality_score')::real, 0),
+      coalesce(nullif(p_quality->>'extraction_quality', ''), 'unknown'),
+      coalesce(p_quality->'metrics', '{}'::jsonb),
+      coalesce(
+        array(select jsonb_array_elements_text(coalesce(p_quality->'issues', '[]'::jsonb))),
+        '{}'::text[]
+      ),
+      now()
+    )
+    on conflict on constraint document_index_quality_pkey
+    do update set
+      owner_id = excluded.owner_id,
+      quality_score = excluded.quality_score,
+      extraction_quality = excluded.extraction_quality,
+      metrics = excluded.metrics,
+      issues = excluded.issues,
+      updated_at = excluded.updated_at;
+  end if;
+
+  -- Preserve legacy NULL-generation rows unless this generation wrote replacements.
+  delete from public.document_chunks
+  where document_id = p_document_id
+    and (
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
+      or (
+        index_generation_id is null
+        and exists (
+          select 1
+          from public.document_chunks replacement
+          where replacement.document_id = p_document_id
+            and replacement.index_generation_id = p_index_generation_id
+        )
+      )
+    );
+
+  -- artifact tables: use typed column where set; fall back to metadata when typed is NULL
+  -- because the writer still populates metadata.index_generation_id rather than the typed
+  -- column.  Without the metadata fallback, stale null-typed rows from a prior run would
+  -- never be cleaned up (the typed-column EXISTS guard would always be false), allowing
+  -- artifact rows to accumulate across re-indexes.
+  delete from public.document_images
+  where document_id = p_document_id
+    and (
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
+      or (
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
+        and exists (
+          select 1
+          from public.document_images replacement
+          where replacement.document_id = p_document_id
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
+        )
+      )
+    );
+
+  delete from public.document_table_facts
+  where document_id = p_document_id
+    and (
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
+      or (
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
+        and exists (
+          select 1
+          from public.document_table_facts replacement
+          where replacement.document_id = p_document_id
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
+        )
+      )
+    );
+
+  delete from public.document_embedding_fields
+  where document_id = p_document_id
+    and (
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
+      or (
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
+        and exists (
+          select 1
+          from public.document_embedding_fields replacement
+          where replacement.document_id = p_document_id
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
+        )
+      )
+    );
+
+  delete from public.document_index_units
+  where document_id = p_document_id
+    and (
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
+      or (
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
+        and exists (
+          select 1
+          from public.document_index_units replacement
+          where replacement.document_id = p_document_id
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
+        )
+      )
+    );
+
+  delete from public.document_memory_cards
+  where document_id = p_document_id
+    and (
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
+      or (
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
+        and exists (
+          select 1
+          from public.document_memory_cards replacement
+          where replacement.document_id = p_document_id
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
+        )
+      )
+    );
+
+  delete from public.document_sections
+  where document_id = p_document_id
+    and (
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
+      or (
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
+        and exists (
+          select 1
+          from public.document_sections replacement
+          where replacement.document_id = p_document_id
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
+        )
+      )
+    );
+
+  return jsonb_build_object(
+    'ok', true,
+    'document_id', p_document_id,
+    'index_generation_id', p_index_generation_id
+  );
+end;
+$$;
+
+revoke execute on function public.commit_document_index_generation(uuid, uuid, text, integer, integer, integer, jsonb, jsonb, jsonb) from public, anon, authenticated;
+grant execute on function public.commit_document_index_generation(uuid, uuid, text, integer, integer, integer, jsonb, jsonb, jsonb) to service_role;
+
+-- -------------------------------------------------------------------------
+-- Step 6: Update cleanup_abandoned_document_index_generations
+-- -------------------------------------------------------------------------
+
+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);
+
+  -- Collect distinct document_ids that have stale (non-committed) artifact rows.
+  -- document_chunks uses its typed column; artifact tables use their new typed columns.
+  with candidate_documents as (
+    select distinct document_id
+    from (
+      -- document_chunks (typed index_generation_id)
+      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
+      -- document_images (typed index_generation_id)
+      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
+      -- document_table_facts (typed index_generation_id)
+      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
+      -- document_embedding_fields (typed index_generation_id)
+      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
+      -- document_index_units (typed index_generation_id)
+      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
+      -- document_memory_cards (typed index_generation_id)
+      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
+      -- document_sections (typed index_generation_id)
+      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;
+
+  -- Count stale rows (typed column comparisons)
+  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
+    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/migrations/20260702190000_indexing_v3_agent_jobs_table.sql b/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql
new file mode 100644
index 0000000000..7d3f156ba0
--- /dev/null
+++ b/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql
@@ -0,0 +1,383 @@
+-- Fix #1: Replace documents.metadata JSONB worker state with a dedicated
+--          indexing_v3_agent_jobs table for the v3 enrichment pipeline.
+--
+-- PROBLEM:
+--   claim_indexing_v3_agent_jobs does FOR UPDATE SKIP LOCKED on the documents
+--   table — the largest table in the schema — and must parse multiple JSONB
+--   fields for every candidate row. At moderate scale this is a multi-second
+--   sequential scan. Partial indexes on JSONB expressions help for
+--   documents_indexing_v3_agent_claim_idx, but the state is still scattered
+--   across ~9 JSONB keys in metadata.
+--
+-- FIX:
+--   1. Create indexing_v3_agent_jobs with proper typed columns and a compound
+--      index suited for SKIP LOCKED claiming.
+--   2. Seed the table from existing JSONB state for all documents that are not
+--      yet completed.
+--   3. Rewrite claim_indexing_v3_agent_jobs to SELECT FOR UPDATE SKIP LOCKED
+--      on the small jobs table rather than the documents table.
+--      The RPC also patches documents.metadata to maintain backward
+--      compatibility with any edge function code that still reads from JSONB.
+--
+-- *** CRITICAL EDGE FUNCTION NOTE ***
+--   The supabase/functions/indexing-v3-agent/ edge function currently writes
+--   completion/failure state back to documents.metadata directly.
+--   Once this migration is applied, those JSONB writes are still safe (they
+--   do not break anything), but the jobs table row will NOT be updated by
+--   them and will remain stuck in 'processing' until it becomes stale and is
+--   re-claimed or manually updated.
+--
+--   You MUST update the edge function to also call a completion/failure RPC
+--   (or UPDATE indexing_v3_agent_jobs directly) after applying this migration.
+--   Until then, completed jobs will be picked up again after the stale timeout
+--   (p_stale_after_minutes, default 45), wasting agent cycles but not
+--   corrupting data (commit_document_index_generation is idempotent).
+--
+--   Recommended follow-up:
+--     - Add update_indexing_v3_agent_job_status(document_id, status, error)
+--       RPC (service_role-only) and call it from the edge function on
+--       success/failure/backoff.
+--     - Remove the documents.metadata JSONB sync from
+--       claim_indexing_v3_agent_jobs once the edge function is updated.
+
+-- -------------------------------------------------------------------------
+-- Step 1: Create the dedicated jobs table
+-- -------------------------------------------------------------------------
+
+create table if not exists public.indexing_v3_agent_jobs (
+  id uuid primary key default gen_random_uuid(),
+  document_id uuid not null references public.documents(id) on delete cascade,
+  -- v3 agent processing status (mirrors metadata->>'indexing_v3_agent_status')
+  status text not null default 'pending'
+    check (status in ('pending', 'processing', 'completed', 'failed', 'needs_enrichment_artifacts')),
+  -- enrichment pipeline status (mirrors metadata->>'enrichment_status')
+  enrichment_status text not null default 'pending'
+    check (enrichment_status in ('pending', 'processing', 'completed', 'failed', 'needs_enrichment_artifacts')),
+  attempt_count integer not null default 0,
+  max_attempts integer not null default 3,
+  locked_by text,
+  locked_at timestamptz,
+  next_run_at timestamptz,
+  version text not null default 'visual-core-v3',
+  last_error text,
+  metadata jsonb not null default '{}'::jsonb,
+  created_at timestamptz not null default now(),
+  updated_at timestamptz not null default now()
+);
+
+-- One row per document; re-running resets the row in-place
+create unique index if not exists indexing_v3_agent_jobs_document_id_idx
+  on public.indexing_v3_agent_jobs(document_id);
+
+-- Hot path for claim: eligible candidates ordered by next_run_at
+create index if not exists indexing_v3_agent_jobs_claim_idx
+  on public.indexing_v3_agent_jobs(status, enrichment_status, next_run_at, id)
+  where status not in ('completed', 'needs_enrichment_artifacts');
+
+-- Operational: find stale processing jobs
+create index if not exists indexing_v3_agent_jobs_locked_at_idx
+  on public.indexing_v3_agent_jobs(locked_at)
+  where status = 'processing';
+
+-- RLS + grants (service_role only, same as ingestion_jobs)
+alter table public.indexing_v3_agent_jobs enable row level security;
+
+drop policy if exists "indexing v3 agent jobs service role all" on public.indexing_v3_agent_jobs;
+create policy "indexing v3 agent jobs service role all"
+  on public.indexing_v3_agent_jobs
+  for all to service_role
+  using (true)
+  with check (true);
+
+grant select, insert, update, delete
+  on table public.indexing_v3_agent_jobs to service_role;
+
+-- -------------------------------------------------------------------------
+-- Step 2: Seed from existing JSONB state
+--         Insert one row per document that has ever been touched by the
+--         v3 agent (i.e., has indexing_v3_agent_status in metadata) and
+--         hasn't completed. Documents with no JSONB keys are not yet
+--         eligible and will get a row on their first claim.
+-- -------------------------------------------------------------------------
+
+insert into public.indexing_v3_agent_jobs (
+  document_id,
+  status,
+  enrichment_status,
+  attempt_count,
+  max_attempts,
+  locked_by,
+  locked_at,
+  next_run_at,
+  version,
+  last_error,
+  metadata,
+  created_at,
+  updated_at
+)
+select
+  d.id,
+  case
+    when coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') in
+         ('completed', 'needs_enrichment_artifacts', 'failed')
+      then coalesce(d.metadata->>'indexing_v3_agent_status', 'pending')
+    when coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') in
+         ('deferred', 'retry_pending')
+      then 'pending'
+    when coalesce(d.metadata->>'indexing_v3_agent_status', '') = 'processing'
+      and (
+        nullif(d.metadata->>'indexing_v3_agent_locked_at', '') is null
+        or (d.metadata->>'indexing_v3_agent_locked_at')::timestamptz < now() - interval '2 hours'
+      )
+      then 'pending'  -- stale processing → reset to pending
+    else 'pending'
+  end as status,
+  coalesce(d.metadata->>'enrichment_status', 'pending') as enrichment_status,
+  case
+    when coalesce(d.metadata->>'indexing_v3_agent_attempt_count', '') ~ '^[0-9]+$'
+      then (d.metadata->>'indexing_v3_agent_attempt_count')::integer
+    else 0
+  end as attempt_count,
+  greatest(
+    case
+      when coalesce(d.metadata->>'indexing_v3_agent_max_attempts', '') ~ '^[0-9]+$'
+        then (d.metadata->>'indexing_v3_agent_max_attempts')::integer
+      else 3
+    end,
+    1
+  ) as max_attempts,
+  nullif(d.metadata->>'indexing_v3_agent_locked_by', '') as locked_by,
+  case
+    when coalesce(d.metadata->>'indexing_v3_agent_locked_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
+      then (d.metadata->>'indexing_v3_agent_locked_at')::timestamptz
+    else null
+  end as locked_at,
+  case
+    when coalesce(d.metadata->>'indexing_v3_agent_next_run_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
+      then (d.metadata->>'indexing_v3_agent_next_run_at')::timestamptz
+    else null
+  end as next_run_at,
+  coalesce(nullif(d.metadata->>'indexing_v3_agent_version', ''), 'visual-core-v3') as version,
+  nullif(d.metadata->>'indexing_v3_agent_last_error', '') as last_error,
+  '{}'::jsonb as metadata,
+  coalesce(d.created_at, now()) as created_at,
+  coalesce(d.updated_at, now()) as updated_at
+from public.documents d
+where d.metadata ? 'indexing_v3_agent_status'
+on conflict (document_id) do nothing;
+
+-- -------------------------------------------------------------------------
+-- Step 3: Rewrite claim_indexing_v3_agent_jobs
+--         Uses SKIP LOCKED on the small jobs table.
+--         Also patches documents.metadata for backward compatibility with
+--         the existing edge function (see CRITICAL NOTE above).
+-- -------------------------------------------------------------------------
+
+create or replace function public.claim_indexing_v3_agent_jobs(
+  p_worker_id text,
+  p_claim_limit integer default 1,
+  p_stale_after_minutes integer default 45
+)
+returns table (
+  id uuid,
+  document_id uuid,
+  batch_id uuid,
+  status text,
+  stage text,
+  progress integer,
+  error_message text,
+  attempt_count integer,
+  max_attempts integer,
+  locked_at timestamptz,
+  locked_by text,
+  documents jsonb
+)
+language plpgsql
+set search_path = public, extensions, pg_temp
+as $$
+-- Dual-write compatibility note:
+-- This RPC claims via the jobs table (SKIP LOCKED) and also patches
+-- documents.metadata so the edge function continues to read correct
+-- state. Once the edge function writes completions to this table,
+-- the documents.metadata patch below should be removed.
+begin
+  -- Backward compatibility: old ingestion path still enqueues by writing
+  -- documents.metadata.indexing_v3_agent_status = 'pending'. Ensure those
+  -- documents get a jobs-table row before claiming.
+  insert into public.indexing_v3_agent_jobs (
+    document_id,
+    status,
+    enrichment_status,
+    next_run_at,
+    version,
+    metadata,
+    created_at,
+    updated_at
+  )
+  select
+    d.id,
+    'pending',
+    coalesce(d.metadata->>'enrichment_status', 'pending') as enrichment_status,
+    case
+      when coalesce(d.metadata->>'indexing_v3_agent_next_run_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
+        then (d.metadata->>'indexing_v3_agent_next_run_at')::timestamptz
+      else null
+    end as next_run_at,
+    coalesce(nullif(d.metadata->>'indexing_v3_agent_version', ''), 'visual-core-v3') as version,
+    '{}'::jsonb as metadata,
+    coalesce(d.created_at, now()) as created_at,
+    now() as updated_at
+  from public.documents d
+  where d.status = 'indexed'
+    and d.metadata ? 'indexing_v3_agent_status'
+    and coalesce(d.metadata->>'indexing_v3_agent_status', 'pending')
+          not in ('completed', 'needs_enrichment_artifacts')
+  on conflict (document_id) do nothing;
+
+  return query
+  with eligible_jobs as (
+    select j.id, j.document_id, j.attempt_count, j.max_attempts
+    from public.indexing_v3_agent_jobs j
+    join public.documents d
+      on d.id = j.document_id
+     and d.status = 'indexed'
+    where j.status not in ('completed', 'needs_enrichment_artifacts')
+      and j.enrichment_status in ('pending', 'failed', 'processing')
+      and j.attempt_count < j.max_attempts
+      and coalesce(j.next_run_at, now()) <= now()
+      and (
+        j.status <> 'processing'
+        or j.locked_at is null
+        or j.locked_at < now() - make_interval(mins => p_stale_after_minutes)
+      )
+    order by coalesce(j.next_run_at, j.updated_at), j.id
+    limit greatest(p_claim_limit, 1)
+    for update of j skip locked
+  ),
+  claimed_jobs as (
+    update public.indexing_v3_agent_jobs j
+    set
+      status = 'processing',
+      enrichment_status = 'processing',
+      locked_by = p_worker_id,
+      locked_at = now(),
+      attempt_count = e.attempt_count + 1,
+      last_error = null,
+      next_run_at = null,
+      updated_at = now()
+    from eligible_jobs e
+    where j.id = e.id
+    returning j.*
+  ),
+  -- Patch documents.metadata for backward compatibility with edge function
+  patched_documents as (
+    update public.documents d
+    set
+      metadata = jsonb_strip_nulls(
+        (coalesce(d.metadata, '{}'::jsonb)
+          - 'indexing_v3_agent_next_run_at'
+          - 'indexing_v3_agent_last_error')
+        || jsonb_build_object(
+          'indexing_v3_agent_status', 'processing',
+          'indexing_v3_agent_version', cj.version,
+          'indexing_v3_agent_locked_by', p_worker_id,
+          'indexing_v3_agent_locked_at', cj.locked_at,
+          'indexing_v3_agent_attempt_count', cj.attempt_count,
+          'indexing_v3_agent_max_attempts', cj.max_attempts,
+          'indexing_v3_agent_updated_at', now(),
+          'enrichment_status', 'processing'
+        )
+      ),
+      updated_at = now()
+    from claimed_jobs cj
+    where d.id = cj.document_id
+      and d.status = 'indexed'  -- safety: only touch documents still eligible
+    returning d.*, cj.id as job_id, cj.attempt_count as job_attempt_count,
+              cj.max_attempts as job_max_attempts, cj.locked_at as job_locked_at
+  )
+  select
+    pd.job_id as id,
+    pd.id as document_id,
+    pd.import_batch_id as batch_id,
+    'processing'::text as status,
+    'v3 enrichment claimed'::text as stage,
+    95::integer as progress,
+    null::text as error_message,
+    pd.job_attempt_count,
+    pd.job_max_attempts,
+    pd.job_locked_at as locked_at,
+    p_worker_id as locked_by,
+    to_jsonb(pd.*) - 'job_id' - 'job_attempt_count' - 'job_max_attempts' - 'job_locked_at' as documents
+  from patched_documents pd;
+end;
+$$;
+
+revoke execute on function public.claim_indexing_v3_agent_jobs(text, integer, integer) from public, anon, authenticated;
+grant execute on function public.claim_indexing_v3_agent_jobs(text, integer, integer) to service_role;
+
+-- -------------------------------------------------------------------------
+-- Step 4: Helper RPC for edge function to complete/fail a job
+--         This unblocks the jobs table from being permanently stuck in
+--         'processing'. Pair with edge function update.
+-- -------------------------------------------------------------------------
+
+create or replace function public.update_indexing_v3_agent_job_status(
+  p_document_id uuid,
+  p_status text,         -- 'completed', 'failed', 'needs_enrichment_artifacts', 'pending'
+  p_error text default null,
+  p_next_run_at timestamptz default null
+)
+returns jsonb
+language plpgsql
+set search_path = public, extensions, pg_temp
+as $$
+declare
+  v_job_id uuid;
+begin
+  if p_status not in ('pending', 'completed', 'failed', 'needs_enrichment_artifacts') then
+    raise exception 'invalid status %', p_status;
+  end if;
+
+  update public.indexing_v3_agent_jobs
+  set
+    status = p_status,
+    enrichment_status = case
+      when p_status = 'completed' then 'completed'
+      when p_status = 'failed' then 'failed'
+      when p_status = 'needs_enrichment_artifacts' then 'needs_enrichment_artifacts'
+      else enrichment_status
+    end,
+    last_error = p_error,
+    next_run_at = case
+      when p_status = 'pending' then coalesce(p_next_run_at, now())
+      else null
+    end,
+    locked_by = null,
+    locked_at = null,
+    updated_at = now()
+  where document_id = p_document_id
+  returning id into v_job_id;
+
+  return jsonb_build_object(
+    'ok', v_job_id is not null,
+    'job_id', v_job_id,
+    'document_id', p_document_id,
+    'status', p_status
+  );
+end;
+$$;
+
+revoke execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) from public, anon, authenticated;
+grant execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) to service_role;
+
+-- -------------------------------------------------------------------------
+-- Step 5: Keep documents_indexing_v3_agent_claim_idx in place for now
+--         since the backward-compat documents.metadata patch still writes
+--         to that JSONB path. It can be dropped after the edge function
+--         migration removes JSONB claim reads entirely.
+-- -------------------------------------------------------------------------
+comment on index public.documents_indexing_v3_agent_claim_idx is
+  'Retained for backward compatibility while edge function still writes enrichment_status / indexing_v3_agent_status to documents.metadata. Drop after edge function migration.';
+
+comment on table public.indexing_v3_agent_jobs is
+  'Dedicated worker-state table for the v3 indexing / enrichment agent. Replaces JSONB state in documents.metadata. claim_indexing_v3_agent_jobs uses SKIP LOCKED here; update_indexing_v3_agent_job_status completes/fails a job. See migration 20260702190000 for transition notes.';
diff --git a/supabase/schema.sql b/supabase/schema.sql
index ef11456ff9..b42a1d9d63 100644
--- a/supabase/schema.sql
+++ b/supabase/schema.sql
@@ -137,6 +137,7 @@ create table if not exists public.document_images (
   image_hash text,
   perceptual_hash text,
   labels text[] not null default '{}',
+  index_generation_id uuid,
   metadata jsonb not null default '{}'::jsonb,
   created_at timestamptz not null default now()
 );
@@ -214,6 +215,7 @@ create table if not exists public.document_sections (
   tags text[] not null default '{}',
   extraction_quality text not null default 'unknown'
     check (extraction_quality in ('good', 'partial', 'poor', 'unknown')),
+  index_generation_id uuid,
   metadata jsonb not null default '{}'::jsonb,
   created_at timestamptz not null default now(),
   updated_at timestamptz not null default now(),
@@ -244,6 +246,7 @@ create table if not exists public.document_memory_cards (
   source_chunk_ids uuid[] not null default '{}',
   source_image_ids uuid[] not null default '{}',
   confidence real not null default 0.5 check (confidence >= 0 and confidence <= 1),
+  index_generation_id uuid,
   metadata jsonb not null default '{}'::jsonb,
   embedding extensions.vector(1536) not null,
   search_tsv tsvector generated always as (
@@ -293,6 +296,7 @@ create table if not exists public.document_table_facts (
   threshold_value text,
   action text,
   normalized_terms text[] not null default '{}',
+  index_generation_id uuid,
   metadata jsonb not null default '{}'::jsonb,
   search_tsv tsvector generated always as (
     to_tsvector(
@@ -328,6 +332,7 @@ create table if not exists public.document_embedding_fields (
   content text not null,
   content_hash text,
   embedding extensions.vector(1536) not null,
+  index_generation_id uuid,
   metadata jsonb not null default '{}'::jsonb,
   search_tsv tsvector generated always as (to_tsvector('english', content)) stored,
   created_at timestamptz not null default now()
@@ -520,7 +525,11 @@ create table if not exists public.storage_cleanup_jobs (
   metadata jsonb not null default '{}'::jsonb,
   completed_at timestamptz,
   created_at timestamptz not null default now(),
-  updated_at timestamptz not null default now()
+  updated_at timestamptz not null default now(),
+  constraint storage_cleanup_jobs_document_id_fkey
+    foreign key (document_id)
+    references public.documents(id)
+    on delete set null
 );
 
 create unique index if not exists documents_owner_content_hash_unique_idx
@@ -530,9 +539,9 @@ create index if not exists import_batches_owner_status_idx on public.import_batc
 create index if not exists documents_status_idx on public.documents(status);
 create index if not exists documents_owner_status_idx on public.documents(owner_id, status, created_at desc);
 create index if not exists documents_import_batch_idx on public.documents(import_batch_id);
-create index if not exists documents_owner_hash_idx on public.documents(owner_id, content_hash);
 create index if not exists documents_search_idx on public.documents using gin(search_tsv);
 create index if not exists documents_title_search_idx on public.documents using gin(title_search_tsv);
+create index if not exists documents_owner_id_covering_idx on public.documents(owner_id, id);
 create index if not exists documents_indexed_owner_title_idx
   on public.documents(owner_id, title, file_name)
   where status = 'indexed';
@@ -660,9 +669,6 @@ create index if not exists document_index_quality_owner_score_idx
   on public.document_index_quality(owner_id, quality_score, updated_at desc);
 create index if not exists ingestion_jobs_document_idx on public.ingestion_jobs(document_id);
 create index if not exists ingestion_jobs_batch_idx on public.ingestion_jobs(batch_id, status);
-create index if not exists ingestion_jobs_claim_idx
-  on public.ingestion_jobs(status, next_run_at, created_at)
-  where status in ('pending', 'processing');
 create index if not exists ingestion_jobs_status_next_run_idx
   on public.ingestion_jobs(status, next_run_at, created_at)
   where status in ('pending', 'processing', 'failed');
@@ -981,58 +987,48 @@ returns table (
 language plpgsql
 set search_path = public, extensions, pg_temp
 as $$
+-- Dual-write compatibility note:
+-- This RPC claims via the jobs table (SKIP LOCKED) and also patches
+-- documents.metadata so the edge function continues to read correct
+-- state. Once the edge function writes completions to this table,
+-- the documents.metadata patch below should be removed.
 begin
   return query
-  with eligible as (
-    select
-      d.id,
-      d.import_batch_id,
-      state.attempt_count,
-      state.max_attempts
-    from public.documents d
-    cross join lateral (
-      select
-        coalesce(d.metadata->>'enrichment_status', 'pending') as enrichment_status,
-        coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') as agent_status,
-        case
-          when coalesce(d.metadata->>'indexing_v3_agent_attempt_count', '') ~ '^[0-9]+$'
-            then (d.metadata->>'indexing_v3_agent_attempt_count')::integer
-          else 0
-        end as attempt_count,
-        greatest(
-          case
-            when coalesce(d.metadata->>'indexing_v3_agent_max_attempts', '') ~ '^[0-9]+$'
-              then (d.metadata->>'indexing_v3_agent_max_attempts')::integer
-            else 3
-          end,
-          1
-        ) as max_attempts,
-        case
-          when coalesce(d.metadata->>'indexing_v3_agent_locked_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
-            then (d.metadata->>'indexing_v3_agent_locked_at')::timestamptz
-          else null
-        end as locked_at,
-        case
-          when coalesce(d.metadata->>'indexing_v3_agent_next_run_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
-            then (d.metadata->>'indexing_v3_agent_next_run_at')::timestamptz
-          else null
-        end as next_run_at
-    ) state
-    where d.status = 'indexed'
-      and state.enrichment_status in ('pending', 'failed', 'processing')
-      and state.agent_status not in ('completed', 'needs_enrichment_artifacts')
-      and state.attempt_count < state.max_attempts
-      and coalesce(state.next_run_at, now()) <= now()
+  with eligible_jobs as (
+    select j.id, j.document_id, j.attempt_count, j.max_attempts
+    from public.indexing_v3_agent_jobs j
+    -- must join documents to confirm document.status = 'indexed'
+    -- and to gate on enrichment_status (also stored in the job row)
+    where j.status not in ('completed', 'needs_enrichment_artifacts')
+      and j.enrichment_status in ('pending', 'failed', 'processing')
+      and j.attempt_count < j.max_attempts
+      and coalesce(j.next_run_at, now()) <= now()
       and (
-        state.agent_status <> 'processing'
-        or state.locked_at is null
-        or state.locked_at < now() - make_interval(mins => p_stale_after_minutes)
+        j.status <> 'processing'
+        or j.locked_at is null
+        or j.locked_at < now() - make_interval(mins => p_stale_after_minutes)
       )
-    order by coalesce(state.next_run_at, d.updated_at), d.id
+    order by coalesce(j.next_run_at, j.updated_at), j.id
     limit greatest(p_claim_limit, 1)
-    for update of d skip locked
+    for update of j skip locked
   ),
-  claimed as (
+  claimed_jobs as (
+    update public.indexing_v3_agent_jobs j
+    set
+      status = 'processing',
+      enrichment_status = 'processing',
+      locked_by = p_worker_id,
+      locked_at = now(),
+      attempt_count = e.attempt_count + 1,
+      last_error = null,
+      next_run_at = null,
+      updated_at = now()
+    from eligible_jobs e
+    where j.id = e.id
+    returning j.*
+  ),
+  -- Patch documents.metadata for backward compatibility with edge function
+  patched_documents as (
     update public.documents d
     set
       metadata = jsonb_strip_nulls(
@@ -1041,52 +1037,36 @@ begin
           - 'indexing_v3_agent_last_error')
         || jsonb_build_object(
           'indexing_v3_agent_status', 'processing',
-          'indexing_v3_agent_version', 'visual-core-v3',
+          'indexing_v3_agent_version', cj.version,
           'indexing_v3_agent_locked_by', p_worker_id,
-          'indexing_v3_agent_locked_at', now(),
-          'indexing_v3_agent_attempt_count', e.attempt_count + 1,
-          'indexing_v3_agent_max_attempts', e.max_attempts,
+          'indexing_v3_agent_locked_at', cj.locked_at,
+          'indexing_v3_agent_attempt_count', cj.attempt_count,
+          'indexing_v3_agent_max_attempts', cj.max_attempts,
           'indexing_v3_agent_updated_at', now(),
           'enrichment_status', 'processing'
         )
       ),
       updated_at = now()
-    from eligible e
-    where d.id = e.id
-    returning d.*, e.attempt_count + 1 as claimed_attempt_count, e.max_attempts as claimed_max_attempts
+    from claimed_jobs cj
+    where d.id = cj.document_id
+      and d.status = 'indexed'  -- safety: only touch documents still eligible
+    returning d.*, cj.id as job_id, cj.attempt_count as job_attempt_count,
+              cj.max_attempts as job_max_attempts, cj.locked_at as job_locked_at
   )
   select
-    c.id,
-    c.id as document_id,
-    c.import_batch_id as batch_id,
+    pd.job_id as id,
+    pd.id as document_id,
+    pd.import_batch_id as batch_id,
     'processing'::text as status,
     'v3 enrichment claimed'::text as stage,
     95::integer as progress,
     null::text as error_message,
-    c.claimed_attempt_count,
-    c.claimed_max_attempts,
-    (c.metadata->>'indexing_v3_agent_locked_at')::timestamptz as locked_at,
-    c.metadata->>'indexing_v3_agent_locked_by' as locked_by,
-    to_jsonb(c.*) - 'claimed_attempt_count' - 'claimed_max_attempts' as documents
-  from claimed c;
-end;
-$$;
-
-create or replace function public.reset_document_index(p_document_id uuid)
-returns void
-language plpgsql
-set search_path = public, extensions, pg_temp
-as $$
-begin
-  perform set_config('statement_timeout', '180000', true);
-  delete from public.document_memory_cards where document_id = p_document_id;
-  delete from public.document_sections where document_id = p_document_id;
-  delete from public.document_table_facts where document_id = p_document_id;
-  delete from public.document_embedding_fields where document_id = p_document_id;
-  delete from public.document_index_quality where document_id = p_document_id;
-  delete from public.document_chunks where document_id = p_document_id;
-  delete from public.document_images where document_id = p_document_id;
-  delete from public.document_pages where document_id = p_document_id;
+    pd.job_attempt_count,
+    pd.job_max_attempts,
+    pd.job_locked_at as locked_at,
+    p_worker_id as locked_by,
+    to_jsonb(pd.*) - 'job_id' - 'job_attempt_count' - 'job_max_attempts' - 'job_locked_at' as documents
+  from patched_documents pd;
 end;
 $$;
 
@@ -1171,13 +1151,7 @@ begin
       updated_at = excluded.updated_at;
   end if;
 
-  -- M13 (audit 2026-07-01): superseded-generation rows always go; legacy
-  -- NULL-generation rows go only when this generation wrote replacement rows
-  -- into the same table (see 20260702000000_commit_generation_preserve_legacy_artifacts).
-  -- Guarantee scope: fully protects document_images/document_memory_cards/
-  -- document_sections; chunk-anchored artifacts (table facts, embedding
-  -- fields, index units) cascade with their legacy chunks via
-  -- source_chunk_id ON DELETE CASCADE when chunks are replaced.
+  -- Preserve legacy NULL-generation rows unless this generation wrote replacements.
   delete from public.document_chunks
   where document_id = p_document_id
     and (
@@ -1193,18 +1167,27 @@ begin
       )
     );
 
+  -- artifact tables: use typed column where set; fall back to metadata when typed is NULL
+  -- because the writer still populates metadata.index_generation_id rather than the typed
+  -- column.  Without the metadata fallback, stale null-typed rows from a prior run would
+  -- never be cleaned up (the typed-column EXISTS guard would always be false), allowing
+  -- artifact rows to accumulate across re-indexes.
   delete from public.document_images
   where document_id = p_document_id
     and (
-      (nullif(metadata->>'index_generation_id', '') is not null
-        and metadata->>'index_generation_id' <> p_index_generation_id::text)
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
       or (
-        nullif(metadata->>'index_generation_id', '') is null
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
         and exists (
           select 1
           from public.document_images replacement
           where replacement.document_id = p_document_id
-            and replacement.metadata->>'index_generation_id' = p_index_generation_id::text
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
         )
       )
     );
@@ -1212,15 +1195,19 @@ begin
   delete from public.document_table_facts
   where document_id = p_document_id
     and (
-      (nullif(metadata->>'index_generation_id', '') is not null
-        and metadata->>'index_generation_id' <> p_index_generation_id::text)
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
       or (
-        nullif(metadata->>'index_generation_id', '') is null
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
         and exists (
           select 1
           from public.document_table_facts replacement
           where replacement.document_id = p_document_id
-            and replacement.metadata->>'index_generation_id' = p_index_generation_id::text
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
         )
       )
     );
@@ -1228,15 +1215,19 @@ begin
   delete from public.document_embedding_fields
   where document_id = p_document_id
     and (
-      (nullif(metadata->>'index_generation_id', '') is not null
-        and metadata->>'index_generation_id' <> p_index_generation_id::text)
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
       or (
-        nullif(metadata->>'index_generation_id', '') is null
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
         and exists (
           select 1
           from public.document_embedding_fields replacement
           where replacement.document_id = p_document_id
-            and replacement.metadata->>'index_generation_id' = p_index_generation_id::text
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
         )
       )
     );
@@ -1244,15 +1235,19 @@ begin
   delete from public.document_index_units
   where document_id = p_document_id
     and (
-      (nullif(metadata->>'index_generation_id', '') is not null
-        and metadata->>'index_generation_id' <> p_index_generation_id::text)
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
       or (
-        nullif(metadata->>'index_generation_id', '') is null
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
         and exists (
           select 1
           from public.document_index_units replacement
           where replacement.document_id = p_document_id
-            and replacement.metadata->>'index_generation_id' = p_index_generation_id::text
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
         )
       )
     );
@@ -1260,15 +1255,19 @@ begin
   delete from public.document_memory_cards
   where document_id = p_document_id
     and (
-      (nullif(metadata->>'index_generation_id', '') is not null
-        and metadata->>'index_generation_id' <> p_index_generation_id::text)
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
       or (
-        nullif(metadata->>'index_generation_id', '') is null
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
         and exists (
           select 1
           from public.document_memory_cards replacement
           where replacement.document_id = p_document_id
-            and replacement.metadata->>'index_generation_id' = p_index_generation_id::text
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
         )
       )
     );
@@ -1276,15 +1275,19 @@ begin
   delete from public.document_sections
   where document_id = p_document_id
     and (
-      (nullif(metadata->>'index_generation_id', '') is not null
-        and metadata->>'index_generation_id' <> p_index_generation_id::text)
+      (index_generation_id is not null and index_generation_id <> p_index_generation_id)
       or (
-        nullif(metadata->>'index_generation_id', '') is null
+        index_generation_id is null
+        and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id
         and exists (
           select 1
           from public.document_sections replacement
           where replacement.document_id = p_document_id
-            and replacement.metadata->>'index_generation_id' = p_index_generation_id::text
+            and (
+              replacement.index_generation_id = p_index_generation_id
+              or (replacement.index_generation_id is null
+                  and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id)
+            )
         )
       )
     );
@@ -1318,9 +1321,12 @@ declare
 begin
   perform set_config('statement_timeout', '180000', true);
 
+  -- Collect distinct document_ids that have stale (non-committed) artifact rows.
+  -- document_chunks uses its typed column; artifact tables use their new typed columns.
   with candidate_documents as (
     select distinct document_id
     from (
+      -- document_chunks (typed index_generation_id)
       select c.document_id
       from public.document_chunks c
       join public.documents d on d.id = c.document_id
@@ -1333,72 +1339,78 @@ begin
             and j.status in ('pending', 'processing')
         )
       union all
+      -- document_images (typed index_generation_id)
       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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-        and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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
+      -- document_table_facts (typed index_generation_id)
       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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-        and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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
+      -- document_embedding_fields (typed index_generation_id)
       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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-        and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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
+      -- document_index_units (typed index_generation_id)
       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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-        and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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
+      -- document_memory_cards (typed index_generation_id)
       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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-        and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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
+      -- document_sections (typed index_generation_id)
       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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-        and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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
@@ -1411,6 +1423,7 @@ begin
   into target_document_ids
   from candidate_documents;
 
+  -- Count stale rows (typed column comparisons)
   select count(*) into chunk_count
   from public.document_chunks c
   join public.documents d on d.id = c.document_id
@@ -1422,43 +1435,43 @@ begin
   from public.document_images a
   join public.documents d on d.id = a.document_id
   where a.document_id = any(target_document_ids)
-    and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-    and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
   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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-    and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
   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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-    and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
   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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-    and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
   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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-    and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
   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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-    and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
   if not coalesce(p_dry_run, true) then
     delete from public.document_chunks c
@@ -1472,43 +1485,43 @@ begin
     using public.documents d
     where d.id = a.document_id
       and a.document_id = any(target_document_ids)
-      and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-      and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
     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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-      and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
     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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-      and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
     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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-      and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
     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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-      and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
 
     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 nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null
-      and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_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', '');
   end if;
 
   return jsonb_build_object(
@@ -2579,6 +2592,43 @@ as $$
       (ts_rank_cd(d.title_search_tsv, query.tsq) * 3.0)
     ) desc
     limit least(greatest(match_count * 2, 24), 96)
+  ),
+  -- Batch-fetch label metadata for all distinct document_ids in the result set.
+  -- One query replaces N per-row calls to document_label_metadata().
+  doc_labels as (
+    select
+      l.document_id,
+      coalesce(
+        jsonb_agg(
+          jsonb_build_object(
+            'id',          l.id,
+            'document_id', l.document_id,
+            'owner_id',    l.owner_id,
+            'label',       l.label,
+            'label_type',  l.label_type,
+            'source',      l.source,
+            'confidence',  l.confidence,
+            'metadata',    l.metadata,
+            'created_at',  l.created_at,
+            'updated_at',  l.updated_at
+          )
+          order by l.confidence desc, l.label
+        ),
+        '[]'::jsonb
+      ) as labels
+    from public.document_labels l
+    where l.document_id in (select distinct ranked.document_id from ranked)
+    group by l.document_id
+  ),
+  -- Batch-fetch summary text for all distinct document_ids in the result set.
+  -- One query replaces N per-row calls to document_summary_text().
+  doc_summaries as (
+    select distinct on (s.document_id)
+      s.document_id,
+      s.summary
+    from public.document_summaries s
+    where s.document_id in (select distinct ranked.document_id from ranked)
+    order by s.document_id
   )
   select
     ranked.id,
@@ -2592,21 +2642,23 @@ as $$
     ranked.retrieval_synopsis,
     ranked.image_ids,
     ranked.source_metadata,
-    coalesce(public.document_label_metadata(ranked.document_id), '[]'::jsonb) as document_labels,
-    public.document_summary_text(ranked.document_id) as document_summary,
+    coalesce(doc_labels.labels,   '[]'::jsonb) as document_labels,
+    doc_summaries.summary                       as document_summary,
     -- Text-only fallback has NO vector cosine similarity. Do not fabricate one:
     -- a synthetic value here was read downstream as a real semantic score and
     -- could label a pure keyword hit as "strong"/"moderate" evidence (>=0.64).
     -- Leave similarity at 0; the lexical signal lives in lexical_score.
-    0::double precision as similarity,
+    0::double precision                                                              as similarity,
     ranked.text_rank,
     -- Cap hybrid_score well below the 0.64 "moderate" threshold so a lexical-only
     -- row can order amongst its peers but can never masquerade as a moderate/strong
     -- cosine match when merged with vector results.
-    least(0.5, 0.18 + (least(ranked.text_rank, 1) * 0.3))::double precision as hybrid_score,
-    least(0.99, 0.4 + (least(ranked.text_rank, 1) * 0.59))::double precision as lexical_score,
-    public.chunk_image_metadata(ranked.image_ids) as images
+    least(0.5,  0.18 + (least(ranked.text_rank, 1) * 0.3))::double precision       as hybrid_score,
+    least(0.99, 0.4  + (least(ranked.text_rank, 1) * 0.59))::double precision      as lexical_score,
+    public.chunk_image_metadata(ranked.image_ids)                                   as images
   from ranked
+  left join doc_labels    on doc_labels.document_id    = ranked.document_id
+  left join doc_summaries on doc_summaries.document_id = ranked.document_id
   order by lexical_score desc, text_rank desc
   limit match_count;
 $$;
@@ -3391,6 +3443,9 @@ $$;
 revoke execute on function public.complete_strict_enrichment_job(uuid, uuid, text, text, text) from public, anon, authenticated;
 grant execute on function public.complete_strict_enrichment_job(uuid, uuid, text, text, text) to service_role;
 
+alter database postgres
+  set app.indexing_v3_agent_base_url = 'https://sjrfecxgysukkwxsowpy.supabase.co';
+
 create or replace function public.invoke_indexing_v3_agent(p_limit integer default 1)
 returns bigint
 language plpgsql
@@ -3399,7 +3454,8 @@ set search_path = public, extensions, vault, pg_temp
 as $$
 declare
   v_request_id bigint;
-  v_secret text;
+  v_secret     text;
+  v_base_url   text;
 begin
   select decrypted_secret
     into v_secret
@@ -3411,8 +3467,16 @@ begin
     raise exception 'indexing_v3_agent_secret is missing from Supabase Vault';
   end if;
 
+  -- Prefer the GUC; fall back to the hardcoded production URL so that
+  -- existing deployments that have not yet set the GUC continue to work.
+  v_base_url := coalesce(
+    nullif(current_setting('app.indexing_v3_agent_base_url', true), ''),
+    'https://sjrfecxgysukkwxsowpy.supabase.co'
+  );
+
   select net.http_post(
-    url := 'https://sjrfecxgysukkwxsowpy.supabase.co/functions/v1/indexing-v3-agent?limit=' || greatest(1, least(coalesce(p_limit, 1), 10))::text,
+    url := v_base_url || '/functions/v1/indexing-v3-agent?limit='
+           || greatest(1, least(coalesce(p_limit, 1), 10))::text,
     headers := jsonb_build_object(
       'Content-Type', 'application/json',
       'x-indexing-agent-secret', v_secret
@@ -3653,6 +3717,7 @@ create table if not exists public.document_index_units (
   source_span jsonb,
   quality_score real not null default 0.7 check (quality_score >= 0 and quality_score <= 1),
   extraction_mode text not null default 'deterministic' check (extraction_mode in ('deterministic', 'model_heavy', 'hybrid')),
+  index_generation_id uuid,
   embedding extensions.vector(1536) not null,
   metadata jsonb not null default '{}'::jsonb,
   search_tsv tsvector generated always as (to_tsvector('english', coalesce(unit_type, '') || ' ' || coalesce(title, '') || ' ' || coalesce(content, ''))) stored,
@@ -3806,7 +3871,129 @@ grant execute on function public.is_committed_document_generation(uuid, jsonb) t
 revoke execute on function public.is_committed_artifact_generation(jsonb, jsonb) from public, anon, authenticated;
 grant execute on function public.is_committed_artifact_generation(jsonb, jsonb) to service_role;
 
+-- Typed overload: NULL keeps legacy artifacts visible; otherwise compare to committed id
+create or replace function public.is_committed_artifact_generation(p_artifact_gen_id uuid, p_document_metadata jsonb)
+returns boolean
+language sql
+stable
+set search_path = public, extensions, pg_temp
+as $$
+  select p_artifact_gen_id is null
+    or p_artifact_gen_id::text = nullif(coalesce(p_document_metadata, '{}'::jsonb)->>'index_generation_id', '');
+$$;
+
+revoke execute on function public.is_committed_artifact_generation(uuid, jsonb) from public, anon, authenticated;
+grant execute on function public.is_committed_artifact_generation(uuid, jsonb) to service_role;
+
 create policy "document index units owner read" on public.document_index_units
   for select to authenticated using (
     exists (select 1 from public.documents d where d.id = document_id and d.owner_id = (select auth.uid()))
   );
+
+-- -------------------------------------------------------------------------
+-- indexing_v3_agent_jobs: dedicated worker-state table (Finding #1)
+-- Replaces JSONB claim state in documents.metadata with typed rows that
+-- support SKIP LOCKED on a small, hot table instead of a full-table scan.
+-- -------------------------------------------------------------------------
+
+create table if not exists public.indexing_v3_agent_jobs (
+  id uuid primary key default gen_random_uuid(),
+  document_id uuid not null references public.documents(id) on delete cascade,
+  -- v3 agent processing status (mirrors metadata->>'indexing_v3_agent_status')
+  status text not null default 'pending'
+    check (status in ('pending', 'processing', 'completed', 'failed', 'needs_enrichment_artifacts')),
+  -- enrichment pipeline status (mirrors metadata->>'enrichment_status')
+  enrichment_status text not null default 'pending'
+    check (enrichment_status in ('pending', 'processing', 'completed', 'failed', 'needs_enrichment_artifacts')),
+  attempt_count integer not null default 0,
+  max_attempts integer not null default 3,
+  locked_by text,
+  locked_at timestamptz,
+  next_run_at timestamptz,
+  version text not null default 'visual-core-v3',
+  last_error text,
+  metadata jsonb not null default '{}'::jsonb,
+  created_at timestamptz not null default now(),
+  updated_at timestamptz not null default now()
+);
+
+-- One row per document; re-running resets the row in-place
+create unique index if not exists indexing_v3_agent_jobs_document_id_idx
+  on public.indexing_v3_agent_jobs(document_id);
+
+-- Hot path for claim: eligible candidates ordered by next_run_at
+create index if not exists indexing_v3_agent_jobs_claim_idx
+  on public.indexing_v3_agent_jobs(status, enrichment_status, next_run_at, id)
+  where status not in ('completed', 'needs_enrichment_artifacts');
+
+-- Operational: find stale processing jobs
+create index if not exists indexing_v3_agent_jobs_locked_at_idx
+  on public.indexing_v3_agent_jobs(locked_at)
+  where status = 'processing';
+
+-- RLS + grants (service_role only, same as ingestion_jobs)
+alter table public.indexing_v3_agent_jobs enable row level security;
+
+create policy "indexing v3 agent jobs service role all"
+  on public.indexing_v3_agent_jobs
+  for all to service_role
+  using (true)
+  with check (true);
+
+grant select, insert, update, delete
+  on table public.indexing_v3_agent_jobs to service_role;
+
+create or replace function public.update_indexing_v3_agent_job_status(
+  p_document_id uuid,
+  p_status text,         -- 'completed', 'failed', 'needs_enrichment_artifacts', 'pending'
+  p_error text default null,
+  p_next_run_at timestamptz default null
+)
+returns jsonb
+language plpgsql
+set search_path = public, extensions, pg_temp
+as $$
+declare
+  v_job_id uuid;
+begin
+  if p_status not in ('pending', 'completed', 'failed', 'needs_enrichment_artifacts') then
+    raise exception 'invalid status %', p_status;
+  end if;
+
+  update public.indexing_v3_agent_jobs
+  set
+    status = p_status,
+    enrichment_status = case
+      when p_status = 'completed' then 'completed'
+      when p_status = 'failed' then 'failed'
+      when p_status = 'needs_enrichment_artifacts' then 'needs_enrichment_artifacts'
+      else enrichment_status
+    end,
+    last_error = p_error,
+    next_run_at = case
+      when p_status = 'pending' then coalesce(p_next_run_at, now())
+      else null
+    end,
+    locked_by = null,
+    locked_at = null,
+    updated_at = now()
+  where document_id = p_document_id
+  returning id into v_job_id;
+
+  return jsonb_build_object(
+    'ok', v_job_id is not null,
+    'job_id', v_job_id,
+    'document_id', p_document_id,
+    'status', p_status
+  );
+end;
+$$;
+
+revoke execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) from public, anon, authenticated;
+grant execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) to service_role;
+
+comment on index public.documents_indexing_v3_agent_claim_idx is
+  'Retained for backward compatibility while edge function still writes enrichment_status / indexing_v3_agent_status to documents.metadata. Drop after edge function migration.';
+
+comment on table public.indexing_v3_agent_jobs is
+  'Dedicated worker-state table for the v3 indexing / enrichment agent. Replaces JSONB state in documents.metadata. claim_indexing_v3_agent_jobs uses SKIP LOCKED here; update_indexing_v3_agent_job_status completes/fails a job. See migration 20260702190000 for transition notes.';
diff --git a/tests/supabase-schema.test.ts b/tests/supabase-schema.test.ts
index e1f556deea..ee94693723 100644
--- a/tests/supabase-schema.test.ts
+++ b/tests/supabase-schema.test.ts
@@ -60,6 +60,14 @@ const preserveLegacyArtifactCommitMigration = readFileSync(
   new URL("../supabase/migrations/20260702000000_commit_generation_preserve_legacy_artifacts.sql", import.meta.url),
   "utf8",
 ).replace(/\s+/g, " ");
+const promoteIndexGenerationIdMigration = readFileSync(
+  new URL("../supabase/migrations/20260702180000_promote_index_generation_id_columns.sql", import.meta.url),
+  "utf8",
+).replace(/\s+/g, " ");
+const indexingV3AgentJobsMigration = readFileSync(
+  new URL("../supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql", import.meta.url),
+  "utf8",
+).replace(/\s+/g, " ");
 
 function extractTextChunkFunction(sql: string) {
   const start = sql.indexOf("function public.match_document_chunks_text");
@@ -177,7 +185,7 @@ describe("Supabase schema Data API grants", () => {
   });
 
   it("preserves NULL-generation artifacts until replacements exist", () => {
-    for (const sql of [schema, preserveLegacyArtifactCommitMigration]) {
+    for (const sql of [preserveLegacyArtifactCommitMigration]) {
       expect(sql).toContain(
         "index_generation_id is null and exists ( select 1 from public.document_chunks replacement",
       );
@@ -192,6 +200,24 @@ describe("Supabase schema Data API grants", () => {
       expect(sql).toContain("from public.document_memory_cards replacement");
       expect(sql).toContain("from public.document_sections replacement");
     }
+
+    for (const sql of [schema, promoteIndexGenerationIdMigration]) {
+      expect(sql).toContain(
+        "index_generation_id is null and exists ( select 1 from public.document_chunks replacement",
+      );
+      expect(sql).toContain("(metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id");
+      expect(sql).toContain("replacement.index_generation_id = p_index_generation_id");
+      expect(sql).toContain(
+        "replacement.index_generation_id is null and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id",
+      );
+      expect(sql).toContain("from public.document_chunks replacement");
+      expect(sql).toContain("from public.document_images replacement");
+      expect(sql).toContain("from public.document_table_facts replacement");
+      expect(sql).toContain("from public.document_embedding_fields replacement");
+      expect(sql).toContain("from public.document_index_units replacement");
+      expect(sql).toContain("from public.document_memory_cards replacement");
+      expect(sql).toContain("from public.document_sections replacement");
+    }
   });
 
   it("can identify and clean abandoned staged reindex generations", () => {
@@ -224,18 +250,31 @@ describe("Supabase schema Data API grants", () => {
     );
     expect(schema).toContain("drop index if exists public.ingestion_job_stages_doc_idx");
     expect(schema).toContain("create index if not exists ingestion_job_stages_document_started_idx");
-    expect(schema).toContain("create or replace function public.claim_indexing_v3_agent_jobs");
-    expect(schema).toContain("where d.status = 'indexed'");
-    expect(schema).toContain("state.enrichment_status in ('pending', 'failed', 'processing')");
-    expect(schema).toContain("'indexing_v3_agent_locked_by', p_worker_id");
-    expect(schema).toContain("'indexing_v3_agent_attempt_count', e.attempt_count + 1");
+    for (const sql of [schema, indexingV3AgentJobsMigration]) {
+      expect(sql).toContain("create table if not exists public.indexing_v3_agent_jobs");
+      expect(sql).toContain("document_id uuid not null references public.documents(id) on delete cascade");
+      expect(sql).toContain("create index if not exists indexing_v3_agent_jobs_claim_idx");
+      expect(sql).toContain("create or replace function public.claim_indexing_v3_agent_jobs");
+      expect(sql).toContain("from public.indexing_v3_agent_jobs j");
+      expect(sql).toContain("j.enrichment_status in ('pending', 'failed', 'processing')");
+      expect(sql).toContain("update public.indexing_v3_agent_jobs j");
+      expect(sql).toContain("'indexing_v3_agent_locked_by', p_worker_id");
+      expect(sql).toContain("'indexing_v3_agent_attempt_count', cj.attempt_count");
+      expect(sql).toContain("create or replace function public.update_indexing_v3_agent_job_status");
+      expect(sql).toContain(
+        "grant execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) to service_role",
+      );
+    }
     expect(schema).toContain(
       "grant execute on function public.claim_indexing_v3_agent_jobs(text, integer, integer) to service_role",
     );
     expect(schema).toContain("alter table public.ingestion_job_stages enable row level security");
     expect(schema).toContain('create policy "ingestion job stages service role all" on public.ingestion_job_stages');
+    expect(schema).toContain("alter table public.indexing_v3_agent_jobs enable row level security");
+    expect(schema).toContain('create policy "indexing v3 agent jobs service role all" on public.indexing_v3_agent_jobs');
     const authenticatedSelectGrant = schema.match(/grant select on table ([^;]+) to authenticated;/)?.[1] ?? "";
     expect(authenticatedSelectGrant).not.toContain("public.ingestion_job_stages");
+    expect(authenticatedSelectGrant).not.toContain("public.indexing_v3_agent_jobs");
   });
 
   it("keeps the cron indexing-v3 invoker in the schema snapshot with service-role-only execute grants", () => {
@@ -245,7 +284,11 @@ describe("Supabase schema Data API grants", () => {
     expect(schema).toContain("set search_path = public, extensions, vault, pg_temp");
     expect(schema).toContain("from vault.decrypted_secrets");
     expect(schema).toContain("where name = 'indexing_v3_agent_secret'");
+    expect(schema).toContain("set app.indexing_v3_agent_base_url = 'https://sjrfecxgysukkwxsowpy.supabase.co';");
+    expect(schema).toContain("nullif(current_setting('app.indexing_v3_agent_base_url', true), '')");
     expect(schema).toContain("select net.http_post(");
+    expect(schema).toContain("v_base_url || '/functions/v1/indexing-v3-agent?limit='");
+    expect(schema).toContain("'https://sjrfecxgysukkwxsowpy.supabase.co'");
     expect(schema).toContain("/functions/v1/indexing-v3-agent?limit=");
     expect(schema).toContain(
       "revoke execute on function public.invoke_indexing_v3_agent(integer) from public, anon, authenticated",