OpenAgriNet · nexus69420 · Aug 30, 2026 · Aug 31, 2026 · Sep 1, 2026 · Sep 1, 2026
diff --git a/docs/api-contracts.md b/docs/api-contracts.md
@@ -388,9 +388,13 @@ 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). 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`
 
@@ -498,13 +502,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. **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`
 
@@ -513,8 +517,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).
 
 ---
 
@@ -535,8 +540,14 @@ 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`.
+
+### `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
 {

diff --git 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
@@ -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
@@ -3067,13 +3068,28 @@ 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(
                 "SELECT COALESCE(MAX(chunk_version), 0) AS max_version FROM chunks WHERE workflow_id = ?",
                 (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(
@@ -3127,6 +3143,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
@@ -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
@@ -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
@@ -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
@@ -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.")