Fix migration P1s: live catalog status, run-owned stage updates, renamed-code reindex

docs/POSTGRES_SCHEMA.md

@@ -522,13 +522,13 @@ Computed from `documents`, `catalogs`, `instances`, `workflows` and `document_cu
 | code | TEXT | `documents.code`. |
 | name | TEXT | `documents.name`. |
 | aliases | TEXT[] | `documents.aliases`. |
-| status | TEXT | `pending_reindex` if the document is live and `metadata.reindex_required` is true. Else `live` if live. Else `pending_prod` if the latest workflow's stage is `approval_for_prod` and the document is not disabled. Else `disabled`. |
+| status | TEXT | `pending_reindex` if the document is live and `metadata.reindex_required` is true. Else `live` if live. Else `pending_prod` if the current (latest) workflow's stage is `approval_for_prod` and the document is not disabled. Else `disabled`. |
 | workflow_id | TEXT | `temporal_workflow_id` of the document's earliest `pipeline` workflow. |
 | chunk_count | INTEGER | Count of the document's `chunks`. |
 | network_visible | BOOLEAN | `metadata.network_visible`, true when absent. |
 | promoted_at | TIMESTAMPTZ | `metadata.index_status.prod.last_indexed_at`, only for live documents. |
 
-"Live" means the latest workflow's `current_stage = 'completed'`, `is_disabled` is false and `metadata.catalog_visible` is true. There are no bootstrap rows: a scheme appears only once a document exists.
+"Live" means all of: `is_disabled` is false; `metadata.catalog_visible` is true; some workflow of the document has `current_stage = 'completed'`; and either the PROD index holds it (`metadata.index_status.prod.status = 'indexed'`) or a completed run was started with PROD skipped (`workflows.config.skip_prod`, set from `DISABLE_PROD_SETTING`). A DEV-only document is therefore not live, and starting a new run does not hide a document that is already live (migration `0003`). There are no bootstrap rows: a scheme appears only once a document exists.
 
 ---
 

migrations/0003_catalog_live_requires_prod_index.sql

@@ -0,0 +1,47 @@
+-- catalog_status: a document is "live" only when it is actually being served.
+--
+-- Before: live = the document's latest workflow was at stage 'completed'. That
+-- advertised DEV-only documents as live, and hid a live document as soon as a
+-- new run (reingest, retry) started.
+--
+-- Now: live = some workflow of the document reached 'completed' AND either the
+-- PROD index holds it (metadata.index_status.prod.status = 'indexed') or it
+-- finished a run that skipped PROD on purpose (workflows.config.skip_prod, set
+-- from DISABLE_PROD_SETTING when the run started).
+CREATE OR REPLACE VIEW catalog_status AS
+SELECT
+  d.id AS document_id,
+  c.id AS catalog_id,
+  c.instance_id,
+  g.name AS instance_name,
+  c.type,
+  d.code,
+  d.name,
+  d.aliases,
+  CASE
+    WHEN x.is_live AND COALESCE((d.metadata->>'reindex_required')::boolean, FALSE) THEN 'pending_reindex'
+    WHEN x.is_live THEN 'live'
+    WHEN cw.current_stage = 'approval_for_prod' AND NOT d.is_disabled THEN 'pending_prod'
+    ELSE 'disabled'
+  END AS status,
+  (SELECT w.temporal_workflow_id FROM workflows w
+    WHERE w.document_id = d.id AND w.workflow_type = 'pipeline'
+    ORDER BY w.started_at ASC, w.id ASC LIMIT 1) AS workflow_id,
+  (SELECT count(*) FROM chunks ch WHERE ch.document_id = d.id)::INTEGER AS chunk_count,
+  COALESCE((d.metadata->>'network_visible')::boolean, TRUE) AS network_visible,
+  CASE WHEN x.is_live
+       THEN (d.metadata->'index_status'->'prod'->>'last_indexed_at')::timestamptz END AS promoted_at
+FROM documents d
+JOIN catalogs c ON c.id = d.catalog_id
+JOIN instances g ON g.id = c.instance_id
+LEFT JOIN document_current_workflow cw ON cw.document_id = d.id
+CROSS JOIN LATERAL (
+  SELECT (NOT d.is_disabled
+          AND COALESCE((d.metadata->>'catalog_visible')::boolean, TRUE)
+          AND EXISTS (SELECT 1 FROM workflows w
+                      WHERE w.document_id = d.id AND w.current_stage = 'completed')
+          AND (d.metadata->'index_status'->'prod'->>'status' = 'indexed'
+               OR EXISTS (SELECT 1 FROM workflows w
+                          WHERE w.document_id = d.id AND w.current_stage = 'completed'
+                            AND w.config->>'skip_prod' = 'true'))) AS is_live
+) x;

pipeline/activities.py

@@ -1599,19 +1599,25 @@ async def update_document_state(
     page_count: int = 0,
     chunk_count: int = 0,
     error_message: str = None,
+    run_workflow_id: str = None,
 ) -> dict:
-    """Update document state in Postgres."""
+    """Update document state in Postgres.
+
+    ``run_workflow_id`` is the reporting run's own Temporal workflow id; the stage
+    (and its job row) is written for that run rather than the document's newest.
+    """
     from . import db
 
     activity.logger.info(f"Updating state for {workflow_id}: stage={stage}")
-    db.update_document_stage(
+    run_job_id = db.update_document_stage(
         workflow_id=workflow_id,
         stage=stage,
         page_count=page_count,
         chunk_count=chunk_count,
         error_message=error_message,
+        run_workflow_id=run_workflow_id,
     )
-    latest_job = db.get_latest_document_job(workflow_id)
+    latest_job = db.get_document_job(run_job_id) if run_job_id else db.get_latest_document_job(workflow_id)
     if latest_job:
         job_updates = {"current_stage": stage}
         if stage in {

pipeline/api/documents.py

@@ -170,6 +170,7 @@ async def start_document_workflow(
             "min_tokens": min_tokens,
             "index_name": index_name,
             "stop_after_ocr": stop_after_ocr,
+            "skip_prod": prod_stage_disabled(),
             "uploaded_by": actor,
         },
         actor=actor,

pipeline/api/upload.py

@@ -188,6 +188,7 @@ async def upload_and_process(
             "min_tokens": min_tokens,
             "index_name": index_name,
             "stop_after_ocr": stop_after_ocr,
+            "skip_prod": prod_stage_disabled(),
             "uploaded_by": actor,
         },
         actor=actor,

pipeline/db.py

@@ -1247,9 +1247,15 @@ def update_document_stage(
     stage: str,
     page_count: Optional[int] = None,
     chunk_count: Optional[int] = None,
-    error_message: Optional[str] = None
-):
-    """Update the stage (of the document's current workflow).
+    error_message: Optional[str] = None,
+    run_workflow_id: Optional[str] = None,
+) -> Optional[str]:
+    """Update the stage of one workflow run; returns that run's workflow UUID.
+
+    ``run_workflow_id`` is the Temporal workflow id of the run reporting the stage.
+    When it matches a run of this document, only that run is written, so a late
+    update from an older run cannot move a newer run's stage. Without it (or when
+    no run matches) the document's current workflow is updated, as before.
 
     ``page_count`` and ``chunk_count`` are accepted for call-site compatibility;
     both are counted from ``pages`` and ``chunks``, so only the audit entry keeps them.
@@ -1259,14 +1265,26 @@ def update_document_stage(
     with get_connection() as conn:
         doc_uuid = _resolve_document_uuid(conn, workflow_id)
         if doc_uuid is None:
-            return
-        cw = conn.execute(
-            "SELECT id, current_stage FROM document_current_workflow WHERE document_id = %s::uuid",
-            (doc_uuid,),
-        ).fetchone()
+            return None
+        cw = None
+        if run_workflow_id:
+            cw = conn.execute(
+                """
+                SELECT id, current_stage FROM workflows
+                WHERE document_id = %s::uuid AND temporal_workflow_id = %s
+                ORDER BY started_at DESC, id DESC LIMIT 1
+                """,
+                (doc_uuid, run_workflow_id),
+            ).fetchone()
         if cw is None:
-            return
+            cw = conn.execute(
+                "SELECT id, current_stage FROM document_current_workflow WHERE document_id = %s::uuid",
+                (doc_uuid,),
+            ).fetchone()
+        if cw is None:
+            return None
         old_stage = cw["current_stage"]
+        run_id = str(cw["id"])
 
         assignments = ["current_stage = %s"]
         params: list = [stage]
@@ -1303,6 +1321,7 @@ def update_document_stage(
             new_value=stage,
             metadata={"page_count": page_count, "chunk_count": chunk_count}
         )
+    return run_id
 
 
 def _root_workflow_id(conn, doc_uuid: str) -> Optional[str]:

pipeline/workflows.py

@@ -103,17 +103,22 @@ async def _mirror_state(
     Local activities avoid starving tiny state updates behind long-running OCR,
     translation, or chunking activities on the shared task queue.
     """
+    args = [workflow_id, stage, page_count, chunk_count, error_message]
+    if workflow.patched("run-owned-stage-v1"):
+        # Tell the activity which run is reporting, so it writes that run's row.
+        args.append(workflow.info().workflow_id)
+
     if workflow.patched("local-state-update-v1"):
         return await workflow.execute_local_activity(
             update_document_state,
-            args=[workflow_id, stage, page_count, chunk_count, error_message],
+            args=args,
             start_to_close_timeout=timedelta(seconds=30),
             retry_policy=STATE_UPDATE_RETRY,
         )
 
     return await workflow.execute_activity(
         update_document_state,
-        args=[workflow_id, stage, page_count, chunk_count, error_message],
+        args=args,
         start_to_close_timeout=timedelta(seconds=30),
         retry_policy=STATE_UPDATE_RETRY,
     )

scripts/migrate_sqlite_to_postgres.py

@@ -45,6 +45,7 @@
 from psycopg.types.json import Jsonb  # noqa: E402
 
 from pipeline import db, migrate  # noqa: E402
+from pipeline.instances import prod_stage_disabled  # noqa: E402
 from pipeline.models import (  # noqa: E402
     ChunkingRun,
     DocumentMetadata,
@@ -534,6 +535,7 @@ def migrate_document(
 
     catalog_id = _catalog_for(conn, _instance_for(conn, _instance_of(doc)), kind)
     code = (_get(doc, "scheme_code") or "").strip().lower()
+    original_code = code
     entry = registry.get(code) if kind == "scheme" and code else None
     if entry is not None:
         consumed.add(code)
@@ -553,6 +555,13 @@ def migrate_document(
     if unique != code:
         report.notes.append(f"{workflow_id}: code {code!r} already used in its catalog, stored as {unique!r}")
         code = unique
+    if original_code and code != original_code:
+        # The legacy index still carries the old code on its points. Postgres now
+        # advertises a different one, so the document must be re-indexed to match.
+        metadata["reindex_required"] = True
+        metadata["reindex_reason"] = (
+            f"scheme_code changed during migration: {original_code!r} -> {code!r}"
+        )
 
     created_at = parse_ts(doc["created_at"]) or datetime.now(timezone.utc)
     updated_at = parse_ts(doc["updated_at"]) or created_at
@@ -570,6 +579,12 @@ def migrate_document(
     report.count("documents", "inserted")
 
     workflows, job_ids = _build_workflows(sq, doc, uploader, report)
+    if prod_stage_disabled():
+        # The legacy deployment finished at DEV ingest; catalog_status needs the
+        # marker to count those completed runs as live without a PROD index entry.
+        for w in workflows:
+            if w["type"] in ("pipeline", "ocr_only"):
+                w["config"].setdefault("skip_prod", True)
     root = next((w for w in workflows if w["type"] in ("pipeline", "ocr_only")), workflows[0])
     for w in workflows:
         started_by = w["started_by"]

tests/test_activities.py

@@ -466,6 +466,30 @@ def test_update_state(self, db_connection):
         assert doc["page_count"] == 5
 
 
+class TestUpdateDocumentStateRunOwnership:
+    @pytest.mark.unit
+    def test_late_update_from_older_run_does_not_touch_newer_job(self, db_connection):
+        from pipeline.activities import update_document_state
+
+        wf = "state-owner-test"
+        db_connection.upsert_document(
+            kind="scheme", code=_next_code(), workflow_id=wf, document_id="doc-owner",
+            filename="t.pdf", filepath="/app/books/t.pdf", stage="ingesting",
+        )
+        retry_id = db_connection.create_document_job(
+            wf, "ocr_retry", temporal_workflow_id=f"{wf}-retry-ocr-1", current_stage="ocr_processing",
+        )
+
+        asyncio.run(update_document_state(
+            workflow_id=wf, stage="completed", run_workflow_id=wf,
+        ))
+
+        retry = db_connection.get_document_job(retry_id)
+        assert (retry["current_stage"], retry["status"]) == ("ocr_processing", "running")
+        pipeline = next(j for j in db_connection.list_document_jobs(wf) if j["job_type"] == "pipeline")
+        assert (pipeline["current_stage"], pipeline["status"]) == ("completed", "completed")
+
+
 class TestMinIOClient:
     """Tests for MinIO client creation."""
 

tests/test_db.py

@@ -716,6 +716,36 @@ def test_stage_updates_follow_the_current_workflow(self, db_connection):
         by_type = {j["job_type"]: j["current_stage"] for j in db_connection.list_document_jobs("wf-follow")}
         assert by_type == {"pipeline": "completed", "ocr_retry": "ocr_review"}
 
+    @pytest.mark.db
+    @pytest.mark.unit
+    def test_stage_update_from_an_older_run_leaves_the_newer_run_alone(self, db_connection):
+        _new_doc(db_connection, "wf-owner", stage="ingesting")
+        db_connection.create_document_job(
+            "wf-owner", "ocr_retry", temporal_workflow_id="wf-owner-retry-ocr-1",
+            current_stage="ocr_processing",
+        )
+
+        # The original pipeline run reports late, after the retry has started.
+        returned = db_connection.update_document_stage(
+            "wf-owner", "completed", run_workflow_id="wf-owner"
+        )
+
+        by_type = {j["job_type"]: j["current_stage"] for j in db_connection.list_document_jobs("wf-owner")}
+        assert by_type == {"pipeline": "completed", "ocr_retry": "ocr_processing"}
+        assert db_connection.get_document("wf-owner")["stage"] == "ocr_processing"
+        pipeline_job = next(j for j in db_connection.list_document_jobs("wf-owner") if j["job_type"] == "pipeline")
+        assert returned == pipeline_job["id"]
+
+    @pytest.mark.db
+    @pytest.mark.unit
+    def test_stage_update_with_an_unknown_run_falls_back_to_the_current_workflow(self, db_connection):
+        _new_doc(db_connection, "wf-fallback", stage="completed")
+        db_connection.create_document_job("wf-fallback", "ocr_retry", current_stage="ocr_processing")
+
+        db_connection.update_document_stage("wf-fallback", "ocr_review", run_workflow_id="no-such-run")
+
+        assert db_connection.get_document("wf-fallback")["stage"] == "ocr_review"
+
     @pytest.mark.db
     @pytest.mark.unit
     def test_stage_timestamps_are_set_once(self, db_connection):
@@ -1037,13 +1067,25 @@ def _scheme_doc(db, workflow_id, code, stage, kind="scheme", **kwargs):
             name=code.upper(), **kwargs,
         )
 
+    @staticmethod
+    def _skip_prod(db, workflow_id):
+        """Mark the document's run as one that skipped PROD, which makes completed mean live."""
+        with db.get_connection() as conn:
+            conn.execute(
+                "UPDATE workflows SET config = config || '{\"skip_prod\": true}'::jsonb "
+                "WHERE temporal_workflow_id = %s",
+                (workflow_id,),
+            )
+
     @pytest.mark.db
     @pytest.mark.unit
     def test_status_rules(self, db_connection):
         db = db_connection
         self._scheme_doc(db, "wf-live", "live-one", "completed")
         self._scheme_doc(db, "wf-stale", "stale-one", "completed")
         db.mark_document_reindex_required("wf-stale", True, "edit")
+        for wf in ("wf-live", "wf-stale", "wf-off"):
+            self._skip_prod(db, wf)
         self._scheme_doc(db, "wf-gate", "gate-one", "approval_for_prod")
         self._scheme_doc(db, "wf-wip", "wip-one", "chunk_review")
         self._scheme_doc(db, "wf-off", "off-one", "completed")
@@ -1065,6 +1107,8 @@ def test_computed_columns(self, db_connection, monkeypatch):
         db = db_connection
         self._scheme_doc(db, "wf-a", "scheme-a", "completed", instance="mh")
         self._scheme_doc(db, "wf-b", "scheme-b", "completed", instance="up")
+        self._skip_prod(db, "wf-a")
+        self._skip_prod(db, "wf-b")
         db.save_chunks("wf-a", [{"chunk_number": 1, "original_text": "x"}])
         db.save_chunks("wf-b", [{"chunk_number": 1, "original_text": "y"},
                                 {"chunk_number": 2, "original_text": "z"}])

tests/test_migrate_sqlite_to_postgres.py

@@ -271,7 +271,9 @@ def test_dry_run_writes_nothing_but_reports_everything(self, db_connection, post
         assert report.tables["documents"]["skipped"] == 2
         assert report.mismatches() == []
 
-    def test_a_full_run_maps_every_table(self, db_connection, postgres_url, legacy_db):
+    def test_a_full_run_maps_every_table(self, db_connection, postgres_url, legacy_db, monkeypatch):
+        # The fixture documents only ever reached the DEV index, as in a deployment with PROD off.
+        monkeypatch.setenv("DISABLE_PROD_SETTING", "true")
         db = db_connection
         report = _load_script().run_migration(legacy_db, postgres_url)
 
@@ -421,6 +423,43 @@ def test_unusable_values_are_repaired_and_noted(self, db_connection, postgres_ur
         notes = " ".join(report.notes)
         assert "valid_to cleared" in notes and "not a valid slug, code derived" in notes
 
+    def test_a_dev_only_legacy_document_is_not_live_when_prod_is_enabled(
+        self, db_connection, postgres_url, legacy_db, monkeypatch
+    ):
+        monkeypatch.delenv("DISABLE_PROD_SETTING", raising=False)
+        _load_script().run_migration(legacy_db, postgres_url)
+
+        with db_connection.get_connection() as conn:
+            status = conn.execute(
+                "SELECT status FROM catalog_status WHERE code = 'pm-kmy'"
+            ).fetchone()["status"]
+        assert status == "disabled"
+
+    def test_renamed_code_flags_the_document_for_reindex(self, db_connection, postgres_url, legacy_db):
+        """A suffixed code no longer matches the points in the old index, so it must be re-indexed."""
+        dup = "doc-dup-scheme"
+        conn = sqlite3.connect(legacy_db)
+        _insert(
+            conn, "documents", workflow_id=dup, document_id="dup-scheme", filename="dup.pdf",
+            filepath="/tmp/dup.pdf", stage="completed", created_at="2026-05-04T10:00:00",
+            updated_at="2026-05-04T10:00:00", instance="mh", document_kind="scheme",
+            scheme_code="pm-kmy", reindex_required=0,
+        )
+        conn.commit()
+        conn.close()
+
+        report = _load_script().run_migration(legacy_db, postgres_url)
+
+        renamed = db_connection.get_document(dup)
+        assert renamed["code"] == "pm-kmy-2"
+        assert renamed["reindex_required"] is True
+        assert "'pm-kmy' -> 'pm-kmy-2'" in renamed["reindex_reason"]
+        assert any("stored as 'pm-kmy-2'" in n for n in report.notes)
+
+        kept = db_connection.get_document(WF1)
+        assert kept["code"] == "pm-kmy"
+        assert kept["reindex_reason"] == "edit"
+
     def test_running_it_twice_changes_nothing(self, db_connection, postgres_url, legacy_db):
         script = _load_script()
         script.run_migration(legacy_db, postgres_url)