From bf1710bd681fffc932af73da8a546ca5f427f168 Mon Sep 17 00:00:00 2001 From: Aayush Date: Sun, 30 Aug 2026 06:53:58 +0530 Subject: [PATCH 1/4] fix(#56): hide documents via query_enabled instead of Marqo purge Include off and Delete flip the search flag so operators can restore without reingest. --- docs/api-contracts.md | 19 +++-- docs/ingestion-pipeline-design.md | 18 ++-- pipeline/db.py | 28 +++++++ pipeline/ingestion_records.py | 3 +- pipeline/routers/content.py | 39 ++++----- pipeline/routers/documents.py | 64 ++++++--------- pipeline/routers/documents_actions.py | 2 +- pipeline/services/indexes.py | 68 +++++++++++++++ pipeline/services/search.py | 5 +- pipeline/vector_store.py | 25 +++++- tests/test_activities.py | 10 ++- tests/test_db.py | 40 +++++++++ tests/test_document_lifecycle.py | 114 +++++++++++++++++--------- tests/test_marqo_doc_scope.py | 33 +++++--- tests/test_search_service.py | 1 + tests/test_tenant_isolation.py | 20 +++-- tests/test_vector_store.py | 8 ++ ui/src/views/DocumentOpsView.jsx | 18 ++-- 18 files changed, 358 insertions(+), 157 deletions(-) diff --git a/docs/api-contracts.md b/docs/api-contracts.md index 4ff64b1..28578b4 100644 --- a/docs/api-contracts.md +++ b/docs/api-contracts.md @@ -370,9 +370,9 @@ Permission: `review`. **Exclude semantics** -- Setting `is_excluded: true` when the document `stage` is `completed` also - removes that chunk from Marqo -- Marks reindex / dirty as needed for later republish +- Setting `is_excluded` when the document `stage` is `completed` flips + `query_enabled` on that Marqo record (no delete, no reingest) +- Text or tag edits still mark reindex / dirty for later republish ### `PUT /documents/{workflow_id}/chunks/{chunk_num}/tags` @@ -480,13 +480,13 @@ Soft-delete document. Permission: `admin`. | Query | Type | Default | Notes | |---|---|---|---| -| `remove_from_search` | bool | `true` | Remove all chunks from Marqo | +| `remove_from_search` | bool | `true` | Hide all chunks in Marqo (`query_enabled:false`) | | `purge_artifacts` | bool | `false` | Delete listed MinIO objects after disable. Default keeps blobs so restore still has sources. | **Effects** 1. Cancel running Temporal workflow if possible -2. Optionally remove chunks from Marqo (fail-closed: disable is not flipped if this fails). Every recorded `document_index_status` index is purged, not only the currently resolved physical index; a row is marked `removed` only after that index's purge succeeds. +2. Optionally hide chunks in Marqo via `query_enabled:false` (fail-closed: disable is not flipped if this fails). Every recorded `document_index_status` index plus the resolved physical index is updated. Records are not deleted. 3. Set `is_disabled=true` in SQLite, turn queries off, exclude chunks 4. Report artifact GC plan; apply MinIO deletes only if `purge_artifacts=true` @@ -495,8 +495,9 @@ There is no HTTP hard-delete. `db.delete_document` refuses a documents-row-only delete; pass `cascade=True` to drop child SQLite rows after MinIO artifacts have `purged_at` (unpurged `minio://` objects refuse the cascade). -**Response** includes `artifact_purge` (`apply`, `would_purge` / `purged`, -`retained`, `already_purged`, `errors`, plus `*_count` fields). +**Response** includes `marqo_updated` (flag flips), `marqo_deleted` (0 on this path), +and `artifact_purge` (`apply`, `would_purge` / `purged`, `retained`, +`already_purged`, `errors`, plus `*_count` fields). --- @@ -517,8 +518,8 @@ with `purged_at` set. CLI equivalent: `scripts/purge_document_artifacts.py` ### `POST /documents/{workflow_id}/restore` -Clear `is_disabled` only. Does **not** automatically re-index Marqo — use -reingest. Permission: `admin`. +Clear `is_disabled` only. Chunks stay excluded until Include is turned on +(flag flip; no reingest). Permission: `admin`. ```json { diff --git a/docs/ingestion-pipeline-design.md b/docs/ingestion-pipeline-design.md index 1b067d1..57de03f 100644 --- a/docs/ingestion-pipeline-design.md +++ b/docs/ingestion-pipeline-design.md @@ -205,13 +205,13 @@ There is **no** bulk approve-ingestion endpoint (only OCR / translation / chunks Activity: `ingest_document_from_db` (builds payload, may export to MinIO, then calls `ingest_to_marqo`) -- Loads chunks including excluded, then **skips** `is_excluded` +- Loads all chunks, including excluded, and writes `query_enabled` from `is_excluded` - Writes tensor + filterable metadata (`doc_id`, `chunk_num`, `instance`, tags, …) - Updates index status; stage → `completed` -Reingest **adds/updates** documents in Marqo from current SQLite chunks; it does -not by itself delete older Marqo hits for edited text. Lifecycle Include-off / -Delete paths **do** remove hits from Marqo. +Reingest **adds/updates** documents in Marqo from current SQLite chunks (excluded +chunks stay in the index with `query_enabled:false`). Include-off / Delete flip +that flag; they do not delete records. --- @@ -226,7 +226,7 @@ These re-drive a stage without restarting the whole pipeline | `OcrOnlyWorkflow` | `POST …/retry-ocr` | Re-run OCR → stop at OCR review | | `TranslationOnlyWorkflow` | `POST …/retry-translation` | Translate again → translation review | | `ChunkingOnlyWorkflow` | `POST …/retry-chunking` | Re-chunk → chunk review | -| `ReingestionWorkflow` | `POST …/reingest` (alias `…/retry-ingestion`) | Push current non-excluded SQLite chunks to Marqo | +| `ReingestionWorkflow` | `POST …/reingest` (alias `…/retry-ingestion`) | Push current SQLite chunks to Marqo (`query_enabled` from `is_excluded`) | **Reconcile** (`POST …/reconcile` and bulk `POST /documents/reconcile`): @@ -256,11 +256,11 @@ These are separate from the stage machine but part of day-2 operations: | Action | Behavior | |---|---| -| **Document Delete** (`DELETE …`) | Soft-hide (`is_disabled`); optionally remove all chunks from Marqo (`remove_from_search=true` by default). MinIO kept unless `purge_artifacts=true`. | +| **Document Delete** (`DELETE …`) | Soft-hide (`is_disabled`); optionally hide all chunks in Marqo via `query_enabled` (`remove_from_search=true` by default). MinIO kept unless `purge_artifacts=true`. | | **Purge artifacts** (`POST …/purge-artifacts`) | Dry-run by default; `apply=true` deletes listed MinIO objects for a disabled doc. | -| **Restore** | Clears `is_disabled` only. Chunks removed from Marqo are **not** put back automatically — use **reingest**. | -| **Chunk exclude** (`PATCH …/chunks/{n}` with `is_excluded=true`) | Hide one chunk from future ingest; if doc `stage=completed`, also remove that chunk from Marqo. | -| **Reingest** | Re-publish current non-excluded SQLite chunks to Marqo. | +| **Restore** | Clears `is_disabled` only. Include-on flips `query_enabled` back (no reingest). | +| **Chunk exclude** (`PATCH …/chunks/{n}` with `is_excluded=true`) | Hide one chunk from search via `query_enabled`; row stays. | +| **Reingest** | Re-publish current SQLite chunks to Marqo (used after text/tag edits, not Include). | --- diff --git a/pipeline/db.py b/pipeline/db.py index 3efa850..aa62ec4 100644 --- a/pipeline/db.py +++ b/pipeline/db.py @@ -8,6 +8,7 @@ import sqlite3 import os import hashlib +from collections import Counter from datetime import datetime from pathlib import Path from contextlib import contextmanager @@ -3050,6 +3051,7 @@ def save_chunks(workflow_id: str, chunks: list[dict]): Save all chunks for a document (bulk upsert). Called when workflow completes to persist data. """ + lost_exclusions: list[str] = [] with _db_lock: with get_connection() as conn: existing_version = conn.execute( @@ -3057,6 +3059,20 @@ def save_chunks(workflow_id: str, chunks: list[dict]): (workflow_id,), ).fetchone() next_version = int(existing_version["max_version"] or 0) + 1 + old_excluded = conn.execute( + """ + SELECT original_text FROM chunks + WHERE workflow_id = ? AND COALESCE(is_excluded, 0) = 1 + """, + (workflow_id,), + ).fetchall() + pool = Counter((row["original_text"] or "").strip() for row in old_excluded) + for chunk in chunks: + key = (chunk.get("original_text") or "").strip() + if pool[key] > 0: + chunk["is_excluded"] = True + pool[key] -= 1 + lost_exclusions = [key for key, remaining in pool.items() if remaining > 0] conn.execute("DELETE FROM chunks WHERE workflow_id = ?", (workflow_id,)) # Preserve manual reviewer tags; only clear auto tags on re-chunk. conn.execute( @@ -3110,6 +3126,18 @@ def save_chunks(workflow_id: str, chunks: list[dict]): (workflow_id, workflow_id), ) conn.commit() + if lost_exclusions: + doc = get_document(workflow_id) or {} + log_audit( + workflow_id=workflow_id, + document_id=doc.get("document_id") or workflow_id, + action_type="chunk_exclusion_lost_on_rechunk", + entity_type="document", + metadata={ + "lost_count": len(lost_exclusions), + "lost_text_prefixes": [text[:80] for text in lost_exclusions[:20]], + }, + ) def get_chunks(workflow_id: str, include_excluded: bool = False) -> list[dict]: diff --git a/pipeline/ingestion_records.py b/pipeline/ingestion_records.py index 657c8c3..ecc9d9c 100644 --- a/pipeline/ingestion_records.py +++ b/pipeline/ingestion_records.py @@ -289,8 +289,6 @@ def prepare_records( records = [] for chunk in chunks: - if chunk.get("is_excluded", False): - continue raw_text = chunk.get("edited_text") or chunk.get("original_text", "") chunk_num = chunk.get("chunk_number", 0) text = clean_text_for_ingestion(raw_text) @@ -323,6 +321,7 @@ def prepare_records( "page_start": chunk.get("page_start", 1), "page_end": chunk.get("page_end", 1), "is_reference": is_reference_section(text), + "query_enabled": not bool(chunk.get("is_excluded", False)), "quality_score": float(quality_score) if str(quality_score).strip().replace(".", "", 1).isdigit() else 0.0, diff --git a/pipeline/routers/content.py b/pipeline/routers/content.py index be2709d..e1f4e95 100644 --- a/pipeline/routers/content.py +++ b/pipeline/routers/content.py @@ -241,6 +241,15 @@ async def update_chunk( if not old_chunk: raise HTTPException(404, f"Chunk {chunk_num} not found") + if ( + data.is_excluded is not None + and bool(data.is_excluded) != bool(old_chunk.get("is_excluded", False)) + and doc.get("stage") == "completed" + ): + indexes.apply_document_query_enabled( + doc, workflow_id, enabled=not bool(data.is_excluded), chunk_num=chunk_num + ) + updated = db.update_chunk( workflow_id, chunk_num, @@ -287,26 +296,6 @@ async def update_chunk( new_value=data.is_excluded ) - # If excluding a chunk and document is completed (already ingested), remove from Marqo - if data.is_excluded and not old_chunk.get("is_excluded", False): - if doc and doc.get("stage") == "completed": - doc_id = doc.get("document_id") - if doc_id: - target_index = indexes.resolve_index(doc.get("instance"), doc.get("index")) - if target_index is not None: - marqo_result = indexes.delete_single_chunk_from_marqo( - doc_id, chunk_num, index_name=target_index, - workflow_id=workflow_id, - ) - if marqo_result.get("deleted"): - documents.log_audit( - workflow_id=workflow_id, - action_type="chunk_removed_from_search", - entity_type="chunk", - entity_id=chunk_num, - metadata={"marqo_id": marqo_result.get("chunk_id")} - ) - if data.reviewer_notes is not None: documents.log_audit( workflow_id=workflow_id, @@ -349,8 +338,12 @@ async def update_chunk( new_value="|".join(sorted(t.key() for t in parsed)), ) - if data.edited_text is not None or data.is_excluded is not None or tags_changed: - reason = "Chunk tags changed; search index is out of sync" if tags_changed and data.edited_text is None and data.is_excluded is None else "Chunk content changed; search index is out of sync" + if data.edited_text is not None or tags_changed: + reason = ( + "Chunk tags changed; search index is out of sync" + if tags_changed and data.edited_text is None + else "Chunk content changed; search index is out of sync" + ) documents.mark_reindex_required( workflow_id, reason, @@ -633,7 +626,7 @@ async def get_document_marqo_status( doc["document_id"] ) sqlite_chunks = db.get_chunks(workflow_id, include_excluded=True) - sqlite_chunk_count = len([c for c in sqlite_chunks if not c.get("is_excluded")]) + sqlite_chunk_count = len(sqlite_chunks) # The document's tenant has no index of its own: report a graceful "no index" # status rather than querying (and leaking) another tenant's physical index. diff --git a/pipeline/routers/documents.py b/pipeline/routers/documents.py index 828965f..72bc829 100644 --- a/pipeline/routers/documents.py +++ b/pipeline/routers/documents.py @@ -633,7 +633,7 @@ async def disable_document( This performs a soft delete: - Marks the document as disabled in SQLite (hidden from list by default) - Turns query_enabled off and marks all SQLite chunks as excluded - - Optionally removes all chunks from Marqo search index + - Optionally hides all chunks in Marqo (query_enabled flag, no delete) - Cancels the workflow if still running - MinIO artifacts stay unless purge_artifacts=true (explicit, default false) @@ -642,7 +642,7 @@ async def disable_document( Args: workflow_id: The document workflow ID - remove_from_search: If True (default), removes chunks from Marqo index + remove_from_search: If True (default), hides chunks in Marqo purge_artifacts: If True, delete listed MinIO objects after disable Requires permission: admin. """ @@ -657,6 +657,7 @@ async def disable_document( "disabled": True, "workflow_cancelled": False, "chunks_excluded": 0, + "marqo_updated": 0, "marqo_deleted": 0, } @@ -665,24 +666,19 @@ async def disable_document( workflow_id ) - # Remove from Marqo FIRST if requested, so a failed purge cannot leave the - # document marked disabled while its chunks stay searchable (mirror the - # fail-closed ordering in set_document_query_enabled). Purge every recorded - # document_index_status index plus the currently resolved physical index; - # a per-tenant delete must never fall through to the default tenant's - # legacy index via a content-md5 doc_id collision. + # Flip search visibility FIRST so a failed update cannot leave the + # document marked disabled while its chunks stay searchable. Records stay + # in Marqo; query_enabled:false hides them. Every recorded index plus the + # resolved physical index is updated. When the tenant has no index, skip. if remove_from_search: - marqo_result = indexes.purge_document_search_indexes( - workflow_id=workflow_id, - document_id=doc.get("document_id"), - instance=doc.get("instance"), - logical_index=doc.get("index"), + result["marqo_updated"] = indexes.apply_document_query_enabled( + doc, workflow_id, False ) - result["marqo_deleted"] = int(marqo_result.get("deleted", 0) or 0) + result["marqo_deleted"] = 0 - # Mark as disabled in SQLite only after the purge succeeded. + # Mark as disabled in SQLite only after the Marqo flip succeeded. db.set_document_disabled(workflow_id, True) - # Same semantics as unchecking Include: off for queries until reingest after restore. + # Same search hide as unchecking Include; Restore + Include on brings search back. db.set_document_query_enabled(workflow_id, False) result["chunks_excluded"] = db.set_all_chunks_excluded(workflow_id, True) @@ -699,7 +695,7 @@ async def disable_document( "remove_from_search": remove_from_search, "purge_artifacts": purge_artifacts, "chunks_excluded": result["chunks_excluded"], - "marqo_deleted": result["marqo_deleted"], + "marqo_updated": result["marqo_updated"], "query_enabled": False, "artifacts_purged": result["artifact_purge"]["purged_count"], "artifacts_retained": result["artifact_purge"]["retained_count"], @@ -748,8 +744,8 @@ async def restore_document(workflow_id: str, user: RequireAdmin): """ Restore a soft-deleted (disabled) document into the list. - Chunks stay excluded and out of Marqo until the operator enables the - document for queries and reingests. + Chunks stay excluded and hidden in search until the operator turns Include on + (flag flip; no reingest). """ doc = access.require_document_for_user(workflow_id, user, permission=Permission.ADMIN) @@ -760,7 +756,7 @@ async def restore_document(workflow_id: str, user: RequireAdmin): workflow_id=workflow_id, document_id=doc.get("document_id", ""), action_type="restore_document", - metadata={"note": "chunks remain excluded; reingest required to republish"}, + metadata={"note": "chunks remain excluded; Include on restores search"}, ) return { @@ -813,37 +809,25 @@ async def set_document_query_enabled( """Enable or disable a document for search queries. When disabled: all chunks are excluded (same as unchecking Include on each) - and fully removed from Marqo. When enabled: chunks are included again and - reindex is marked required (reingest republishes to Marqo). + and hidden in Marqo via query_enabled. When enabled: chunks are included + again and the same flag is flipped back (no reingest). This does not soft-delete the document (it stays in the list). """ doc = access.require_document_for_user(workflow_id, user, permission=Permission.ADMIN) was_enabled = bool(doc["query_enabled"]) if doc.get("query_enabled") is not None else True chunks_touched = 0 - marqo_deleted = 0 + marqo_updated = 0 if not body.query_enabled: - # Purge Marqo before flipping DB so a failed purge does not leave - # "queries off" while chunks remain searchable. Every recorded index - # plus the currently resolved physical index is purged; status rows - # are marked removed only for indexes that actually succeeded. - marqo_result = indexes.purge_document_search_indexes( - workflow_id=workflow_id, - document_id=doc.get("document_id"), - instance=doc.get("instance"), - logical_index=doc.get("index"), - ) - marqo_deleted = int(marqo_result.get("deleted", 0) or 0) + # Flip Marqo before SQLite so a failed update does not leave + # "queries off" while chunks remain searchable. + marqo_updated = indexes.apply_document_query_enabled(doc, workflow_id, False) chunks_touched = db.set_all_chunks_excluded(workflow_id, True) updated = db.set_document_query_enabled(workflow_id, False) or doc elif not was_enabled and body.query_enabled: + marqo_updated = indexes.apply_document_query_enabled(doc, workflow_id, True) updated = db.set_document_query_enabled(workflow_id, True) or doc chunks_touched = db.set_all_chunks_excluded(workflow_id, False) - document_service.mark_reindex_required( - workflow_id, - "Document included for queries; reingest to republish chunks to Marqo", - metadata={"actor": user.user_id}, - ) updated = db.get_document(workflow_id) or updated else: updated = db.set_document_query_enabled(workflow_id, body.query_enabled) or doc @@ -858,7 +842,7 @@ async def set_document_query_enabled( metadata={ "actor": user.user_id, "chunks_touched": chunks_touched, - "marqo_deleted": marqo_deleted, + "marqo_updated": marqo_updated, }, ) return document_service.document_summary_from_row(updated) diff --git a/pipeline/routers/documents_actions.py b/pipeline/routers/documents_actions.py index a0a3f5c..8fa37b1 100644 --- a/pipeline/routers/documents_actions.py +++ b/pipeline/routers/documents_actions.py @@ -60,7 +60,7 @@ async def reingest_document( ) # Get chunks from SQLite - chunks = db.get_chunks(workflow_id, include_excluded=False) + chunks = db.get_chunks(workflow_id, include_excluded=True) if not chunks: raise HTTPException(400, f"No chunks found for document. The document may need to be reprocessed from scratch.") diff --git a/pipeline/services/indexes.py b/pipeline/services/indexes.py index 159d458..6a2e275 100644 --- a/pipeline/services/indexes.py +++ b/pipeline/services/indexes.py @@ -190,3 +190,71 @@ def purge_document_search_indexes( db.mark_document_search_removed(workflow_id, index_name=index_name) purged.append(index_name) return {"deleted": deleted, "indexes": purged} + + +def set_document_chunks_query_enabled( + document_id: str, + index_name: str, + workflow_id: str, + enabled: bool, + chunk_num: Optional[int] = None, +) -> dict: + """Flip ``query_enabled`` on one document's records (or one chunk). No delete.""" + store = vector_store.get_vector_store() + scope = vector_store.marqo_doc_scope_filter(document_id, workflow_id) + if chunk_num is not None: + scope = vector_store.merge_filter_strings( + scope, vector_store.term_filter("chunk_num", chunk_num) + ) + try: + hits = vector_store.search_all_hits( + store, + index_name, + scope, + ["doc_id", "workflow_id", "chunk_num"], + ) + except vector_store.VectorStoreError as exc: + if vector_store.index_missing_error(exc): + return {"updated": 0, "reason": "index_missing"} + return {"updated": 0, "error": str(exc)} + ids = [hit.get("_id") for hit in hits if hit.get("_id")] + try: + result = store.set_query_enabled(index_name, ids, enabled) + except vector_store.VectorStoreError as exc: + return {"updated": 0, "error": str(exc)} + result["index_name"] = index_name + return result + + +def apply_document_query_enabled( + doc: dict, + workflow_id: str, + enabled: bool, + chunk_num: Optional[int] = None, +) -> int: + """Flip ``query_enabled`` on every recorded index plus the resolved one.""" + doc_id = doc.get("document_id") + if not doc_id: + return 0 + names = { + row["index_name"] + for row in db.list_document_index_status(workflow_id) + if row.get("index_name") + } + target = resolve_index(doc.get("instance"), doc.get("index")) + if target: + names.add(target) + if not names: + return 0 + updated = 0 + for index_name in sorted(names): + result = set_document_chunks_query_enabled( + doc_id, index_name, workflow_id, enabled, chunk_num=chunk_num + ) + if result.get("error"): + raise HTTPException( + 502, + f"Failed to update search visibility ({index_name}): {result['error']}", + ) + updated += int(result.get("updated") or 0) + return updated diff --git a/pipeline/services/search.py b/pipeline/services/search.py index 9c0c8ba..ff7f316 100644 --- a/pipeline/services/search.py +++ b/pipeline/services/search.py @@ -16,6 +16,7 @@ from .. import db from ..vector_store import ( + QUERY_ENABLED_FILTER, VectorStore, VectorStoreError, build_domain_tags_filter, @@ -334,7 +335,9 @@ def run_search( "Use an index created with the passage schema that includes 'domain_tags' " "(for example: documents-index-tags)." ) - filter_string = merge_filter_strings(reference_filter, tag_filter, instance_filter) + filter_string = merge_filter_strings( + reference_filter, tag_filter, instance_filter, QUERY_ENABLED_FILTER + ) if filter_string: request["filter_string"] = filter_string diff --git a/pipeline/vector_store.py b/pipeline/vector_store.py index 145f7e8..c790061 100644 --- a/pipeline/vector_store.py +++ b/pipeline/vector_store.py @@ -216,6 +216,9 @@ def merge_filter_strings(*parts: str | None) -> str | None: return " AND ".join(clauses) +QUERY_ENABLED_FILTER = term_filter("query_enabled", "true") + + # -- domain tags ------------------------------------------------------------- # # Tags are stored in one flat, pipe-delimited ``domain_tags`` text field because a @@ -284,7 +287,7 @@ def build_domain_tags_filter(tags: Iterable[str]) -> str | None: # Optional fields: an index that predates them is a schema *drift*, not a # mismatch that should stop an ingest. -_OPTIONAL_PASSAGE_FIELDS = {"domain_tags", "instance"} +_OPTIONAL_PASSAGE_FIELDS = {"domain_tags", "instance", "query_enabled"} def passage_index_settings( @@ -320,6 +323,7 @@ def passage_index_settings( {"name": "page_start", "type": "int", "features": ["filter"]}, {"name": "page_end", "type": "int", "features": ["filter"]}, {"name": "is_reference", "type": "bool", "features": ["filter"]}, + {"name": "query_enabled", "type": "bool", "features": ["filter"]}, {"name": "quality_score", "type": "float", "features": ["filter"]}, {"name": "priority_rank", "type": "float", "features": ["filter"]}, {"name": "domain_tags", "type": "text", "features": ["filter"]}, @@ -744,6 +748,12 @@ def update_documents(self, index: str, records: Sequence[dict]) -> Any: """Partial update of existing records (ops backfill scripts).""" ... + def set_query_enabled( + self, index: str, record_ids: Sequence[str], enabled: bool + ) -> dict: + """Flip ``query_enabled`` on existing records. No re-embed.""" + ... + def delete_document( self, document_id: str, index: str, workflow_id: Optional[str] = None ) -> dict: @@ -924,6 +934,19 @@ def update_documents(self, index: str, records: Sequence[dict]) -> Any: except Exception as error: raise VectorStoreError(str(error)) from error + def set_query_enabled( + self, index: str, record_ids: Sequence[str], enabled: bool + ) -> dict: + """Partial-update ``query_enabled`` on existing records. No re-embed.""" + ids = [rid for rid in record_ids if rid] + if not ids: + return {"updated": 0} + self.update_documents( + index, + [{"_id": rid, "query_enabled": bool(enabled)} for rid in ids], + ) + return {"updated": len(ids)} + # -- purges -------------------------------------------------------------- # # These return a result dict rather than raising, because callers must diff --git a/tests/test_activities.py b/tests/test_activities.py index 6fef07a..92ed136 100644 --- a/tests/test_activities.py +++ b/tests/test_activities.py @@ -276,8 +276,8 @@ def test_prepare_records_basic(self): assert record["doc_id"] == "test-doc" @pytest.mark.unit - def test_prepare_records_excludes_excluded_chunks(self): - """Test that excluded chunks are not included in records.""" + def test_prepare_records_keeps_excluded_with_query_enabled_false(self): + """Excluded chunks stay in the payload so search can filter them.""" from pipeline.ingestion_records import prepare_ingestion_records chunks = [ @@ -305,8 +305,10 @@ def test_prepare_records_excludes_excluded_chunks(self): chunks=chunks ) - assert len(records) == 1 - assert records[0]["text"] == "Included" + assert len(records) == 2 + by_text = {r["text"]: r for r in records} + assert by_text["Included"]["query_enabled"] is True + assert by_text["Excluded"]["query_enabled"] is False @pytest.mark.unit def test_prepare_records_uses_edited_text(self): diff --git a/tests/test_db.py b/tests/test_db.py index 03c086e..f36cfdc 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -400,6 +400,46 @@ def test_update_chunk(self, db_connection, sample_document): assert chunk["edited_text"] == "Edited text" assert chunk["is_reviewed"] == 1 + @pytest.mark.db + @pytest.mark.unit + def test_save_chunks_carries_exclusion_by_unchanged_text(self, db_connection, sample_document): + wf = sample_document["workflow_id"] + db_connection.save_chunks( + wf, + [ + {"chunk_number": 1, "original_text": "keep me", "page_start": 1, "page_end": 1}, + {"chunk_number": 2, "original_text": "hide me", "page_start": 1, "page_end": 1}, + ], + ) + db_connection.update_chunk(wf, 2, is_excluded=True) + db_connection.save_chunks( + wf, + [ + {"chunk_number": 1, "original_text": "hide me", "page_start": 1, "page_end": 1}, + {"chunk_number": 2, "original_text": "keep me", "page_start": 1, "page_end": 1}, + ], + ) + chunks = {c["original_text"]: c for c in db_connection.get_chunks(wf, include_excluded=True)} + assert chunks["hide me"]["is_excluded"] is True + assert chunks["keep me"]["is_excluded"] is False + + @pytest.mark.db + @pytest.mark.unit + def test_save_chunks_audits_lost_exclusion(self, db_connection, sample_document): + wf = sample_document["workflow_id"] + db_connection.save_chunks( + wf, + [{"chunk_number": 1, "original_text": "gone soon", "page_start": 1, "page_end": 1}], + ) + db_connection.update_chunk(wf, 1, is_excluded=True) + db_connection.save_chunks( + wf, + [{"chunk_number": 1, "original_text": "brand new", "page_start": 1, "page_end": 1}], + ) + logs = db_connection.get_audit_logs(wf, action_type="chunk_exclusion_lost_on_rechunk") + assert logs + assert logs[0]["action_type"] == "chunk_exclusion_lost_on_rechunk" + class TestAuditLogging: """Tests for audit logging functionality.""" diff --git a/tests/test_document_lifecycle.py b/tests/test_document_lifecycle.py index a3f28dc..58e6949 100644 --- a/tests/test_document_lifecycle.py +++ b/tests/test_document_lifecycle.py @@ -192,7 +192,7 @@ def test_require_document_admin_pattern_for_lifecycle(lifecycle_doc): def test_query_enabled_route_requires_admin(lifecycle_doc, monkeypatch): - monkeypatch.setattr(indexes, "delete_chunks_from_marqo", lambda *a, **k: {"deleted": 0}) + monkeypatch.setattr(indexes, "apply_document_query_enabled", lambda *a, **k: 0) monkeypatch.setattr(indexes, "resolve_index", lambda *a, **k: "t-tenant-a-vet") with pytest.raises(HTTPException) as exc: @@ -215,6 +215,28 @@ def test_query_enabled_route_requires_admin(lifecycle_doc, monkeypatch): assert summary.query_enabled is False +def test_query_enable_does_not_mark_reindex(lifecycle_indexed_doc, monkeypatch): + monkeypatch.setattr(indexes, "apply_document_query_enabled", lambda *a, **k: 2) + admin = _admin_in("tenant-a") + _run( + documents.set_document_query_enabled( + lifecycle_indexed_doc, + DocumentQueryEnabledUpdate(query_enabled=False), + admin, + ) + ) + _run( + documents.set_document_query_enabled( + lifecycle_indexed_doc, + DocumentQueryEnabledUpdate(query_enabled=True), + admin, + ) + ) + row = db_mod.get_document(lifecycle_indexed_doc) + assert int(row["query_enabled"]) == 1 + assert int(row.get("reindex_required") or 0) == 0 + + def test_hard_delete_chunk_route_requires_admin(lifecycle_doc, monkeypatch): monkeypatch.setattr( indexes, "delete_single_chunk_from_marqo", lambda *a, **k: {"deleted": False, "reason": "not_found"} @@ -231,7 +253,7 @@ def test_hard_delete_chunk_route_requires_admin(lifecycle_doc, monkeypatch): def test_disable_document_route_requires_admin(lifecycle_doc, monkeypatch): - monkeypatch.setattr(indexes, "delete_chunks_from_marqo", lambda *a, **k: {"deleted": 0}) + monkeypatch.setattr(indexes, "apply_document_query_enabled", lambda *a, **k: 0) monkeypatch.setattr(indexes, "resolve_index", lambda *a, **k: "t-tenant-a-vet") with pytest.raises(HTTPException) as exc: @@ -304,14 +326,19 @@ def index(self, name): assert "error" not in one -def test_query_enabled_purge_uses_resolve_index(lifecycle_indexed_doc, monkeypatch): +def test_query_enabled_update_uses_resolve_index(lifecycle_indexed_doc, monkeypatch): calls = [] - def _fake_delete(doc_id, index_name="documents-index", workflow_id=None): - calls.append({"doc_id": doc_id, "index_name": index_name, "workflow_id": workflow_id}) - return {"deleted": 3, "index_name": index_name} + def _fake_apply(doc, workflow_id, enabled, chunk_num=None): + calls.append({ + "index_name": indexes.resolve_index(doc.get("instance"), doc.get("index")), + "workflow_id": workflow_id, + "enabled": enabled, + "chunk_num": chunk_num, + }) + return 2 - monkeypatch.setattr(indexes, "delete_chunks_from_marqo", _fake_delete) + monkeypatch.setattr(indexes, "apply_document_query_enabled", _fake_apply) _run( documents.set_document_query_enabled( @@ -322,8 +349,8 @@ def _fake_delete(doc_id, index_name="documents-index", workflow_id=None): ) assert len(calls) == 1 assert calls[0]["index_name"] == "t-tenant-a-vet" - # #73: the purge must be scoped to the document it was asked about. assert calls[0]["workflow_id"] == lifecycle_indexed_doc + assert calls[0]["enabled"] is False assert calls[0]["index_name"] != "documents-index" @@ -346,11 +373,16 @@ def _fake_single(doc_id, chunk_num, index_name="documents-index", workflow_id=No def test_chunk_exclude_on_completed_uses_resolve_index(lifecycle_indexed_doc, monkeypatch): calls = [] - def _fake_single(doc_id, chunk_num, index_name="documents-index", workflow_id=None): - calls.append({"doc_id": doc_id, "chunk_num": chunk_num, "index_name": index_name, "workflow_id": workflow_id}) - return {"deleted": True, "chunk_id": "c2"} + def _fake_apply(doc, workflow_id, enabled, chunk_num=None): + calls.append({ + "index_name": indexes.resolve_index(doc.get("instance"), doc.get("index")), + "workflow_id": workflow_id, + "enabled": enabled, + "chunk_num": chunk_num, + }) + return 1 - monkeypatch.setattr(indexes, "delete_single_chunk_from_marqo", _fake_single) + monkeypatch.setattr(indexes, "apply_document_query_enabled", _fake_apply) _run( content.update_chunk( @@ -362,11 +394,12 @@ def _fake_single(doc_id, chunk_num, index_name="documents-index", workflow_id=No ) assert len(calls) == 1 assert calls[0]["index_name"] == "t-tenant-a-vet" - # #73: the purge must be scoped to the document it was asked about. assert calls[0]["workflow_id"] == lifecycle_indexed_doc + assert calls[0]["chunk_num"] == 2 + assert calls[0]["enabled"] is False -def test_lifecycle_purge_skips_when_tenant_has_no_index(db_connection, monkeypatch): +def test_lifecycle_update_skips_when_tenant_has_no_index(db_connection, monkeypatch): db_mod.create_tenant_row("ghost", display_name="Ghost") db_mod.upsert_document( document_id="doc-ghost", @@ -378,30 +411,29 @@ def test_lifecycle_purge_skips_when_tenant_has_no_index(db_connection, monkeypat ) called = {"n": 0} - def _fake_delete(*a, **k): + def _fake_set(*a, **k): called["n"] += 1 - return {"deleted": 0} + return {"updated": 0} - monkeypatch.setattr(indexes, "delete_chunks_from_marqo", _fake_delete) - # ghost has no registered index -> resolve_index returns None -> skip purge + monkeypatch.setattr(indexes, "set_document_chunks_query_enabled", _fake_set) admin = _admin_in("ghost") res = _run(documents.disable_document("wf-ghost", admin, remove_from_search=True)) - assert res["marqo_deleted"] == 0 + assert res["marqo_updated"] == 0 assert called["n"] == 0 def test_disable_document_502_before_flip_on_marqo_error(lifecycle_indexed_doc, monkeypatch): - """A failed Marqo purge must 502 and leave the document NOT disabled — never - hidden-but-still-searchable (mirror set_document_query_enabled ordering).""" + """A failed Marqo update must 502 and leave the document NOT disabled.""" monkeypatch.setattr( - indexes, "delete_chunks_from_marqo", lambda *a, **k: {"deleted": 0, "error": "marqo down"} + indexes, + "set_document_chunks_query_enabled", + lambda *a, **k: {"updated": 0, "error": "marqo down"}, ) with pytest.raises(HTTPException) as exc: _run(documents.disable_document(lifecycle_indexed_doc, _admin_in("tenant-a"), remove_from_search=True)) assert exc.value.status_code == 502 - # DB was NOT flipped — the purge failed before any state change. row = db_mod.get_document(lifecycle_indexed_doc) assert int(row["is_disabled"]) == 0 assert int(row["query_enabled"]) == 1 @@ -562,25 +594,25 @@ def test_purge_artifacts_refuses_live_document(lifecycle_doc): assert exc.value.status_code == 400 -def test_disable_marks_index_status_removed(lifecycle_indexed_doc, monkeypatch): +def test_disable_keeps_index_status_indexed(lifecycle_indexed_doc, monkeypatch): db_mod.upsert_document_index_status( lifecycle_indexed_doc, "t-tenant-a-vet", status="indexed", chunk_count_indexed=2 ) - monkeypatch.setattr(indexes, "delete_chunks_from_marqo", lambda *a, **k: {"deleted": 2}) + monkeypatch.setattr( + indexes, "set_document_chunks_query_enabled", lambda *a, **k: {"updated": 2} + ) _run( documents.disable_document( lifecycle_indexed_doc, _admin_in("tenant-a"), remove_from_search=True ) ) status = db_mod.get_document_index_status(lifecycle_indexed_doc, "t-tenant-a-vet") - assert status["status"] == "removed" - assert int(status["chunk_count_indexed"]) == 0 + assert status["status"] == "indexed" + assert int(status["chunk_count_indexed"]) == 2 -def test_disable_purges_every_recorded_index_before_marking_removed( - lifecycle_indexed_doc, monkeypatch -): - """Historical indexes stay searchable unless they are purged, not just marked.""" +def test_disable_hides_every_recorded_index(lifecycle_indexed_doc, monkeypatch): + """Historical indexes stay in Marqo and must have query_enabled flipped too.""" db_mod.upsert_document_index_status( lifecycle_indexed_doc, "t-tenant-a-vet", status="indexed", chunk_count_indexed=2 ) @@ -589,22 +621,22 @@ def test_disable_purges_every_recorded_index_before_marking_removed( ) calls = [] - def _fake_delete(doc_id, index_name="documents-index", workflow_id=None): + def _fake_set(document_id, index_name, workflow_id, enabled, chunk_num=None): calls.append(index_name) - return {"deleted": 2, "index_name": index_name} + return {"updated": 2, "index_name": index_name} - monkeypatch.setattr(indexes, "delete_chunks_from_marqo", _fake_delete) + monkeypatch.setattr(indexes, "set_document_chunks_query_enabled", _fake_set) _run( documents.disable_document( lifecycle_indexed_doc, _admin_in("tenant-a"), remove_from_search=True ) ) assert set(calls) == {"t-tenant-a-vet", "old-index"} - assert db_mod.get_document_index_status(lifecycle_indexed_doc, "t-tenant-a-vet")["status"] == "removed" - assert db_mod.get_document_index_status(lifecycle_indexed_doc, "old-index")["status"] == "removed" + assert db_mod.get_document_index_status(lifecycle_indexed_doc, "t-tenant-a-vet")["status"] == "indexed" + assert db_mod.get_document_index_status(lifecycle_indexed_doc, "old-index")["status"] == "indexed" -def test_query_off_does_not_mark_unpurged_historical_index( +def test_query_off_502_leaves_historical_index_indexed( lifecycle_indexed_doc, monkeypatch ): db_mod.upsert_document_index_status( @@ -614,12 +646,12 @@ def test_query_off_does_not_mark_unpurged_historical_index( lifecycle_indexed_doc, "old-index", status="indexed", chunk_count_indexed=2 ) - def _fake_delete(doc_id, index_name="documents-index", workflow_id=None): + def _fake_set(document_id, index_name, workflow_id, enabled, chunk_num=None): if index_name == "old-index": - return {"deleted": 0, "error": "old index down"} - return {"deleted": 2, "index_name": index_name} + return {"updated": 0, "error": "old index down"} + return {"updated": 2, "index_name": index_name} - monkeypatch.setattr(indexes, "delete_chunks_from_marqo", _fake_delete) + monkeypatch.setattr(indexes, "set_document_chunks_query_enabled", _fake_set) with pytest.raises(HTTPException) as exc: _run( documents.set_document_query_enabled( diff --git a/tests/test_marqo_doc_scope.py b/tests/test_marqo_doc_scope.py index 8c5a5a6..0d07ab5 100644 --- a/tests/test_marqo_doc_scope.py +++ b/tests/test_marqo_doc_scope.py @@ -71,7 +71,7 @@ def get_settings(self): "allFields": [{"name": name, "type": "text"} for name in sorted(self.fields)] } - def search(self, q="", filter_string="", limit=10, attributes_to_retrieve=None): + def search(self, q="", filter_string="", limit=10, offset=0, attributes_to_retrieve=None): self.searches.append(filter_string) if attributes_to_retrieve and "_id" in attributes_to_retrieve: # A structured legacy index rejects `_id` here (prod hotfix #55). @@ -92,11 +92,20 @@ def search(self, q="", filter_string="", limit=10, attributes_to_retrieve=None): ) ] keep = list(attributes_to_retrieve or []) + ["_id"] - return {"hits": [{k: hit[k] for k in keep if k in hit} for hit in hits[:limit]]} + page = hits[offset:offset + limit] + return {"hits": [{k: hit[k] for k in keep if k in hit} for hit in page]} def delete_documents(self, ids): self.records[:] = [r for r in self.records if r["_id"] not in set(ids)] + def update_documents(self, records): + by_id = {r["_id"]: r for r in self.records} + for rec in records: + existing = by_id.get(rec.get("_id")) + if existing is not None: + existing.update({k: v for k, v in rec.items() if k != "_id"}) + return {"updated": len(records)} + def _install(monkeypatch, index: _FakeIndex): fake = type( @@ -180,11 +189,12 @@ def test_disabling_one_document_does_not_purge_its_alias( documents.disable_document("wf-second", _admin_in("tenant-a"), remove_from_search=True) ) - assert result["marqo_deleted"] == 2, "purged more than the document being disabled" - surviving = sorted(r["workflow_id"] for r in index.records) - assert surviving == ["wf-first"] * 3, ( - "disabling one document deleted its alias's vectors" - ) + assert result["marqo_updated"] == 2 + assert len(index.records) == 5, "disable must not delete search records" + second = [r for r in index.records if r["workflow_id"] == "wf-second"] + first = [r for r in index.records if r["workflow_id"] == "wf-first"] + assert all(r.get("query_enabled") is False for r in second) + assert all(r.get("query_enabled") is not False for r in first) def test_query_disable_purge_is_scoped_to_one_document( @@ -204,10 +214,11 @@ def test_query_disable_purge_is_scoped_to_one_document( ) ) - surviving = sorted(r["workflow_id"] for r in index.records) - assert surviving == ["wf-second"] * 2, ( - "turning off queries for one document deleted its alias's vectors" - ) + assert len(index.records) == 5 + first = [r for r in index.records if r["workflow_id"] == "wf-first"] + second = [r for r in index.records if r["workflow_id"] == "wf-second"] + assert all(r.get("query_enabled") is False for r in first) + assert all(r.get("query_enabled") is not False for r in second) def test_single_chunk_purge_is_scoped_to_one_document(monkeypatch): diff --git a/tests/test_search_service.py b/tests/test_search_service.py index 3239445..cbaf5f7 100644 --- a/tests/test_search_service.py +++ b/tests/test_search_service.py @@ -68,6 +68,7 @@ def test_run_search_happy_path_calls_store_and_caps_per_doc(): call_args = store.search.call_args assert call_args.args[0] == "documents-index" assert call_args.kwargs["q"].startswith("query:") + assert "query_enabled:true" in (call_args.kwargs.get("filter_string") or "") assert result["final_count"] == 3 # 2 from d1 + 1 from d2 assert [h["doc_id"] for h in result["hits"]] == ["d1", "d1", "d2"] assert result["candidate_count"] == 4 diff --git a/tests/test_tenant_isolation.py b/tests/test_tenant_isolation.py index 292e918..95eeb25 100644 --- a/tests/test_tenant_isolation.py +++ b/tests/test_tenant_isolation.py @@ -145,6 +145,14 @@ def delete_documents(self, ids): _INDEX_HITS[self.name] = [h for h in current if h.get("_id") not in remove] return {"deleted": len(remove)} + def update_documents(self, records): + by_id = {hit.get("_id"): hit for hit in _INDEX_HITS.get(self.name, [])} + for rec in records: + existing = by_id.get(rec.get("_id")) + if existing is not None: + existing.update({k: v for k, v in rec.items() if k != "_id"}) + return {"updated": len(records)} + class _FakeClient: def __init__(self, url=None, **kwargs): @@ -846,8 +854,8 @@ def test_upload_create_instance_requires_upload_in_that_tenant(seeded): # --- Fix 5: doc-delete resolves the doc's OWN tenant index -------------------- -def test_disable_document_deletes_from_own_tenant_index_not_legacy(seeded, marqo_stub): - """Deleting document WF_A's chunks must target tenant-A's index, never the +def test_disable_document_hides_in_own_tenant_index_not_legacy(seeded, marqo_stub): + """Hiding document WF_A's chunks must target tenant-A's index, never the legacy/default ``documents-index`` (which holds the DEFAULT tenant's records).""" marqo_stub["t-tenant-a-vet"] = [{"_id": "a1", "doc_id": "d-a", "instance": A, "text": "a"}] # A decoy in the legacy/default index that must remain untouched. @@ -855,11 +863,11 @@ def test_disable_document_deletes_from_own_tenant_index_not_legacy(seeded, marqo res = _run(document_routes.disable_document(WF_A, _tenant_admin_in(A), remove_from_search=True)) - # tenant-A's own index had its chunk removed... - assert marqo_stub["t-tenant-a-vet"] == [] - assert res["marqo_deleted"] == 1 - # ...and the legacy/default index was never touched. + assert len(marqo_stub["t-tenant-a-vet"]) == 1 + assert marqo_stub["t-tenant-a-vet"][0].get("query_enabled") is False + assert res["marqo_updated"] == 1 assert len(marqo_stub["documents-index"]) == 1 + assert marqo_stub["documents-index"][0].get("query_enabled") is not False searched = {name for name, _ in _SEARCH_CALLS} assert "t-tenant-a-vet" in searched assert "documents-index" not in searched diff --git a/tests/test_vector_store.py b/tests/test_vector_store.py index 82c8d8c..f32dcb1 100644 --- a/tests/test_vector_store.py +++ b/tests/test_vector_store.py @@ -596,3 +596,11 @@ def __init__(self, url=None, **_kwargs): with pytest.raises(ValueError): get_vector_store(client_factory=lambda: None, url="http://x") + + +def test_passage_schema_declares_query_enabled_filter(): + settings = vector_store_mod.passage_index_settings() + fields = {item["name"]: item for item in settings["allFields"]} + assert fields["query_enabled"]["type"] == "bool" + assert "filter" in fields["query_enabled"]["features"] + assert vector_store_mod.QUERY_ENABLED_FILTER == "query_enabled:true" diff --git a/ui/src/views/DocumentOpsView.jsx b/ui/src/views/DocumentOpsView.jsx index abb3931..6276d1d 100644 --- a/ui/src/views/DocumentOpsView.jsx +++ b/ui/src/views/DocumentOpsView.jsx @@ -336,8 +336,8 @@ export default function DocumentOpsView() { try { const result = await fetchJson(`/documents/${workflowId}?remove_from_search=true`, { method: 'DELETE' }) const excluded = result?.chunks_excluded ?? 0 - const marqo = result?.marqo_deleted ?? 0 - setStatusMessage(`Document deleted. ${excluded} chunk(s) off for queries; ${marqo} removed from Marqo. Restore later, then Include + Reingest to republish.`) + const marqo = result?.marqo_updated ?? result?.marqo_deleted ?? 0 + setStatusMessage(`Document deleted. ${excluded} chunk(s) off for queries; ${marqo} hidden in search. Restore, then Include to show them again.`) setShowDeleteConfirm(false) await load() } catch (error) { @@ -352,7 +352,7 @@ export default function DocumentOpsView() { clearStatus() try { await fetchJson(`/documents/${workflowId}/restore`, { method: 'POST' }) - setStatusMessage('Document restored to the list. Still off for queries — turn Include on and reingest to republish.') + setStatusMessage('Document restored to the list. Still off for queries — turn Include on to show them in search.') await load() } catch (error) { setStatusMessage(error.message, 'error') @@ -372,8 +372,8 @@ export default function DocumentOpsView() { }) setStatusMessage( enabled - ? 'Document included for queries (all chunks included). Reingest to republish to Marqo.' - : 'Document excluded from queries — all chunks off and removed from Marqo.' + ? 'Document included for queries (all chunks included and visible in search).' + : 'Document excluded from queries — all chunks off and hidden in search.' ) await load() } catch (error) { @@ -392,7 +392,7 @@ export default function DocumentOpsView() { headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ is_excluded: excluded }), }) - setStatusMessage(excluded ? `Chunk ${chunkNumber} excluded from queries.` : `Chunk ${chunkNumber} included for queries (reingest to republish).`) + setStatusMessage(excluded ? `Chunk ${chunkNumber} excluded from queries.` : `Chunk ${chunkNumber} included for queries.`) await load() } catch (error) { setStatusMessage(error.message, 'error') @@ -737,7 +737,7 @@ export default function DocumentOpsView() {

Include & delete

- Include works like chunk Include: off cascades to every chunk and clears Marqo. Delete hides the doc and fully removes it from Marqo — restore + Include + Reingest to bring search back. + Include works like chunk Include: off cascades to every chunk and hides them in search. Delete hides the doc the same way — restore + Include to bring search back.

@@ -1499,8 +1499,8 @@ export default function DocumentOpsView() { Delete this document? - Soft-hides the document, turns Include off on every chunk, and fully removes those chunks from Marqo. - Restore brings it back to the list only — turn Include on and reingest to republish search. + Soft-hides the document, turns Include off on every chunk, and hides those chunks in search. + Restore brings it back to the list only — turn Include on to show them in search again. From 0f8dfbb5db40e714b0799aeb8afbd035bd8003c1 Mon Sep 17 00:00:00 2001 From: Aayush Date: Mon, 31 Aug 2026 16:44:02 +0530 Subject: [PATCH 2/4] fix(#56): fail closed on Marqo query_enabled updates Count only confirmed item successes, restore earlier indexes when a later one fails, and 409 when the index schema has no query_enabled field. SQLite is not flipped unless every target index succeeds. --- docs/api-contracts.md | 14 +++++- pipeline/services/indexes.py | 68 ++++++++++++++++++++++++++- pipeline/vector_store.py | 59 +++++++++++++++++++++-- tests/test_document_lifecycle.py | 81 ++++++++++++++++++++++++++++++++ tests/test_marqo_doc_scope.py | 7 ++- tests/test_vector_store.py | 33 +++++++++++++ 6 files changed, 251 insertions(+), 11 deletions(-) diff --git a/docs/api-contracts.md b/docs/api-contracts.md index 28578b4..a68f4e3 100644 --- a/docs/api-contracts.md +++ b/docs/api-contracts.md @@ -371,7 +371,11 @@ Permission: `review`. **Exclude semantics** - Setting `is_excluded` when the document `stage` is `completed` flips - `query_enabled` on that Marqo record (no delete, no reingest) + `query_enabled` on that Marqo record (no delete, no reingest). Permission: + `review` (the only gate on the `is_excluded` column). +- Returns **409** if a target index has no `query_enabled` field (never falls + back to delete). **502** if a Marqo item update fails; earlier indexes in + the same call are restored and SQLite is not changed. - Text or tag edits still mark reindex / dirty for later republish ### `PUT /documents/{workflow_id}/chunks/{chunk_num}/tags` @@ -486,7 +490,7 @@ Soft-delete document. Permission: `admin`. **Effects** 1. Cancel running Temporal workflow if possible -2. Optionally hide chunks in Marqo via `query_enabled:false` (fail-closed: disable is not flipped if this fails). Every recorded `document_index_status` index plus the resolved physical index is updated. Records are not deleted. +2. Optionally hide chunks in Marqo via `query_enabled:false` (fail-closed: disable is not flipped if this fails). Every recorded `document_index_status` index plus the resolved physical index is updated. Records are not deleted. **409** if an index lacks `query_enabled`; **502** if an item update fails (already-flipped indexes in this call are restored). 3. Set `is_disabled=true` in SQLite, turn queries off, exclude chunks 4. Report artifact GC plan; apply MinIO deletes only if `purge_artifacts=true` @@ -521,6 +525,12 @@ with `purged_at` set. CLI equivalent: `scripts/purge_document_artifacts.py` Clear `is_disabled` only. Chunks stay excluded until Include is turned on (flag flip; no reingest). Permission: `admin`. +### `POST /documents/{workflow_id}/query-enabled` + +Document Include. Permission: `admin` (lifecycle; not a second chunk-exclude +gate). Body: `{ "query_enabled": true|false }`. Same Marqo flag-flip and +fail-closed 409/502 rules as disable. Does not set `is_disabled`. + ```json { "workflow_id": "…", diff --git a/pipeline/services/indexes.py b/pipeline/services/indexes.py index 6a2e275..6c382d7 100644 --- a/pipeline/services/indexes.py +++ b/pipeline/services/indexes.py @@ -201,6 +201,18 @@ def set_document_chunks_query_enabled( ) -> dict: """Flip ``query_enabled`` on one document's records (or one chunk). No delete.""" store = vector_store.get_vector_store() + try: + field_names = store.field_names(index_name) + except vector_store.VectorStoreError as exc: + if vector_store.index_missing_error(exc): + return {"updated": 0, "reason": "index_missing"} + return {"updated": 0, "error": str(exc)} + if "query_enabled" not in field_names: + return { + "updated": 0, + "reason": "missing_query_enabled_field", + "index_name": index_name, + } scope = vector_store.marqo_doc_scope_filter(document_id, workflow_id) if chunk_num is not None: scope = vector_store.merge_filter_strings( @@ -222,10 +234,44 @@ def set_document_chunks_query_enabled( result = store.set_query_enabled(index_name, ids, enabled) except vector_store.VectorStoreError as exc: return {"updated": 0, "error": str(exc)} + failed = result.get("failed") or [] + if failed: + succeeded_ids = list(result.get("succeeded_ids") or []) + if succeeded_ids: + try: + store.set_query_enabled(index_name, succeeded_ids, not enabled) + except vector_store.VectorStoreError: + logging.getLogger(__name__).error( + "Failed to revert partial query_enabled update on %s", index_name + ) + return { + "updated": 0, + "error": f"{len(failed)} record(s) failed to update", + "failed": failed, + "index_name": index_name, + } result["index_name"] = index_name return result +def _restore_query_enabled_indexes( + document_id: str, + workflow_id: str, + index_names: list[str], + enabled: bool, + chunk_num: Optional[int], +) -> None: + """Best-effort undo of indexes already flipped in this apply call.""" + for index_name in reversed(index_names): + undo = set_document_chunks_query_enabled( + document_id, index_name, workflow_id, not enabled, chunk_num=chunk_num + ) + if undo.get("error") or undo.get("failed"): + logging.getLogger(__name__).error( + "query_enabled rollback failed on %s: %s", index_name, undo + ) + + def apply_document_query_enabled( doc: dict, workflow_id: str, @@ -247,14 +293,32 @@ def apply_document_query_enabled( if not names: return 0 updated = 0 + flipped: list[str] = [] for index_name in sorted(names): result = set_document_chunks_query_enabled( doc_id, index_name, workflow_id, enabled, chunk_num=chunk_num ) - if result.get("error"): + if result.get("reason") == "index_missing": + continue + if result.get("reason") == "missing_query_enabled_field": + _restore_query_enabled_indexes( + doc_id, workflow_id, flipped, enabled, chunk_num + ) + raise HTTPException( + 409, + f"Soft-disable is unavailable on index '{index_name}': " + "query_enabled is not in the schema. Recreate the index with " + "the passage schema; records were not deleted.", + ) + if result.get("error") or result.get("failed"): + _restore_query_enabled_indexes( + doc_id, workflow_id, flipped, enabled, chunk_num + ) + detail = result.get("error") or "record update failed" raise HTTPException( 502, - f"Failed to update search visibility ({index_name}): {result['error']}", + f"Failed to update search visibility ({index_name}): {detail}", ) + flipped.append(index_name) updated += int(result.get("updated") or 0) return updated diff --git a/pipeline/vector_store.py b/pipeline/vector_store.py index c790061..9f087cc 100644 --- a/pipeline/vector_store.py +++ b/pipeline/vector_store.py @@ -378,6 +378,38 @@ def field_names_from_settings(settings: Any) -> set[str]: } +def _unwrap_item_failures(result: Any, requested_ids: Sequence[str]) -> list[dict]: + """Per-item failures from a Marqo add/update payload (200 + ``errors: true``).""" + if not isinstance(result, dict) or not result.get("errors"): + return [] + items = list(result.get("items") or []) + if not items: + return [ + { + "_id": rid, + "status": None, + "error": "update reported errors with no per-item details", + "message": None, + "code": None, + } + for rid in requested_ids + ] + failures = [] + for item in items: + if item.get("status") == 200: + continue + failures.append( + { + "_id": item.get("_id"), + "status": item.get("status"), + "error": item.get("error"), + "message": item.get("message"), + "code": item.get("code"), + } + ) + return failures + + def index_missing_error(err: Exception | str) -> bool: """True when Marqo has no such index — equivalent to zero searchable chunks. @@ -751,7 +783,11 @@ def update_documents(self, index: str, records: Sequence[dict]) -> Any: def set_query_enabled( self, index: str, record_ids: Sequence[str], enabled: bool ) -> dict: - """Flip ``query_enabled`` on existing records. No re-embed.""" + """Flip ``query_enabled`` on existing records. No re-embed. + + Returns ``updated`` (confirmed successes) and ``failed`` (per-item + errors). A Marqo 200 with ``errors: true`` is not treated as success. + """ ... def delete_document( @@ -937,15 +973,28 @@ def update_documents(self, index: str, records: Sequence[dict]) -> Any: def set_query_enabled( self, index: str, record_ids: Sequence[str], enabled: bool ) -> dict: - """Partial-update ``query_enabled`` on existing records. No re-embed.""" + """Partial-update ``query_enabled`` on existing records. No re-embed. + + Confirmed successes are ``updated``; per-item Marqo failures are + ``failed``. A 200 with ``errors: true`` never counts as a full success. + """ ids = [rid for rid in record_ids if rid] if not ids: - return {"updated": 0} - self.update_documents( + return {"updated": 0, "failed": [], "succeeded_ids": []} + result = self.update_documents( index, [{"_id": rid, "query_enabled": bool(enabled)} for rid in ids], ) - return {"updated": len(ids)} + failed = _unwrap_item_failures(result, ids) + failed_ids = {item.get("_id") for item in failed if item.get("_id")} + succeeded_ids = [rid for rid in ids if rid not in failed_ids] + if failed and not any(item.get("_id") for item in failed): + succeeded_ids = [] + return { + "updated": len(succeeded_ids), + "failed": failed, + "succeeded_ids": succeeded_ids, + } # -- purges -------------------------------------------------------------- # diff --git a/tests/test_document_lifecycle.py b/tests/test_document_lifecycle.py index 58e6949..4fd712f 100644 --- a/tests/test_document_lifecycle.py +++ b/tests/test_document_lifecycle.py @@ -665,3 +665,84 @@ def _fake_set(document_id, index_name, workflow_id, enabled, chunk_num=None): assert int(row["query_enabled"]) == 1 assert db_mod.get_document_index_status(lifecycle_indexed_doc, "old-index")["status"] == "indexed" + +def test_later_index_failure_restores_earlier_index(lifecycle_indexed_doc, monkeypatch): + db_mod.upsert_document_index_status( + lifecycle_indexed_doc, "aaa-index", status="indexed", chunk_count_indexed=2 + ) + db_mod.upsert_document_index_status( + lifecycle_indexed_doc, "zzz-index", status="indexed", chunk_count_indexed=2 + ) + calls: list[tuple[str, bool]] = [] + + def _fake_set(document_id, index_name, workflow_id, enabled, chunk_num=None): + calls.append((index_name, enabled)) + if index_name == "zzz-index" and enabled is False: + return {"updated": 0, "error": "zzz down"} + return {"updated": 2, "index_name": index_name} + + monkeypatch.setattr(indexes, "set_document_chunks_query_enabled", _fake_set) + with pytest.raises(HTTPException) as exc: + _run( + documents.set_document_query_enabled( + lifecycle_indexed_doc, + DocumentQueryEnabledUpdate(query_enabled=False), + _admin_in("tenant-a"), + ) + ) + assert exc.value.status_code == 502 + assert ("aaa-index", False) in calls + assert ("zzz-index", False) in calls + assert ("aaa-index", True) in calls + assert int(db_mod.get_document(lifecycle_indexed_doc)["query_enabled"]) == 1 + + +def test_query_enabled_409_when_index_lacks_field(lifecycle_indexed_doc, monkeypatch): + db_mod.upsert_document_index_status( + lifecycle_indexed_doc, "aaa-ok", status="indexed", chunk_count_indexed=2 + ) + db_mod.upsert_document_index_status( + lifecycle_indexed_doc, "zzz-legacy", status="indexed", chunk_count_indexed=2 + ) + calls: list[tuple[str, bool]] = [] + + def _fake_set(document_id, index_name, workflow_id, enabled, chunk_num=None): + calls.append((index_name, enabled)) + if index_name == "zzz-legacy": + return { + "updated": 0, + "reason": "missing_query_enabled_field", + "index_name": index_name, + } + return {"updated": 2, "index_name": index_name} + + monkeypatch.setattr(indexes, "set_document_chunks_query_enabled", _fake_set) + with pytest.raises(HTTPException) as exc: + _run( + documents.set_document_query_enabled( + lifecycle_indexed_doc, + DocumentQueryEnabledUpdate(query_enabled=False), + _admin_in("tenant-a"), + ) + ) + assert exc.value.status_code == 409 + assert "zzz-legacy" in str(exc.value.detail) + assert "query_enabled" in str(exc.value.detail).lower() + assert ("aaa-ok", False) in calls + assert ("aaa-ok", True) in calls + assert int(db_mod.get_document(lifecycle_indexed_doc)["query_enabled"]) == 1 + + +def test_set_chunks_reports_missing_query_enabled_field(monkeypatch): + class _NoFlagStore: + def field_names(self, index): + return {"doc_id", "workflow_id", "chunk_num"} + + monkeypatch.setattr(indexes.vector_store, "get_vector_store", lambda: _NoFlagStore()) + result = indexes.set_document_chunks_query_enabled( + "doc-1", "legacy-index", "wf-1", False + ) + assert result["reason"] == "missing_query_enabled_field" + assert result["index_name"] == "legacy-index" + assert result["updated"] == 0 + diff --git a/tests/test_marqo_doc_scope.py b/tests/test_marqo_doc_scope.py index 0d07ab5..e9d4e4c 100644 --- a/tests/test_marqo_doc_scope.py +++ b/tests/test_marqo_doc_scope.py @@ -62,7 +62,7 @@ def __init__(self, records: list[dict], fields: set[str] | None = None): self.fields = ( fields if fields is not None - else {"doc_id", "workflow_id", "chunk_num", "instance"} + else {"doc_id", "workflow_id", "chunk_num", "instance", "query_enabled"} ) self.searches: list[str] = [] @@ -104,7 +104,10 @@ def update_documents(self, records): existing = by_id.get(rec.get("_id")) if existing is not None: existing.update({k: v for k, v in rec.items() if k != "_id"}) - return {"updated": len(records)} + return { + "errors": False, + "items": [{"_id": rec.get("_id"), "status": 200} for rec in records], + } def _install(monkeypatch, index: _FakeIndex): diff --git a/tests/test_vector_store.py b/tests/test_vector_store.py index f32dcb1..f9628eb 100644 --- a/tests/test_vector_store.py +++ b/tests/test_vector_store.py @@ -92,6 +92,9 @@ def delete_documents(self, ids): self.deleted.append(list(ids)) self.records[:] = [r for r in self.records if r["_id"] not in set(ids)] + def update_documents(self, records): + return {"errors": False, "items": [{"_id": rec.get("_id"), "status": 200} for rec in records]} + def _install(monkeypatch, index: _FakeIndex, *, exists: bool = False) -> _FakeIndex: """Patch the ``marqo`` module so the store's lazy import picks up the fake. @@ -604,3 +607,33 @@ def test_passage_schema_declares_query_enabled_filter(): assert fields["query_enabled"]["type"] == "bool" assert "filter" in fields["query_enabled"]["features"] assert vector_store_mod.QUERY_ENABLED_FILTER == "query_enabled:true" + + +def test_set_query_enabled_counts_only_confirmed_item_successes(monkeypatch): + class _Partial(_FakeIndex): + def update_documents(self, records): + return { + "errors": True, + "items": [ + {"_id": "a", "status": 200}, + {"_id": "b", "status": 400, "error": "bad field", "code": "x"}, + ], + } + + _install(monkeypatch, _Partial([]), exists=True) + result = MarqoStore().set_query_enabled(TENANT_INDEX, ["a", "b"], False) + assert result["updated"] == 1 + assert result["succeeded_ids"] == ["a"] + assert [item["_id"] for item in result["failed"]] == ["b"] + + +def test_set_query_enabled_errors_flag_without_items_fails_all(monkeypatch): + class _Opaque(_FakeIndex): + def update_documents(self, records): + return {"errors": True} + + _install(monkeypatch, _Opaque([]), exists=True) + result = MarqoStore().set_query_enabled(TENANT_INDEX, ["a", "b"], True) + assert result["updated"] == 0 + assert result["succeeded_ids"] == [] + assert {item["_id"] for item in result["failed"]} == {"a", "b"} From 459bec543c97d0eadde9c66885d00645f292379e Mon Sep 17 00:00:00 2001 From: Aayush Date: Tue, 1 Sep 2026 15:20:00 +0530 Subject: [PATCH 3/4] test(#56): advertise query_enabled on non-legacy isolation fake Disable/include now 409s when the index schema has no query_enabled. The tenant-isolation fake only advertised instance, so hide-in-own-index never reached the isolation assertion. Legacy fakes still omit the field. --- tests/test_tenant_isolation.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/tests/test_tenant_isolation.py b/tests/test_tenant_isolation.py index 95eeb25..d8390d9 100644 --- a/tests/test_tenant_isolation.py +++ b/tests/test_tenant_isolation.py @@ -82,9 +82,10 @@ def _viewer_in(instance: str): # physical indexes that "exist" in this fake Marqo (realistic get_index semantics: # creating a brand-new index name must not report it as pre-existing). _EXISTING_INDEXES: set[str] = set() -# physical indexes that DON'T advertise the filterable `instance` field — i.e. the -# legacy single-tenant index. Restricted search must fail closed on these -# (match nothing) unless ALLOW_UNSCOPED_LEGACY_SEARCH is on. +# physical indexes that DON'T advertise filterable `instance` / `query_enabled` +# — i.e. the legacy single-tenant index. Restricted search must fail closed on +# these (match nothing) unless ALLOW_UNSCOPED_LEGACY_SEARCH is on. Hide/include +# must 409 rather than fall back to delete. _LEGACY_INDEXES: set[str] = set() # records of (physical_index_name, search_kwargs) for assertions _SEARCH_CALLS: list[tuple] = [] @@ -106,9 +107,10 @@ def __init__(self, name): self.name = name def get_settings(self): - # Advertise the filterable ``instance`` field so the per-chunk tenant - # filter engages for restricted callers — UNLESS this index is registered - # as legacy (no ``instance`` field). Restricted search then fail-closes. + # Advertise filterable ``instance`` and ``query_enabled`` so tenant + # isolation and hide-via-flag paths engage — UNLESS this index is + # registered as legacy (neither field). Restricted search then + # fail-closes; disable/include 409s instead of deleting. fields = [ {"name": "text"}, {"name": "domain_tags"}, @@ -116,6 +118,7 @@ def get_settings(self): ] if self.name not in _LEGACY_INDEXES: fields.insert(0, {"name": "instance"}) + fields.append({"name": "query_enabled"}) return {"allFields": fields} def get_stats(self): From 736e1354ff621e4924bbc6ed46a655694935a6a2 Mon Sep 17 00:00:00 2001 From: Aayush Date: Tue, 1 Sep 2026 15:46:51 +0530 Subject: [PATCH 4/4] test(#56): Marqo-shaped isolation fake and legacy 409 Non-legacy hide now goes through the same update_documents item payload as production. A tenant index without query_enabled 409s and does not delete or disable. --- tests/test_tenant_isolation.py | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/tests/test_tenant_isolation.py b/tests/test_tenant_isolation.py index ed4c8ed..f8d3c31 100644 --- a/tests/test_tenant_isolation.py +++ b/tests/test_tenant_isolation.py @@ -150,11 +150,13 @@ def delete_documents(self, ids): def update_documents(self, records): by_id = {hit.get("_id"): hit for hit in _INDEX_HITS.get(self.name, [])} + items = [] for rec in records: existing = by_id.get(rec.get("_id")) if existing is not None: existing.update({k: v for k, v in rec.items() if k != "_id"}) - return {"updated": len(records)} + items.append({"_id": rec.get("_id"), "status": 200}) + return {"errors": False, "items": items} class _FakeClient: @@ -883,6 +885,22 @@ def test_disable_document_hides_in_own_tenant_index_not_legacy(seeded, marqo_stu assert "documents-index" not in searched +def test_disable_document_409s_when_tenant_index_lacks_query_enabled(seeded, marqo_stub): + """Legacy schema (no query_enabled) must 409, never fall back to delete.""" + marqo_stub["t-tenant-a-vet"] = [{"_id": "a1", "doc_id": "d-a", "instance": A, "text": "a"}] + _LEGACY_INDEXES.add("t-tenant-a-vet") + + with pytest.raises(HTTPException) as exc: + _run(document_routes.disable_document(WF_A, _tenant_admin_in(A), remove_from_search=True)) + + assert _status(exc) == 409 + assert "query_enabled" in str(exc.value.detail).lower() + assert len(marqo_stub["t-tenant-a-vet"]) == 1 + assert marqo_stub["t-tenant-a-vet"][0].get("query_enabled") is not False + row = db_mod.get_document(WF_A) + assert int(row["is_disabled"] or 0) == 0 + + # --- Fix 6: deleted-doc (orphan) audit rows are not leaked to the default tenant