From 3e73c191d7bda0701fdbc76c57be119aee3adfc9 Mon Sep 17 00:00:00 2001 From: Naveen Date: Thu, 12 Nov 2026 16:06:55 +0530 Subject: [PATCH 1/3] fix(migration): flag documents whose code was renamed for reindex When the importer suffixes a duplicate scheme_code (or repairs an invalid one), the existing index points still carry the old code. Mark the document reindex_required with a reason so catalog_status reports pending_reindex. --- scripts/migrate_sqlite_to_postgres.py | 8 ++++++++ tests/test_migrate_sqlite_to_postgres.py | 25 ++++++++++++++++++++++++ 2 files changed, 33 insertions(+) diff --git a/scripts/migrate_sqlite_to_postgres.py b/scripts/migrate_sqlite_to_postgres.py index 366e54b..008103a 100644 --- a/scripts/migrate_sqlite_to_postgres.py +++ b/scripts/migrate_sqlite_to_postgres.py @@ -534,6 +534,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 +554,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 diff --git a/tests/test_migrate_sqlite_to_postgres.py b/tests/test_migrate_sqlite_to_postgres.py index 5af95ae..089d501 100644 --- a/tests/test_migrate_sqlite_to_postgres.py +++ b/tests/test_migrate_sqlite_to_postgres.py @@ -421,6 +421,31 @@ 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_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) From a0f269edc8b19f86ed4a69cc8e959d53d1a85763 Mon Sep 17 00:00:00 2001 From: Naveen Date: Thu, 12 Nov 2026 16:08:10 +0530 Subject: [PATCH 2/3] fix(workflows): write stage updates to the reporting run update_document_stage now takes the reporting run's Temporal workflow id and updates that run's row instead of the document's newest workflow, so a late update from an older run cannot change a newer run's stage or job status. --- pipeline/activities.py | 12 +++++++++--- pipeline/db.py | 37 ++++++++++++++++++++++++++++--------- pipeline/workflows.py | 9 +++++++-- tests/test_activities.py | 24 ++++++++++++++++++++++++ tests/test_db.py | 30 ++++++++++++++++++++++++++++++ tests/test_workflows.py | 1 + 6 files changed, 99 insertions(+), 14 deletions(-) diff --git a/pipeline/activities.py b/pipeline/activities.py index cac2215..5da0248 100644 --- a/pipeline/activities.py +++ b/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 { diff --git a/pipeline/db.py b/pipeline/db.py index b8a70e8..a055664 100644 --- a/pipeline/db.py +++ b/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]: diff --git a/pipeline/workflows.py b/pipeline/workflows.py index 4ce63fb..45b7eb1 100644 --- a/pipeline/workflows.py +++ b/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, ) diff --git a/tests/test_activities.py b/tests/test_activities.py index 89a561c..35f0dce 100644 --- a/tests/test_activities.py +++ b/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.""" diff --git a/tests/test_db.py b/tests/test_db.py index 01820a6..169995e 100644 --- a/tests/test_db.py +++ b/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): diff --git a/tests/test_workflows.py b/tests/test_workflows.py index d56c608..957ae0b 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -99,6 +99,7 @@ async def update_document_state( page_count: int = 0, chunk_count: int = 0, error_message: str = None, + run_workflow_id: str = None, ) -> dict: call_log.append(f"state:{stage}") return {} From e7b30557a036a6ec27702af1e465b74be72260d3 Mon Sep 17 00:00:00 2001 From: Naveen Date: Thu, 12 Nov 2026 16:11:39 +0530 Subject: [PATCH 3/3] fix(catalog): live requires a PROD index, not just a completed run catalog_status counted a document as live when its latest workflow was completed. A DEV-only document therefore showed as live, and starting a new run hid a live one. Migration 0003 redefines the view: live needs some completed workflow and either metadata.index_status.prod.status = 'indexed' or a completed run that skipped PROD (workflows.config.skip_prod, recorded at upload from DISABLE_PROD_SETTING). The importer sets skip_prod on legacy runs when DISABLE_PROD_SETTING is on. --- docs/POSTGRES_SCHEMA.md | 4 +- .../0003_catalog_live_requires_prod_index.sql | 47 +++++++++++++++++ pipeline/api/documents.py | 1 + pipeline/api/upload.py | 1 + scripts/migrate_sqlite_to_postgres.py | 7 +++ tests/test_db.py | 14 ++++++ tests/test_migrate_sqlite_to_postgres.py | 16 +++++- tests/test_schema_v2.py | 50 ++++++++++++++++++- tests/test_scheme_catalog.py | 18 ++++++- 9 files changed, 152 insertions(+), 6 deletions(-) create mode 100644 migrations/0003_catalog_live_requires_prod_index.sql diff --git a/docs/POSTGRES_SCHEMA.md b/docs/POSTGRES_SCHEMA.md index f9e260e..67a1803 100644 --- a/docs/POSTGRES_SCHEMA.md +++ b/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. --- diff --git a/migrations/0003_catalog_live_requires_prod_index.sql b/migrations/0003_catalog_live_requires_prod_index.sql new file mode 100644 index 0000000..c55f595 --- /dev/null +++ b/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; diff --git a/pipeline/api/documents.py b/pipeline/api/documents.py index b431755..938fc3e 100644 --- a/pipeline/api/documents.py +++ b/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, diff --git a/pipeline/api/upload.py b/pipeline/api/upload.py index b7d4be1..d631d0e 100644 --- a/pipeline/api/upload.py +++ b/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, diff --git a/scripts/migrate_sqlite_to_postgres.py b/scripts/migrate_sqlite_to_postgres.py index 008103a..6ce12c8 100644 --- a/scripts/migrate_sqlite_to_postgres.py +++ b/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, @@ -578,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"] diff --git a/tests/test_db.py b/tests/test_db.py index 169995e..b4bea7d 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -1067,6 +1067,16 @@ 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): @@ -1074,6 +1084,8 @@ def test_status_rules(self, 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") @@ -1095,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"}]) diff --git a/tests/test_migrate_sqlite_to_postgres.py b/tests/test_migrate_sqlite_to_postgres.py index 089d501..5867590 100644 --- a/tests/test_migrate_sqlite_to_postgres.py +++ b/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,18 @@ 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" diff --git a/tests/test_schema_v2.py b/tests/test_schema_v2.py index bfb8c77..4de3f56 100644 --- a/tests/test_schema_v2.py +++ b/tests/test_schema_v2.py @@ -26,6 +26,14 @@ def _doc(db, workflow_id, **kwargs): return db.upsert_document(workflow_id=workflow_id, **kwargs) +def _prod_indexed(db, workflow_id, monkeypatch, collection="prod-docs"): + """Record a successful PROD index, which is what makes a completed document live.""" + monkeypatch.setenv("PROD_VECTOR_DB_COLLECTION_NAME", collection) + db.upsert_document_index_status( + workflow_id, collection, last_indexed_at="2026-05-01T10:00:00+00:00", status="indexed" + ) + + def _instance(db, instance="mh"): return db.get_or_create_instance(instance) @@ -501,7 +509,7 @@ def test_one_row_per_document_with_catalog_instance_and_type(self, db_connection @pytest.mark.db @pytest.mark.unit - def test_status_follows_stage_flags_and_disabling(self, db_connection): + def test_status_follows_stage_flags_and_disabling(self, db_connection, monkeypatch): db = db_connection _doc(db, "wf-live", code="live", stage="completed") _doc(db, "wf-stale", code="stale", stage="completed") @@ -514,6 +522,8 @@ def test_status_follows_stage_flags_and_disabling(self, db_connection): db.update_document_fields("wf-hidden", catalog_visible=False) _doc(db, "wf-gate-off", code="gate-off", stage="approval_for_prod") db.set_document_disabled("wf-gate-off", True) + for wf in ("wf-live", "wf-stale", "wf-off", "wf-hidden"): + _prod_indexed(db, wf, monkeypatch) status = {code: row["status"] for code, row in self._rows(db).items()} assert status == { @@ -526,6 +536,40 @@ def test_status_follows_stage_flags_and_disabling(self, db_connection): "gate-off": "disabled", } + @pytest.mark.db + @pytest.mark.unit + def test_a_dev_only_completed_document_is_not_live(self, db_connection): + db = db_connection + _doc(db, "wf-dev", code="dev-only", stage="completed") + db.upsert_document_index_status("wf-dev", "documents-index", status="indexed") + assert self._rows(db)["dev-only"]["status"] == "disabled" + + @pytest.mark.db + @pytest.mark.unit + def test_a_completed_run_that_skipped_prod_is_live(self, db_connection): + db = db_connection + _doc(db, "wf-skip", code="skipped", stage="completed") + with db.get_connection() as conn: + conn.execute( + "UPDATE workflows SET config = config || '{\"skip_prod\": true}'::jsonb " + "WHERE temporal_workflow_id = 'wf-skip'" + ) + assert self._rows(db)["skipped"]["status"] == "live" + + @pytest.mark.db + @pytest.mark.unit + def test_a_new_run_does_not_hide_a_live_document(self, db_connection, monkeypatch): + db = db_connection + _doc(db, "wf-run", code="running", stage="completed") + _prod_indexed(db, "wf-run", monkeypatch) + assert self._rows(db)["running"]["status"] == "live" + + db.create_document_job( + "wf-run", "reingest", temporal_workflow_id="wf-run-reingest-1", current_stage="ingesting", + ) + + assert self._rows(db)["running"]["status"] == "live" + @pytest.mark.db @pytest.mark.unit def test_chunk_count_is_counted_from_chunks(self, db_connection): @@ -556,7 +600,9 @@ def test_promoted_at_comes_from_the_prod_index_of_live_documents_only(self, db_c _doc(db, "wf-p2", code="not-live", stage="chunk_review") _doc(db, "wf-p3", code="dev-only", stage="completed") for wf in ("wf-p1", "wf-p2"): - db.upsert_document_index_status(wf, "prod-docs", last_indexed_at="2026-05-01T10:00:00+00:00") + db.upsert_document_index_status( + wf, "prod-docs", last_indexed_at="2026-05-01T10:00:00+00:00", status="indexed" + ) db.upsert_document_index_status("wf-p3", "documents-index", last_indexed_at="2026-05-02T10:00:00+00:00") rows = self._rows(db) assert rows["promoted"]["promoted_at"] is not None diff --git a/tests/test_scheme_catalog.py b/tests/test_scheme_catalog.py index 4e4cab7..30130e7 100644 --- a/tests/test_scheme_catalog.py +++ b/tests/test_scheme_catalog.py @@ -14,7 +14,23 @@ def catalog_db(db_connection): def _doc(db, workflow_id, *, code, kind="scheme", name=None, stage="completed", instance="default", filename=None, **extra): - """Register a document the v2 way: it needs a kind (catalog) and a code.""" + """Register a document the v2 way: it needs a kind (catalog) and a code. + + A completed document counts as live here because its run skipped PROD (the + ``DISABLE_PROD_SETTING`` deployment); see ``catalog_status`` for the rule. + """ + doc_id = _upsert(db, workflow_id, filename, stage, instance, kind, code, name, extra) + if stage == "completed": + with db.get_connection() as conn: + conn.execute( + "UPDATE workflows SET config = config || '{\"skip_prod\": true}'::jsonb " + "WHERE temporal_workflow_id = %s", + (workflow_id,), + ) + return doc_id + + +def _upsert(db, workflow_id, filename, stage, instance, kind, code, name, extra): return db.upsert_document( workflow_id=workflow_id, document_id=f"doc-{workflow_id}",