feat(core-api): source-document provenance on ingest commit

core-api/src/core_api/middleware/request_timeout.py

@@ -39,7 +39,22 @@
 # well past the hot-path budget; cancelling it mid-flight is pointless
 # (the purge is one transaction per tenant and idempotent on retry), so
 # it opts out and is bounded by the storage client's own timeout instead.
-_TIMEOUT_OPT_OUT_PATHS: frozenset[str] = frozenset({"/api/v1/memories/bulk", "/api/v1/admin/org/purge-data"})
+# ``/ingest/{preview,commit}`` are LLM-bound operator-plane calls that
+# scale with document size (preview: one extraction round per ~3k-token
+# section; commit: strong-mode enrichment across up-to-BULK_MAX_ITEMS
+# chunks). A fact-dense wiki page blew the 45s budget mid-commit,
+# 504ing AFTER earlier chunks had persisted — the same
+# partial-write-then-cancel shape as the bulk case above. Both are
+# bounded by the per-call LLM timeouts + the caller's own client
+# timeout instead.
+_TIMEOUT_OPT_OUT_PATHS: frozenset[str] = frozenset(
+    {
+        "/api/v1/memories/bulk",
+        "/api/v1/admin/org/purge-data",
+        "/api/v1/ingest/preview",
+        "/api/v1/ingest/commit",
+    }
+)
 
 
 def _is_opted_out(path: str) -> bool:

core-api/src/core_api/schemas.py

@@ -424,6 +424,17 @@ class IngestCommitRequest(BaseModel):
     # call (cache-hit). Backward-compatible: omitting it just disables
     # the cache for future previews of this content.
     doc_hash: str | None = None
+    # Source-document provenance (cold-start A0.1). One commit distills one
+    # source document, so both fields are request-level and apply to every
+    # fact in the batch. ``source_doc_id`` is the stable id in the source
+    # system (e.g. a Confluence page id) — stamped into each memory's
+    # metadata and onto the parent Document so a later re-sync can find
+    # everything derived from that doc. ``ts_valid_start`` is the source's
+    # last-edited time — stamped as each memory's temporal-validity start
+    # so imported facts inherit the age of the page they came from instead
+    # of landing as brand-new knowledge.
+    source_doc_id: str | None = None
+    ts_valid_start: datetime | None = None
 
 
 class RelationUpsertOut(BaseModel):

core-api/src/core_api/services/ingest_service.py

@@ -16,7 +16,7 @@
 
 from core_api.clients.storage_client import get_storage_client
 from core_api.config import settings
-from core_api.constants import MEMORY_TYPES
+from core_api.constants import BULK_MAX_ITEMS, MEMORY_TYPES
 from core_api.providers._retry import call_with_fallback
 from core_api.schemas import (
     BulkMemoryCreate,
@@ -871,6 +871,8 @@ async def _write_parent_ingest_document(
         "uploaded_at": datetime.now(UTC).isoformat(),
         "ingest_ms": ingest_ms,
         "agent_id": request.agent_id,
+        "source_doc_id": request.source_doc_id,
+        "ts_valid_start": request.ts_valid_start.isoformat() if request.ts_valid_start else None,
     }
     summary = _summarize_batch_for_embedding(survivors)
     if summary is not None:
@@ -1040,6 +1042,8 @@ async def ingest_commit(request: IngestCommitRequest) -> dict:
             }
             if request.doc_hash:
                 metadata["doc_hash"] = request.doc_hash
+            if request.source_doc_id:
+                metadata["source_doc_id"] = request.source_doc_id
             salience_value = getattr(fact, "salience", None)
             if salience_value is not None:
                 metadata["salience"] = salience_value
@@ -1050,47 +1054,66 @@ async def ingest_commit(request: IngestCommitRequest) -> dict:
                     source_uri=effective_source,
                     run_id=run_id,
                     metadata=metadata,
+                    # Explicit source timestamp wins over enrichment's
+                    # content-inferred value (bulk path only fills
+                    # ts_valid_start from enrichment when it's None).
+                    ts_valid_start=request.ts_valid_start,
                 )
             )
