From f2a0dbf81f0605563f9f38facdf15b6e87d3bf34 Mon Sep 17 00:00:00 2001 From: erni Date: Sun, 5 Jul 2026 10:33:11 +0300 Subject: [PATCH 1/5] feat(core-api): stamp source-document provenance on ingest commit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add optional request-level source_doc_id and ts_valid_start to IngestCommitRequest. One commit distills one source document, so both apply to every fact in the batch: source_doc_id lands in each memory's metadata (and on the parent ingest-sources Document) so a later re-sync can find everything derived from that document; ts_valid_start carries the source's last-edited time onto each memory's temporal-validity column so imported facts inherit the age of the page they came from. Both fields are optional — existing callers see the exact prior write shape. Co-Authored-By: Claude Fable 5 Signed-off-by: erni --- core-api/src/core_api/schemas.py | 11 +++ .../src/core_api/services/ingest_service.py | 8 +++ tests/test_ingest_commit.py | 67 +++++++++++++++++++ 3 files changed, 86 insertions(+) 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..d1a7f0773 100644 --- a/core-api/src/core_api/services/ingest_service.py +++ b/core-api/src/core_api/services/ingest_service.py @@ -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,6 +1054,10 @@ 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( diff --git a/tests/test_ingest_commit.py b/tests/test_ingest_commit.py index 423f5f844..91e40b869 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,72 @@ async def test_text_input_fallback_when_neither_set(captured): assert captured.writes[0].metadata["ingest_url"] is None +# --------------------------------------------------------------------------- +# 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 # --------------------------------------------------------------------------- From 4c844f2cda70fcef785bb708d9079088a9846a54 Mon Sep 17 00:00:00 2001 From: erni Date: Sun, 5 Jul 2026 15:23:56 +0300 Subject: [PATCH 2/5] fix(core-api): split ingest commit bulk writes at BULK_MAX_ITEMS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fact-dense documents can extract more than 100 facts (first hit importing real wiki pages: 115-194 facts per page). ingest_commit passed all survivors in a single BulkMemoryCreate, whose items list caps at BULK_MAX_ITEMS — the request 500'd on the model's own validator before reaching the bulk endpoint. Survivors now commit in <=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. A chunk-level HTTP failure no longer aborts the whole commit — later chunks are independent, the failed chunk's facts count as errored, and a retry re-attempts only what failed. Co-Authored-By: Claude Fable 5 Signed-off-by: erni --- .../src/core_api/services/ingest_service.py | 96 +++++++++++-------- tests/test_ingest_commit.py | 47 +++++++++ 2 files changed, 104 insertions(+), 39 deletions(-) diff --git a/core-api/src/core_api/services/ingest_service.py b/core-api/src/core_api/services/ingest_service.py index d1a7f0773..966c912a0 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, @@ -1060,45 +1060,63 @@ async def ingest_commit(request: IngestCommitRequest) -> dict: 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/tests/test_ingest_commit.py b/tests/test_ingest_commit.py index 91e40b869..16e10585a 100644 --- a/tests/test_ingest_commit.py +++ b/tests/test_ingest_commit.py @@ -533,6 +533,53 @@ 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) # --------------------------------------------------------------------------- From b67594c3288e4c6d37afba58df0c6ab825819fad Mon Sep 17 00:00:00 2001 From: erni Date: Sun, 5 Jul 2026 15:35:24 +0300 Subject: [PATCH 3/5] fix(core-api): opt ingest preview/commit out of the request-wide timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both 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). Importing a real fact-dense wiki page blew the 45s blanket budget mid-commit and 504'd AFTER earlier chunks had already persisted — the same partial-write-then-cancel shape that motivated the /memories/bulk opt-out (CAURA-602). Both routes are bounded by the per-call LLM timeouts and the caller's own client timeout instead. Co-Authored-By: Claude Fable 5 Signed-off-by: erni --- .../src/core_api/middleware/request_timeout.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) 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: From 0219ed4551c87946a160c11e939695e3641ed82a Mon Sep 17 00:00:00 2001 From: erni Date: Wed, 8 Jul 2026 10:10:37 +0300 Subject: [PATCH 4/5] fix(storage): exclude archived and cancelled memories from scored search MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scored-search status filter only excluded outdated/conflicted, so memories archived by the source-deletion tombstone flow (and cancelled rows) kept ranking in /api/v1/search — archival removed nothing from recall. Add both statuses to the exclusion; the explicit status_filter override still allows inspection. Integration tests cover archived, cancelled, and the active-sibling-still-returned case. Co-Authored-By: Claude Fable 5 Signed-off-by: erni --- .../services/postgres_service.py | 10 ++- tests/test_integration_search.py | 74 +++++++++++++++++++ 2 files changed, 83 insertions(+), 1 deletion(-) 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..d31992da7 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,17 @@ 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_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" + ) From 3c31872fe39845ea302af419e05f950ca6167328 Mon Sep 17 00:00:00 2001 From: erni Date: Sun, 12 Jul 2026 07:43:35 +0300 Subject: [PATCH 5/5] style: apply ruff format to ingest_service and postgres_service Co-Authored-By: Claude Fable 5 Signed-off-by: erni --- core-api/src/core_api/services/ingest_service.py | 5 +---- .../src/core_storage_api/services/postgres_service.py | 4 +--- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/core-api/src/core_api/services/ingest_service.py b/core-api/src/core_api/services/ingest_service.py index 966c912a0..83608e33a 100644 --- a/core-api/src/core_api/services/ingest_service.py +++ b/core-api/src/core_api/services/ingest_service.py @@ -1071,10 +1071,7 @@ async def ingest_commit(request: IngestCommitRequest) -> dict: 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) - ] + 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, 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 d31992da7..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 @@ -1393,9 +1393,7 @@ async def memory_scored_search( # explicit-``status_filter`` override applies for inspection. scored_stmt = scored_stmt.where( or_( - Memory.status.notin_( - ("outdated", "conflicted", "archived", "cancelled") - ), + Memory.status.notin_(("outdated", "conflicted", "archived", "cancelled")), and_(Memory.status == "conflicted", _exact_lexical_match), ) )