feat(celery Wave 6 #34): chunk_id schema canonical unification (#1738)

* feat(celery Wave 6 #34): chunk_id schema canonical unification

Per docs/modularization/indexing-redesign-design-pack.md §K.11.1 #34
(huangheng T1 obs B + Wave 5 P5B chunk_id 5th item deferred).

## Changes

aperag/indexing/vector.py
- VectorBackend.upsert_point(chunk_id=) → upsert_point(point_id=)
  (canonical Qdrant naming, aligned with summary/vision protocols)
- InMemoryVectorBackend record key renamed chunk_id → point_id
- Vector worker callsite passes point_id=chunk["chunk_id"]; chunk_id
  remains in payload for hybrid-dedup with fulltext (§C.6 trade-off lock)

aperag/indexing/worker_factory.py
- _QdrantPointBackend.upsert_point(): single point_id keyword (drop
  pre-Wave-6 dual chunk_id|point_id transition shim)
- Drop merged_payload.setdefault("chunk_id", identifier) — adapter no
  longer injects misleading chunk_id into summary/vision payloads
  (their points are not chunks); each modality controls its payload

aperag/indexing/reconciler.py
- utc_now audit comment: utc_now usage in reconciler is
  application-level wall-clock for lease comparison + gmt_updated
  stamps, distinct from Wave 5 P5B server_default=CURRENT_TIMESTAMP
  ORM-creation defaults. No further migration needed.

tests/unit_test/indexing/test_chunk_id_schema_canonical.py (new)
- 7 contract tests pin canonical naming:
  * VectorBackend / SummaryBackend / VisionBackend Protocols use point_id
  * InMemoryVectorBackend round-trips point_id at record level + chunk_id at payload level
  * Legacy chunk_id= keyword raises TypeError on InMemoryVectorBackend
  * _QdrantPointBackend.upsert_point uses single point_id param
  * _QdrantPointBackend does not inject chunk_id into payload
  * Parser chunks.jsonl chunk_id field naming preserved

tests/integration/test_cleanup_fan_out.py
tests/unit_test/indexing/test_t1_3_vector_fulltext.py
tests/unit_test/indexing/test_t2_1_runtime.py
tests/unit_test/indexing/test_t3_1_dispatcher_path_c.py
- backend.upsert_point(chunk_id=...) → point_id=...
- record reads p["chunk_id"] → p["point_id"] / p["payload"]["chunk_id"]

## Production-readiness 三类 layer (per §K.11.4)

- must-be-real: parser layer chunk_id field schema unified across
  vector/summary/vision backend protocol surfaces; remaining utc_now
  usage audited (reconciler.py only, application-level, distinct from
  ORM defaults Wave 5 P5B already migrated)
- may-be-gated: vector worker still keeps chunk_id in payload for
  hybrid-dedup with fulltext (§C.6 contract preserved)
- fully-resolves: huangheng T1 obs B + Wave 5 P5B chunk_id 5th item
  deferred (per spec §K.11.1 row 34)

## simple-stable 4 guardrail (per feedback_simple_stable_zero_maintenance.md)

1. 不无限扩范围 ✅ — scope limited to vector protocol + adapter, no
   cross-cutting touch
2. 功能做实 ✅ — real schema rename, no transitional shim left behind
3. 简单稳定 ✅ — drops dual-API polymorphism + drops hidden payload
   side-effect (setdefault chunk_id), each layer's contract is now
   self-evident from signature
4. 私有化免维护 ✅ — operator/dev reading code sees canonical naming
   without needing to track which arg name belongs to which modality

## hard-cut policy (per earayu2 msg=30c81478)

No production data → schema breaking change applied directly. No
backward-compat shim, no deprecation window. Test that the legacy
chunk_id= keyword raises TypeError catches accidental regression.

## Pre-check (per K.11.5 Pattern 2)

grep upsert_point across aperag/indexing/ + tests/ — all callsites
audited, all callers migrated, contract test pins canonical name.

## Local gates

- 152/152 indexing unit tests pass
- 21/21 indexing integration tests pass (cleanup_fan_out +
  dispatch_with_parse + inline_mode_smoke + parse_async_roundtrip)
- 7/7 new contract tests pass
- ruff check + ruff format clean

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

* feat(celery Wave 6 #38): application cache cross-loop lazy-rebuild + WARN log

Per docs/modularization/indexing-redesign-design-pack.md §K.11.1 #38
(huangheng PR #1734 sync point 7 + feedback_announce_equals_landed.md
narrative-truth invariant).

## Problem (pre-Wave-6 narrative-truth violation)

`aperag.cache.application_runtime.get_application_cache()` keyed the
cached `ApplicationCache` instance on the running asyncio event loop
(the Redis async client is loop-bound). When the running loop
changed (test process restart, worker reinit, asyncio
re-initialisation, etc.), the function silently swapped the cache
for a `NoopApplicationCacheBackend` with `enabled=False`. From that
point on the worker paid the full LiteLLM / embedding round-trip on
every request, with no log line, no metric, and no operator-visible
signal that caching had degraded to zero.

## Fix

When the loop-switch branch fires:
- emit a WARN log with operator-actionable phrasing
- bump
  `application_cache_metrics["application_runtime"]["loop_switch_rebuild"]`
  (uses the existing ApplicationCacheMetrics instance — no new
  metrics infra)
- reset `_async_cache=None` and fall through to the normal
  initialisation path (which re-establishes the real Redis client on
  the new loop)

The Noop fallback is now reached only when Redis is genuinely
unreachable on the new loop — never as a silent loop-switch
downgrade.

## Tests (4 new)

tests/unit_test/cache/test_application_cache.py
- test_application_cache_rebuilds_on_loop_switch_instead_of_silent_noop:
  two `_run_in_new_loop()` calls produce distinct ApplicationCache
  instances, second wraps real `ApplicationRedisCacheBackend` (not
  `NoopApplicationCacheBackend`); metric counter incremented to 1;
  WARN log captured.
- test_application_cache_does_not_increment_loop_switch_metric_on_first_build:
  first-ever build does not bump the loop-switch counter.
- test_application_cache_repeat_call_in_same_loop_is_singleton_no_metric_bump:
  same-loop repeat returns identical singleton, no metric bump.
- test_application_cache_falls_back_to_noop_when_redis_unreachable_on_new_loop:
  if Redis genuinely fails on the new loop, rebuild attempts the
  real client, falls back to Noop, but metric still records the
  loop switch (operators see the rebuild attempt + the Redis
  failure WARN that was already there).

## Production-readiness 三类 layer (per §K.11.4)

- must-be-real: real WARN log + real metric counter + real Redis
  client rebuild on cross-loop call (no silent zero-cache)
- may-be-gated: first cross-loop request may pay rebuild cost
  (re-ping Redis) — observable via metric, not silent
- fully-resolves: huangheng PR #1734 sync point 7 +
  feedback_announce_equals_landed.md narrative-truth lock

## simple-stable 4 guardrail

1. 不无限扩范围 ✅ — single-function fix in
   application_runtime.py, reuses existing
   `application_cache_metrics` infra, no new public API
2. 功能做实 ✅ — real cache rebuild not a placeholder
3. 简单稳定 ✅ — fewer paths than before (one rebuild path replaces
   the silent-Noop branch), each branch logged
4. 私有化免维护 ✅ — operator can grep
   `loop_switch_rebuild` in metrics + log "rebuilding for new event
   loop" to diagnose worker setup issues

## Local gates

- 43/43 cache unit tests pass (39 existing + 4 new)
- ruff check + ruff format clean

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

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: earayu

aperag/cache/application_runtime.py

@@ -43,17 +43,39 @@ def application_cache_policy(namespace: str, *, enabled: bool | None = None) ->
 
 
 async def get_application_cache() -> ApplicationCache:
+    """Return the singleton :class:`ApplicationCache` for the running
+    event loop, lazily building it if needed.
+
+    Wave 6 #38 (per `feedback_announce_equals_landed.md` narrative-truth
+    invariant): when the running event loop changes (test process
+    starts a new loop, worker process reinitialises, etc.) the cached
+    instance — whose underlying ``async_redis.Redis`` client is bound
+    to the prior loop — is no longer usable. Pre-Wave-6 we silently
+    swapped it for a :class:`NoopApplicationCacheBackend`, which meant
+    callers paid LiteLLM/embedding cost on every request from then on
+    with no signal that caching had degraded.
+
+    Wave 6 fix: rebuild the cache on the new loop (re-establish the
+    real Redis client) and emit a WARN log + bump
+    ``application_cache_metrics["application_runtime"]["loop_switch_rebuild"]``
+    so operators can observe loop-switch frequency. Rebuild can fail
+    (Redis unreachable on the new loop) — the fallback is still a
+    Noop backend, but only when Redis is actually broken, not when
+    the loop merely changed.
+    """
     global _async_cache, _async_cache_loop
     loop = asyncio.get_running_loop()
     if _async_cache is not None and _async_cache_loop is loop:
         return _async_cache
     if _async_cache is not None and _async_cache_loop is not loop:
-        _async_cache = ApplicationCache(
-            backend=NoopApplicationCacheBackend(),
-            default_policy=application_cache_policy("default", enabled=False),
+        logger.warning(
+            "Application cache rebuilding for new event loop: prior cache was bound "
+            "to a different loop and its Redis client cannot be reused. "
+            "Frequent loop switches indicate a worker / test setup issue."
         )
-        _async_cache_loop = loop
-        return _async_cache
+        application_cache_metrics.increment("application_runtime", "loop_switch_rebuild")
+        _async_cache = None
+        _async_cache_loop = None
 
     if not settings.cache_enabled:
         _async_cache = ApplicationCache(

aperag/indexing/reconciler.py

@@ -325,7 +325,19 @@ async def reconcile_collection_summaries_hook(
     )
 
     def _reclaim_stale_and_claim_pending() -> list[tuple[str, str, int, str]]:
-        """Sync DB-only worker. Returns list of claimed dispatch tuples."""
+        """Sync DB-only worker. Returns list of claimed dispatch tuples.
+
+        Note (Wave 6 #34 utc_now audit): the ``utc_now`` calls in this
+        function are application-level wall-clock reads used to compute
+        lease expiry windows and stamp ``gmt_last_reconciled`` /
+        ``gmt_updated`` for in-flight rows. They are intentionally
+        Python-side and distinct from the ORM-default
+        ``server_default=CURRENT_TIMESTAMP`` migration done in Wave 5
+        P5B for ``_LineageEntityRow`` / ``_LineageRelationRow``: those
+        cover row-creation defaults; reconciler updates need a wall-
+        clock value materialised inside the worker process for lease
+        comparison logic, so server-side defaults do not apply here.
+        """
         from aperag.utils.utils import utc_now as _utc_now
 
         claimed_dispatches: list[tuple[str, str, int, str]] = []

aperag/indexing/vector.py

@@ -75,17 +75,24 @@ def delete_by_filter(self, *, document_id: str, parse_version: str) -> int:
     def upsert_point(
         self,
         *,
-        chunk_id: str,
+        point_id: str,
         embedding: list[float],
         payload: dict[str, Any],
     ) -> None:
-        """Idempotent point insert keyed on ``chunk_id`` (Qdrant point id)."""
+        """Idempotent point insert keyed on ``point_id`` (Qdrant point id).
+
+        For the vector modality the caller passes the parser-emitted
+        ``chunk_id`` as ``point_id`` (one chunk → one point). The
+        ``chunk_id`` value is also written into ``payload["chunk_id"]``
+        by the worker so retrieval can echo it back; vector + fulltext
+        share that payload field for hybrid-dedup (§C.6).
+        """
 
 
 class InMemoryVectorBackend:
     """Process-local in-memory backend for unit tests.
 
-    Stores points in a dict keyed by ``chunk_id``. Implements the
+    Stores points in a dict keyed by ``point_id``. Implements the
     :class:`VectorBackend` protocol so vector.sync can target it
     transparently.
     """
@@ -95,22 +102,22 @@ def __init__(self) -> None:
 
     def delete_by_filter(self, *, document_id: str, parse_version: str) -> int:
         deleted = 0
-        for chunk_id in list(self._points):
-            payload = self._points[chunk_id].get("payload", {})
+        for point_id in list(self._points):
+            payload = self._points[point_id].get("payload", {})
             if payload.get("document_id") == document_id and payload.get("parse_version") == parse_version:
-                self._points.pop(chunk_id)
+                self._points.pop(point_id)
                 deleted += 1
         return deleted
 
     def upsert_point(
         self,
         *,
-        chunk_id: str,
+        point_id: str,
         embedding: list[float],
         payload: dict[str, Any],
     ) -> None:
-        self._points[chunk_id] = {
-            "chunk_id": chunk_id,
+        self._points[point_id] = {
+            "point_id": point_id,
             "embedding": list(embedding),
             "payload": dict(payload),
         }
@@ -126,10 +133,10 @@ def points_for_document(self, document_id: str, parse_version: str | None = None
             if parse_version is not None and payload.get("parse_version") != parse_version:
                 continue
             out.append(record)
-        return sorted(out, key=lambda r: r["chunk_id"])
+        return sorted(out, key=lambda r: r["point_id"])
 
     def all_points(self) -> list[dict[str, Any]]:
-        return sorted(self._points.values(), key=lambda r: r["chunk_id"])
+        return sorted(self._points.values(), key=lambda r: r["point_id"])
 
 
 def _placeholder_embedding(text: str, dim: int = SIMULATOR_EMBEDDING_DIM) -> list[float]:
@@ -229,7 +236,7 @@ async def sync(
                 "page_idx": chunk.get("page_idx"),
             }
             self._backend.upsert_point(
-                chunk_id=chunk_id,
+                point_id=chunk_id,
                 embedding=embedding,
                 payload=payload,
             )

aperag/indexing/worker_factory.py

@@ -94,10 +94,16 @@ class _QdrantPointBackend:
     vision modalities consume.
 
     All three modalities share the same Qdrant-shaped surface (delete
-    by ``(document_id, parse_version)`` filter, upsert by
-    ``chunk_id``/``point_id``). One adapter class satisfies the three
-    Protocols structurally — no inheritance needed because the
-    Protocols are ``@runtime_checkable``.
+    by ``(document_id, parse_version)`` filter, upsert by ``point_id``).
+    One adapter class satisfies the three Protocols structurally — no
+    inheritance needed because the Protocols are ``@runtime_checkable``.
+
+    The ``point_id`` is the canonical name for the Qdrant point
+    identifier across all three modalities (per Wave 6 #34 schema
+    unification). Vector workers pass the parser-emitted ``chunk_id``
+    here; summary uses ``summary:<doc>:<v>``; vision uses
+    ``vision:<doc>:<v>:<image_id>``. Each worker controls its own
+    payload — the adapter does not inject anything.
     """
 
     def __init__(self, *, connector: Any) -> None:
@@ -120,43 +126,34 @@ def delete_by_filter(self, *, document_id: str, parse_version: str) -> int:
     def upsert_point(
         self,
         *,
-        chunk_id: str | None = None,
-        point_id: str | None = None,
+        point_id: str,
         embedding: list[float],
         payload: dict[str, Any],
     ) -> None:
-        # Vector modality calls with ``chunk_id``; summary / vision
-        # modalities call with ``point_id``. Both end up as the
-        # underlying Qdrant point id.
-        #
         # Qdrant only accepts unsigned-integer or UUID point ids. The
         # T1.1 parser produces chunk ids of the form
         # ``<sha-prefix>:<index>`` (e.g. ``f766a946575ec3b4:0000``)
         # which Qdrant rejects with HTTP 400 "is not a valid point
         # ID". Map the caller-supplied string id into a deterministic
         # UUID5 so retries land on the same point and the upsert is
-        # idempotent — and stash the original id in the payload so
-        # the read path can still surface it to clients.
+        # idempotent. The caller's payload is forwarded verbatim — it
+        # already carries whatever modality-specific identifier the
+        # read path needs (vector keeps ``chunk_id`` in payload for
+        # hybrid-dedup; summary uses ``summary_text``; vision uses
+        # ``image_id``).
         import uuid
 
         from aperag.vectorstore.dto import VectorPoint
 
-        identifier = chunk_id if chunk_id is not None else point_id
-        if not identifier:
-            raise ValueError("upsert_point requires either chunk_id or point_id")
-        identifier = str(identifier)
-        qdrant_id = str(uuid.uuid5(uuid.NAMESPACE_OID, identifier))
-        merged_payload = dict(payload)
-        # Preserve the original id under a stable key so the read
-        # path can echo it back; ``chunk_id`` is what vector modality
-        # already writes so we don't overwrite it.
-        merged_payload.setdefault("chunk_id", identifier)
+        if not point_id:
+            raise ValueError("upsert_point requires point_id")
+        qdrant_id = str(uuid.uuid5(uuid.NAMESPACE_OID, str(point_id)))
         self._connector.upsert(
             [
                 VectorPoint(
                     id=qdrant_id,
                     vector=list(embedding),
-                    payload=merged_payload,
+                    payload=dict(payload),
                 )
             ]
         )

tests/integration/test_cleanup_fan_out.py

@@ -136,7 +136,7 @@ async def factory_closure(row: DocumentIndex):
     pv_a = "doc-del-a-pv0001"[:16]
     _insert_row(engine, document_id="doc-del", parse_version=pv_a, modality=Modality.VECTOR)
     factory_backend.upsert_point(
-        chunk_id="chunk-fac",
+        point_id="chunk-fac",
         embedding=[0.0] * 16,
         payload={
             "document_id": "doc-del",
@@ -150,7 +150,7 @@ async def factory_closure(row: DocumentIndex):
         },
     )
     fallback_backend.upsert_point(
-        chunk_id="chunk-fallback",
+        point_id="chunk-fallback",
         embedding=[0.0] * 16,
         payload={
             "document_id": "doc-del",
@@ -245,7 +245,7 @@ async def factory_closure(row: DocumentIndex):
         backend = backends[(col, modality)]
         if modality is Modality.VECTOR:
             backend.upsert_point(
-                chunk_id=chunk_id,
+                point_id=chunk_id,
                 embedding=[0.0] * 16,
                 payload={
                     "document_id": document_id,
@@ -360,7 +360,7 @@ def test_orphan_parse_version_gc_uses_worker_factory(engine):
     _set_updated_at(engine, old_id, _utcnow() - timedelta(hours=2))
 
     backend.upsert_point(
-        chunk_id="chunk-orphan",
+        point_id="chunk-orphan",
         embedding=[0.0] * 16,
         payload={
             "document_id": "doc-orphan",
@@ -415,7 +415,7 @@ def test_collection_deletion_cascade_uses_worker_factory(engine):
         collection_id="col-doomed",
     )
     backend.upsert_point(
-        chunk_id="chunk-col",
+        point_id="chunk-col",
         embedding=[0.0] * 16,
         payload={
             "document_id": "doc-col",
@@ -489,7 +489,7 @@ def test_workers_map_only_path_unchanged_for_existing_callers(engine):
     pv = "compatparsever01"[:16]
     _insert_row(engine, document_id="doc-compat", parse_version=pv, modality=Modality.VECTOR)
     backend.upsert_point(
-        chunk_id="chunk-compat",
+        point_id="chunk-compat",
         embedding=[0.0] * 16,
         payload={
             "document_id": "doc-compat",