-        bulk_data = BulkMemoryCreate(
-            tenant_id=request.tenant_id,
-            fleet_id=request.fleet_id,
-            agent_id=request.agent_id,
-            items=bulk_items,
-        )
-        try:
-            bulk_response = await create_memories_bulk(bulk_data, bulk_attempt_id=run_id)
-            created = bulk_response.created
-            skipped_in_loop = bulk_response.duplicates
-            errored = bulk_response.errors
-            # Surface per-item error reasons in the logs so the cleanup
-            # message at the bottom of this function still points at the
-            # offending facts.
-            for item in bulk_response.results:
-                if item.status == "error":
-                    # Mirror the legacy "fact[N]" log format the
-                    # P1.C-lite runbook + operator greps depend on.
-                    logger.warning(
-                        "ingest_commit: fact[%d] write failed (run_id=%s): %s",
-                        item.index,
-                        run_id,
-                        item.error,
-                    )
-        except HTTPException as e:
-            # A 4xx/5xx from the bulk endpoint aborts the whole batch
-            # (e.g. 504 from the bulk-embedding timeout). Mirror the
-            # prior behaviour where a non-409 escape raised through
-            # gather and aborted the run — but now the run_id is still
-            # logged so the operator can locate any partial rows.
-            logger.exception(
-                "ingest_commit: bulk write failed with HTTP %d (run_id=%s) — "
-                "0 facts persisted on this attempt; safe to retry",
-                e.status_code,
-                run_id,
+        # Fact-dense documents can exceed ``BULK_MAX_ITEMS`` (first hit
+        # importing real wiki pages: 115-194 facts per page — the single
+        # BulkMemoryCreate 500'd on its own validator). Split into
+        # ≤BULK_MAX_ITEMS chunks. Attempt-id scheme: chunk 0 keeps the
+        # bare ``run_id`` (byte-identical to the pre-split contract, so
+        # a retry that straddles a deploy still dedups); later chunks
+        # append ``#<n>``, deterministic so a retry of the same run
+        # re-derives the same ids and sees ``duplicate_attempt``.
+        created = 0
+        skipped_in_loop = 0
+        errored = 0
+        chunks = [bulk_items[i : i + BULK_MAX_ITEMS] for i in range(0, len(bulk_items), BULK_MAX_ITEMS)]
+        for chunk_index, chunk in enumerate(chunks):
+            bulk_data = BulkMemoryCreate(
+                tenant_id=request.tenant_id,
+                fleet_id=request.fleet_id,
+                agent_id=request.agent_id,
+                items=chunk,
             )
-            created = 0
-            skipped_in_loop = 0
-            errored = len(survivors)
+            attempt_id = run_id if chunk_index == 0 else f"{run_id}#{chunk_index}"
+            try:
+                bulk_response = await create_memories_bulk(bulk_data, bulk_attempt_id=attempt_id)
+                created += bulk_response.created
+                skipped_in_loop += bulk_response.duplicates
+                errored += bulk_response.errors
+                # Surface per-item error reasons in the logs so the cleanup
+                # message at the bottom of this function still points at the
+                # offending facts.
+                for item in bulk_response.results:
+                    if item.status == "error":
+                        # Mirror the legacy "fact[N]" log format the
+                        # P1.C-lite runbook + operator greps depend on.
+                        logger.warning(
+                            "ingest_commit: fact[%d] write failed (run_id=%s): %s",
+                            item.index + chunk_index * BULK_MAX_ITEMS,
+                            run_id,
+                            item.error,
+                        )
+            except HTTPException as e:
+                # A 4xx/5xx from the bulk endpoint aborts this chunk
+                # (e.g. 504 from the bulk-embedding timeout). Later
+                # chunks are independent — keep going; a retry of the
+                # run re-dedups the committed chunks via their attempt
+                # ids and re-attempts only what failed.
+                logger.exception(
+                    "ingest_commit: bulk chunk %d/%d failed with HTTP %d (run_id=%s) — "
+                    "%d facts not persisted on this attempt; safe to retry",
+                    chunk_index + 1,
+                    len(chunks),
+                    e.status_code,
+                    run_id,
+                    len(chunk),
+                )
+                errored += len(chunk)
     else:
         # All facts pre-deduped by the content-hash sweep above; nothing
         # to do here.

core-storage-api/src/core_storage_api/services/postgres_service.py

@@ -1385,9 +1385,15 @@ async def memory_scored_search(
             # conflicted row un-demoted, and load_and_serialize still injects its
             # supersedes successor so both sides are visible. ``outdated`` stays
             # fully excluded.
+            # ``archived`` and ``cancelled`` joined the exclusion after an
+            # A/B recall demo surfaced an archived row ranking in search:
+            # archival is the source-deletion contract's "out of recall,
+            # restorable" state (connector re-sync tombstones), and it
+            # meant nothing while search still served the rows. Same
+            # explicit-``status_filter`` override applies for inspection.
             scored_stmt = scored_stmt.where(
                 or_(
-                    Memory.status.notin_(("outdated", "conflicted")),
+                    Memory.status.notin_(("outdated", "conflicted", "archived", "cancelled")),
                     and_(Memory.status == "conflicted", _exact_lexical_match),
                 )
             )

tests/test_ingest_commit.py

@@ -122,6 +122,7 @@ async def fake_create_memories_bulk(data, *, bulk_attempt_id):
                     source_uri=item.source_uri,
                     run_id=item.run_id,
                     metadata=item.metadata,
+                    ts_valid_start=item.ts_valid_start,
                     # ``write_mode`` doesn't exist on ``BulkMemoryItem``
                     # (the bulk path is implicitly strong-mode for
                     # ingest), so surface "strong" to keep the legacy
@@ -532,6 +533,119 @@ async def test_text_input_fallback_when_neither_set(captured):
     assert captured.writes[0].metadata["ingest_url"] is None
 
 
+# ---------------------------------------------------------------------------
+# Fact-dense documents — bulk writes split at BULK_MAX_ITEMS
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_over_bulk_cap_splits_into_chunks(captured):
+    """>100 facts (real wiki pages hit 115-194) must split into
+    ≤BULK_MAX_ITEMS bulk calls instead of 500ing on BulkMemoryCreate's
+    own validator. Chunk 0 keeps the bare run_id attempt key
+    (pre-split contract); later chunks append #<n>."""
+    from core_api.constants import BULK_MAX_ITEMS
+
+    n_facts = BULK_MAX_ITEMS + 50
+    req = _request("t1", *[f"fact number {i}" for i in range(n_facts)], run_id="run-big")
+    result = await ingest_service.ingest_commit(request=req)
+
+    assert result["memories_created"] == n_facts
+    assert len(captured.writes) == n_facts
+    assert [len(b.items) for b in captured.bulk_calls] == [BULK_MAX_ITEMS, 50]
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_chunk_attempt_ids_deterministic(captured, monkeypatch):
+    """Retrying the same run re-derives identical per-chunk attempt ids
+    so already-committed chunks dedup as duplicate_attempt."""
+    from core_api.services import ingest_service as svc
+
+    attempt_ids: list[str] = []
+    real_bulk = svc.create_memories_bulk
+
+    async def spy(data, *, bulk_attempt_id):
+        attempt_ids.append(bulk_attempt_id)
+        return await real_bulk(data, bulk_attempt_id=bulk_attempt_id)
+
+    monkeypatch.setattr(svc, "create_memories_bulk", spy)
+    from core_api.constants import BULK_MAX_ITEMS
+
+    n_facts = 2 * BULK_MAX_ITEMS + 1
+    req = _request("t1", *[f"unique fact {i}" for i in range(n_facts)], run_id="run-big")
+    await ingest_service.ingest_commit(request=req)
+
+    assert attempt_ids == ["run-big", "run-big#1", "run-big#2"]
+
+
+# ---------------------------------------------------------------------------
+# A0.1 — source-document provenance (source_doc_id + ts_valid_start)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_source_doc_provenance_stamped_on_every_item(captured):
+    """Request-level ``source_doc_id`` lands in every item's metadata and
+    ``ts_valid_start`` on every item's temporal-validity column — one commit
+    distills one source document, so both apply batch-wide."""
+    from datetime import UTC, datetime
+
+    last_edited = datetime(2024, 3, 7, 12, 0, tzinfo=UTC)
+    req = _request(
+        "t1",
+        "fact one",
+        "fact two",
+        source_doc_id="confluence:12345",
+        ts_valid_start=last_edited,
+    )
+    await ingest_service.ingest_commit(request=req)
+
+    assert len(captured.writes) == 2
+    for mc in captured.writes:
+        assert mc.metadata["source_doc_id"] == "confluence:12345"
+        assert mc.ts_valid_start == last_edited
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_source_doc_provenance_absent_when_not_supplied(captured):
+    """Callers that don't pass the new fields see the exact pre-A0.1 write
+    shape: no ``source_doc_id`` metadata key, ``ts_valid_start=None`` (so
+    enrichment's content-inferred value still applies downstream)."""
+    req = _request("t1", "fact one")
+    await ingest_service.ingest_commit(request=req)
+
+    mc = captured.writes[0]
+    assert "source_doc_id" not in mc.metadata
+    assert mc.ts_valid_start is None
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_parent_document_carries_source_doc_provenance(captured):
+    """The parent ``ingest-sources`` Document records the same source-doc
+    provenance so a re-sync can map run → source document without reading
+    per-memory metadata."""
+    from datetime import UTC, datetime
+
+    last_edited = datetime(2024, 3, 7, 12, 0, tzinfo=UTC)
+    req = _request(
+        "t1",
+        "fact one",
+        source_doc_id="confluence:12345",
+        ts_valid_start=last_edited,
+    )
+    await ingest_service.ingest_commit(request=req)
+
+    assert len(captured.parent_doc_writes) == 1
+    data = captured.parent_doc_writes[0]["data"]
+    assert data["source_doc_id"] == "confluence:12345"
+    assert data["ts_valid_start"] == last_edited.isoformat()
+
+
 # ---------------------------------------------------------------------------
 # PR #3 — P1.E suggested_type validation at commit
 # ---------------------------------------------------------------------------

tests/test_integration_search.py

@@ -436,3 +436,77 @@ async def test_outdated_exact_match_still_excluded(self, db, tenant_id):
         assert not any("vortex" in r.content for r in results), (
             "outdated exact-match should remain excluded"
         )
+
+
+@pytest.mark.integration
+class TestArchivedAndCancelledExcluded:
+    """``archived``/``cancelled`` memories must not rank in scored search.
+
+    Archival is the connector source-deletion contract's "out of recall,
+    restorable" state (re-sync tombstones a doc's memories archived and can
+    restore them later). The exclusion list originally covered only
+    ``outdated``/``conflicted``, so archived rows kept surfacing in
+    /api/v1/search — an A/B recall demo returned a memory whose source doc
+    had been archived. Even an exact lexical match stays hidden: unlike
+    ``conflicted``, these statuses mean "not currently part of the corpus",
+    not "a competing claim exists".
+    """
+
+    async def test_archived_excluded_even_on_exact_match(self, db, tenant_id):
+        await _insert_memory(
+            tenant_id,
+            "Deployment host is plzkv cluster in region east",
+            weight=0.7,
+            status="archived",
+        )
+        await _insert_memory(
+            tenant_id,
+            "Deployment pipeline uses staging green gate before promote",
+            weight=0.7,
+            status="active",
+        )
+
+        from core_api.services.memory_service import search_memories
+
+        results = await search_memories(tenant_id, "plzkv deployment host", top_k=10)
+        assert not any("plzkv" in r.content for r in results), (
+            "archived memory leaked into scored search results"
+        )
+
+    async def test_cancelled_excluded(self, db, tenant_id):
+        await _insert_memory(
+            tenant_id,
+            "Initiative xrretq budget is nine million",
+            weight=0.7,
+            status="cancelled",
+        )
+
+        from core_api.services.memory_service import search_memories
+
+        results = await search_memories(tenant_id, "xrretq budget", top_k=10)
+        assert not any("xrretq" in r.content for r in results), (
+            "cancelled memory leaked into scored search results"
+        )
+
+    async def test_active_sibling_still_returned(self, db, tenant_id):
+        archived = await _insert_memory(
+            tenant_id,
+            "Runbook mmzzt covers failover drill steps",
+            weight=0.7,
+            status="archived",
+        )
+        live = await _insert_memory(
+            tenant_id,
+            "Runbook mmzzt failover drill runs quarterly",
+            weight=0.7,
+            status="active",
+        )
+
+        from core_api.services.memory_service import search_memories
+
+        results = await search_memories(tenant_id, "mmzzt failover drill", top_k=10)
+        result_ids = {str(r.id) for r in results}
+        assert str(archived["id"]) not in result_ids
+        assert str(live["id"]) in result_ids, (
+            "active memory was not returned alongside excluded archived sibling"
+        )