diff --git a/core-api/src/core_api/middleware/request_timeout.py b/core-api/src/core_api/middleware/request_timeout.py index 9ade6e494..090b529a2 100644 --- a/core-api/src/core_api/middleware/request_timeout.py +++ b/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: diff --git a/core-api/src/core_api/schemas.py b/core-api/src/core_api/schemas.py index 8e92d9f3a..ac36061d5 100644 --- a/core-api/src/core_api/schemas.py +++ b/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): diff --git a/core-api/src/core_api/services/ingest_service.py b/core-api/src/core_api/services/ingest_service.py index a57500735..83608e33a 100644 --- a/core-api/src/core_api/services/ingest_service.py +++ b/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 ``#``, 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. diff --git a/core-storage-api/src/core_storage_api/services/postgres_service.py b/core-storage-api/src/core_storage_api/services/postgres_service.py index 7251f40e2..f8dcf0bae 100644 --- a/core-storage-api/src/core_storage_api/services/postgres_service.py +++ b/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), ) ) diff --git a/tests/test_ingest_commit.py b/tests/test_ingest_commit.py index 423f5f844..16e10585a 100644 --- a/tests/test_ingest_commit.py +++ b/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 #.""" + 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 # --------------------------------------------------------------------------- diff --git a/tests/test_integration_search.py b/tests/test_integration_search.py index d56d9a218..b2eb23626 100644 --- a/tests/test_integration_search.py +++ b/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" + )