feat(celery Wave 5 P4): production robustness 3-pack — cleanup transient/intentional split + parse_version short-circuit + reconciler stuck-parse re-enqueue

Wave 5 Phase 4 (PR-C / task #29) — closes 3 latent issues surfaced
through Wave 4 ratify trail:

* **P4-1 cleanup transient-vs-intentional split**
  (per architect msg=6aa8ca88 T2 ratify minor obs A):
  Pre-Wave-5 ``_resolve_cleanup_worker`` collapsed any factory
  exception into "drop the DB row". Transient infra errors
  (Qdrant blip, ES network glitch) lost the retry signal — DB row
  was deleted before next cycle could retry. Wave 5 P4 returns a
  new ``CleanupWorkerResolution(worker, transient)`` distinguishing
  the two: ``WorkerFactoryError`` is a by-design gate (drop the
  row to bound index growth), any other Exception is transient
  (skip DB row drop so next cycle retries when backend recovers).
  Counts surface ``transient_deferred`` so operators can track
  recovery rate.

* **P4-2 parse_version short-circuit**
  (per huangheng T3 chunk 2 obs B + Wave 5 backlog item):
  Pre-Wave-5 ``parse_document`` always re-runs DocParser even when
  the resulting artifact directory is byte-identical (parse_version
  is content-derived). Wave 5 P4 adds ``short_circuit_if_artifacts_exist=True``
  default — if all three derived artifacts (markdown.md /
  outline.json / chunks.jsonl) already exist under the canonical
  ``derived/parse_<version>/`` path, skip DocParser + writes
  entirely. Eliminates the ~30s OCR / Word rerun cost on rebuild
  of unchanged content. Tests can pass ``False`` to force re-parse.

* **P4-3 reconciler stuck-document parse re-enqueue**
  (per architect Wave 4 T3 chunk 2 obs A — production gap close):
  Pre-Wave-5 parse failures (DocParser raise / source missing)
  silently dropped the document_id in the parse worker; operator
  saw ``document.status == PENDING`` forever with no signal to
  re-trigger. Wave 5 P4 adds
  ``reconcile_stuck_documents_for_parse_reenqueue`` to the
  reconciler loop. Detects documents with
  ``Document.gmt_created < now - cooldown_seconds`` AND zero
  ``document_index`` rows AND ``Document.gmt_updated < now -
  cooldown_seconds`` (cooldown filter prevents 30-s tick storm),
  then pushes a fresh ``ParseDispatchPayload`` matching the upload
  handler's contract. ``gmt_updated`` bumps after each push so
  the cooldown predicate rate-limits re-enqueue.

Three-class tag (Wave 3 production-readiness invariant):
* must-be-real: cleanup transient/intentional split prevents
  silent retry-signal loss; parse_version short-circuit eliminates
  OCR rerun waste; stuck-parse reconciler closes the document.status
  PENDING-forever gap
* may-be-gated: ``short_circuit_if_artifacts_exist=False`` for
  tests pinning DocParser invocation count
* fully-resolves: 3 of Wave 5 P1 backlog items (per architect
  ratify trail msg=6aa8ca88 + huangheng obs trail)

Deltas:
* ``aperag/indexing/cleanup.py`` — ``CleanupWorkerResolution``
  dataclass + WorkerFactoryError-vs-Exception split in
  ``_resolve_cleanup_worker``; both ``cleanup_orphan_parse_versions``
  + ``cleanup_for_deleted_documents`` honor ``resolution.transient``
  to skip DB row drop on transient infra failures; collection
  cascade aggregates ``transient_deferred``.
* ``aperag/indexing/parser.py`` — ``_all_artifacts_present``
  predicate + ``short_circuit_if_artifacts_exist`` parameter +
  early-return ParseResult when all 3 artifacts present.
* ``aperag/indexing/reconciler.py`` — ``STUCK_PARSE_COOLDOWN_SECONDS``
  + ``reconcile_stuck_documents_for_parse_reenqueue`` async scan +
  ``_select_stuck_documents_for_reenqueue`` SQL query +
  ``_build_parse_payload_for_document`` payload reconstruction +
  ``_resolve_collection_parser_config`` / ``_resolve_collection_modalities``
  (mirror ``document_service`` shape) + ``_mark_stuck_documents_reenqueued``
  bumps gmt_updated; ``run_reconcile_loop`` calls the new scan.
* ``aperag/indexing/__init__.py`` — re-export new symbols.
* ``tests/integration/test_p4_robustness_3pack.py`` (new) — 13
  tests across 3 layers (cleanup transient/intentional split / parse
  short-circuit / reconciler re-enqueue with cooldown + payload
  shape + skip-when-indexed + skip-when-no-object-path).
* ``tests/unit_test/indexing/test_t3_1_dispatcher_path_c.py`` —
  added ``transient_deferred: 0`` to expected empty-counts dict.

Local gates:
* ``pytest tests/unit_test/indexing/ tests/integration/`` — 232
  passed / 48 skipped (incl. 13 new ``test_p4_robustness_3pack``).
* ``ruff check aperag/ tests/integration/test_p4_robustness_3pack.py``
  — clean.
* ``ruff format`` — applied.

Branch: local ``chenyexuan/celery-wave5-p4`` based on main
``19d3d70`` (Wave 4 PR #1731 squash); will rebase onto
``bryce/celery-wave5`` once Bryce opens the Wave 5 draft PR.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: earayu

aperag/indexing/__init__.py

@@ -158,9 +158,11 @@
     HEARTBEAT_STALE_SECONDS,
     RECONCILE_BATCH_SIZE,
     RECONCILE_INTERVAL_SECONDS,
+    STUCK_PARSE_COOLDOWN_SECONDS,
     reconcile_failed_retry,
     reconcile_pending_dispatch,
     reconcile_running_reclaim,
+    reconcile_stuck_documents_for_parse_reenqueue,
     run_reconcile_loop,
 )
 from aperag.indexing.summary import (
@@ -276,13 +278,15 @@
     "process_one_parse_task",
     "run_parse_worker",
     "run_parse_worker_loop",
-    # Reconciler (T2.1)
+    # Reconciler (T2.1 + Wave 5 P4 stuck-parse re-enqueue)
     "RECONCILE_INTERVAL_SECONDS",
     "RECONCILE_BATCH_SIZE",
     "HEARTBEAT_STALE_SECONDS",
+    "STUCK_PARSE_COOLDOWN_SECONDS",
     "reconcile_pending_dispatch",
     "reconcile_failed_retry",
     "reconcile_running_reclaim",
+    "reconcile_stuck_documents_for_parse_reenqueue",
     "run_reconcile_loop",
     # Cleanup (T2.1 + T3.1 path C)
     "CLEANUP_INTERVAL_SECONDS",

aperag/indexing/cleanup.py

@@ -84,6 +84,7 @@
 
 import asyncio
 import logging
+from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import Any, Awaitable, Callable, Mapping, Optional
 
@@ -92,6 +93,7 @@
 
 from aperag.indexing.base import ModalityWorker
 from aperag.indexing.models import DocumentIndex, Modality
+from aperag.indexing.worker_factory import WorkerFactoryError
 
 logger = logging.getLogger(__name__)
 
@@ -164,45 +166,92 @@ def _is_graph_worker(worker: ModalityWorker) -> bool:
     return getattr(worker, "modality", None) is Modality.GRAPH
 
 
+@dataclass(frozen=True)
+class CleanupWorkerResolution:
+    """Wave 5 P4 T2 — outcome of :func:`_resolve_cleanup_worker`.
+
+    Distinguishes the two failure modes that pre-Wave-5 collapsed into
+    a single ``None`` return:
+
+    * **intentional gate** (``worker is None`` AND ``transient is False``)
+      — :class:`WorkerFactoryError` raised because the modality is
+      Wave-N-gated by design (graph extractor not wired / vision
+      multimodal not configured), the operator deliberately disabled
+      the modality via ``collection.config``, or the row's modality
+      string is unknown. The cleanup loop should drop the DB row so
+      the index does not grow unboundedly while the gate is active.
+
+    * **transient infrastructure error** (``worker is None`` AND
+      ``transient is True``) — DB connection blip / Qdrant
+      unreachable / ES unhealthy / Redis network glitch. The cleanup
+      loop must NOT drop the DB row — the next cycle (5 min later)
+      retries automatically once the backend recovers. Pre-Wave-5
+      this collapsed into the gate path and silently lost the retry
+      signal.
+
+    * **resolved worker** (``worker is not None``, ``transient ignored``)
+      — happy path; caller proceeds with backend cleanup.
+    """
+
+    worker: Optional[ModalityWorker]
+    transient: bool
+
+
 async def _resolve_cleanup_worker(
     *,
     workers: Optional[Mapping[Modality, ModalityWorker]],
     worker_factory: Optional[WorkerFactoryForRow],
     row: DocumentIndex,
-) -> Optional[ModalityWorker]:
+) -> CleanupWorkerResolution:
     """Resolve the cleanup worker for a row from factory or static map.
 
-    Wave 4 T2: production wires the factory (per-(collection, modality)
-    lazy materialisation); tests typically pass a pre-built ``workers``
-    mapping. When both are provided the factory wins — production
-    deployments override the legacy mapping with the per-row factory.
-
-    Returns ``None`` when neither source can supply a worker; the
-    caller logs + counts the row as ``backend_skipped``. A factory
-    that raises :class:`WorkerFactoryError` (Wave 4 vision multimodal
-    gate / partial config) also returns ``None`` so the cleanup cycle
-    drops the DB row even when the backend is unreachable, matching
-    the existing "skip backend, drop row" semantics.
+    Wave 4 T2 + Wave 5 P4 (transient-vs-intentional split): production
+    wires the factory (per-(collection, modality) lazy materialisation);
+    tests typically pass a pre-built ``workers`` mapping. When both
+    are provided the factory wins — production deployments override
+    the legacy mapping with the per-row factory.
+
+    Returns a :class:`CleanupWorkerResolution`:
+
+    * ``worker is not None``: backend cleanup proceeds.
+    * ``worker is None, transient=False``: intentional gate / unknown
+      modality / no source — caller drops DB row.
+    * ``worker is None, transient=True``: transient infrastructure
+      error — caller skips DB row drop so next cycle retries.
+
+    Pre-Wave-5 this returned ``None`` for both failure modes,
+    causing transient errors to silently lose their retry signal.
     """
     if worker_factory is not None:
         try:
-            return await worker_factory(row)
-        except Exception as exc:  # noqa: BLE001 — surface via log, count as skip
+            return CleanupWorkerResolution(worker=await worker_factory(row), transient=False)
+        except WorkerFactoryError as exc:
             logger.warning(
-                "cleanup worker_factory failed modality=%s row id=%d collection=%s: %s — counting as backend_skipped",
+                "cleanup worker_factory gate raised modality=%s row id=%d collection=%s: %s — "
+                "counting as backend_skipped (intentional gate, dropping DB row)",
                 row.modality,
                 row.id,
                 row.collection_id,
                 exc,
             )
-            return None
+            return CleanupWorkerResolution(worker=None, transient=False)
+        except Exception as exc:  # noqa: BLE001 — transient infra error, retry next cycle
+            logger.warning(
+                "cleanup worker_factory transient failure modality=%s row id=%d collection=%s: %s — "
+                "skipping DB row drop, will retry next cycle",
+                row.modality,
+                row.id,
+                row.collection_id,
+                exc,
+            )
+            return CleanupWorkerResolution(worker=None, transient=True)
     if workers is None:
-        return None
+        return CleanupWorkerResolution(worker=None, transient=False)
     try:
         modality = Modality(row.modality)
     except ValueError:
-        return None
-    return workers.get(modality)
+        return CleanupWorkerResolution(worker=None, transient=False)
+    return CleanupWorkerResolution(worker=workers.get(modality), transient=False)
 
 
 # ---------------------------------------------------------------------
@@ -301,6 +350,7 @@ async def cleanup_orphan_parse_versions(
         "rows_deleted": 0,
         "graph_noop": 0,
         "backend_skipped": 0,
+        "transient_deferred": 0,
     }
 
     delete_ids: list[int] = []
@@ -315,11 +365,18 @@ async def cleanup_orphan_parse_versions(
             )
             continue
 
-        worker = await _resolve_cleanup_worker(
+        resolution = await _resolve_cleanup_worker(
             workers=workers,
             worker_factory=worker_factory,
             row=row,
         )
+        if resolution.transient:
+            # Wave 5 P4: transient infra error — skip both backend
+            # delete AND DB row drop so the next cleanup cycle (5 min
+            # later) retries automatically once the backend recovers.
+            counts["transient_deferred"] += 1
+            continue
+        worker = resolution.worker
         if worker is None:
             logger.warning(
                 "cleanup no worker registered for modality=%s row id=%d — skipping backend delete",
@@ -424,6 +481,7 @@ async def cleanup_for_deleted_documents(
         "graph_lineage_cleaned": 0,
         "rows_deleted": 0,
         "backend_skipped": 0,
+        "transient_deferred": 0,
     }
 
     # Per-document, per-modality dedup so graph lineage cleanup runs
@@ -444,11 +502,18 @@ async def cleanup_for_deleted_documents(
             )
             continue
 
-        worker = await _resolve_cleanup_worker(
+        resolution = await _resolve_cleanup_worker(
             workers=workers,
             worker_factory=worker_factory,
             row=row,
         )
+        if resolution.transient:
+            # Wave 5 P4: transient infra error — skip backend delete
+            # AND DB row drop so the caller can retry on a later cycle
+            # / re-invocation once the backend recovers.
+            counts["transient_deferred"] += 1
+            continue
+        worker = resolution.worker
         if worker is None:
             logger.warning(
                 "cleanup no worker for modality=%s row id=%d document=%s — skipping backend",
@@ -623,6 +688,7 @@ async def cleanup_for_deleted_collections(
         "graph_lineage_cleaned": 0,
         "rows_deleted": 0,
         "backend_skipped": 0,
+        "transient_deferred": 0,
         "collections_cleaned": 0,
     }
     if not collection_ids:
@@ -641,7 +707,13 @@ async def cleanup_for_deleted_collections(
             worker_factory=worker_factory,
             document_ids=document_ids,
         )
-        for key in ("backend_deleted", "graph_lineage_cleaned", "rows_deleted", "backend_skipped"):
+        for key in (
+            "backend_deleted",
+            "graph_lineage_cleaned",
+            "rows_deleted",
+            "backend_skipped",
+            "transient_deferred",
+        ):
             counts[key] += sub_counts[key]
 
     # Sweep any rows that path B did not catch (no document_id match

aperag/indexing/parser.py

@@ -329,6 +329,29 @@ def _document_md5(source_bytes: bytes) -> str:
     return hashlib.md5(source_bytes).hexdigest()
 
 
+def _all_artifacts_present(
+    *,
+    store: _SyncObjectStore,
+    markdown_path: str,
+    outline_path: str,
+    chunks_path: str,
+) -> bool:
+    """Wave 5 P4 short-circuit predicate: all three canonical
+    derived artifacts must exist for the cached parse to be valid.
+
+    Uses ``ObjectStore.obj_exists`` (cheap metadata check) rather
+    than ``read_or_none`` so the predicate stays cost-bounded for
+    every parse call. ``chunks.jsonl`` is checked last because it is
+    the only artifact downstream modality workers actually read; if
+    it is missing the previous parse was interrupted mid-write and
+    re-parsing is required regardless.
+    """
+    try:
+        return store.obj_exists(markdown_path) and store.obj_exists(outline_path) and store.obj_exists(chunks_path)
+    except Exception:  # noqa: BLE001 — predicate fails closed (re-parse)
+        return False
+
+
 def _docparser_extract_markdown(
     *,
     source_bytes: bytes,
@@ -398,6 +421,7 @@ def parse_document(
     source_filename: str | None = None,
     parser_config: dict[str, Any] | None = None,
     config: ParseConfig | None = None,
+    short_circuit_if_artifacts_exist: bool = True,
 ) -> ParseResult:
     """Parse the source bytes and persist the three shared artifacts.
 
@@ -406,6 +430,16 @@ def parse_document(
     them atomically). The artifact paths are the canonical
     ``derived/parse_<version>/`` layout per design pack §C.1.
 
+    Wave 5 P4 short-circuit: when ``short_circuit_if_artifacts_exist``
+    is True (default) and all three derived artifacts (``markdown.md``
+    / ``outline.json`` / ``chunks.jsonl``) already exist in the object
+    store under the canonical ``derived/parse_<version>/`` path, the
+    parser **skips DocParser + writes entirely** and returns the
+    existing :class:`ParseResult`. This eliminates the ~30s OCR / Word
+    rerun cost when a document is re-uploaded with identical content
+    or a rebuild is dispatched against an already-parsed version
+    (per huangheng T3 chunk 2 obs B + architect Wave 5 P4 lock).
+
     Dispatch (Wave 4 T3 chunk 1):
     - ``source_filename`` ends in a known text extension (``.md`` /
       ``.markdown`` / ``.txt`` / ``.text``) **or** is ``None`` →
@@ -433,6 +467,12 @@ def parse_document(
             to the real parser chain. Ignored on the simulator path.
         config: Parsing knobs that influence the parse_version. Pass
             ``None`` to use simulator defaults.
+        short_circuit_if_artifacts_exist: When True (default), reuse
+            existing canonical artifacts if all three are already in
+            the object store under the resolved
+            ``derived/parse_<version>/`` path. Pass ``False`` to force
+            a re-parse + re-write (used by tests pinning DocParser
+            invocation count).
     """
     from aperag.mcp.tools.parse_version import compute_parse_version
 
@@ -446,6 +486,45 @@ def parse_document(
         chunking_config=cfg.chunking.serialize(),
     )
 
+    markdown_path = derived_artifact(
+        collection_id=collection_id,
+        document_id=document_id,
+        parse_version=parse_version,
+        filename="markdown.md",
+    )
+    outline_path = derived_artifact(
+        collection_id=collection_id,
+        document_id=document_id,
+        parse_version=parse_version,
+        filename="outline.json",
+    )
+    chunks_path = derived_artifact(
+        collection_id=collection_id,
+        document_id=document_id,
+        parse_version=parse_version,
+        filename="chunks.jsonl",
+    )
+
+    if short_circuit_if_artifacts_exist and _all_artifacts_present(
+        store=store,
+        markdown_path=markdown_path,
+        outline_path=outline_path,
+        chunks_path=chunks_path,
+    ):
+        logger.info(
+            "indexing parser short-circuit collection=%s document=%s parse_version=%s "
+            "(all derived artifacts already present; skipping DocParser + writes)",
+            collection_id,
+            document_id,
+            parse_version,
+        )
+        return ParseResult(
+            parse_version=parse_version,
+            markdown_path=markdown_path,
+            outline_path=outline_path,
+            chunks_path=chunks_path,
+        )
+
     if extension is None or extension in _SIMULATOR_EXTENSIONS:
         try:
             markdown = source_bytes.decode("utf-8")
@@ -470,25 +549,6 @@ def parse_document(
     outline = _build_outline(markdown)
     chunks = _split_chunks(markdown, cfg.chunking)
 
-    markdown_path = derived_artifact(
-        collection_id=collection_id,
-        document_id=document_id,
-        parse_version=parse_version,
-        filename="markdown.md",
-    )
-    outline_path = derived_artifact(
-        collection_id=collection_id,
-        document_id=document_id,
-        parse_version=parse_version,
-        filename="outline.json",
-    )
-    chunks_path = derived_artifact(
-        collection_id=collection_id,
-        document_id=document_id,
-        parse_version=parse_version,
-        filename="chunks.jsonl",
-    )
-
     write_atomic(store, markdown_path, markdown.encode("utf-8"))
     write_atomic(
         store,