From 56cad93a386c06757d6cd1c330e337fb9f85f039 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 28 Apr 2026 17:35:06 -0700 Subject: [PATCH] feat: add bulk= keyword param to ingest_turn (#194) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Additive, keyword-only `bulk: bool = False` on `ingest_turn` for research-line / batch ingest callers (e.g. wonder_ingest). When True, duplicate-sentence corroboration writes are skipped — the only currently-redundant per-turn side effect. Canonical beliefs row and v2.0 ingest_log row always produced; final belief state is identical either way. Default-off preserves prior behaviour for all existing callers. Closes #194. --- src/aelfrice/ingest.py | 30 ++++++++++-- tests/test_ingest_bulk.py | 99 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 124 insertions(+), 5 deletions(-) create mode 100644 tests/test_ingest_bulk.py diff --git a/src/aelfrice/ingest.py b/src/aelfrice/ingest.py index 868f2103e..f2873127d 100644 --- a/src/aelfrice/ingest.py +++ b/src/aelfrice/ingest.py @@ -63,6 +63,8 @@ def ingest_turn( session_id: str | None = None, created_at: str | None = None, source_id: str = "", # noqa: ARG001 + *, + bulk: bool = False, ) -> int: """Ingest a single conversation turn. @@ -85,12 +87,24 @@ def ingest_turn( `source_id` is still accepted for adapter parity but is not yet persisted. + `bulk` (v1.6+, keyword-only) signals the call is part of a + multi-turn batch (e.g. a research-line `wonder_ingest` pipeline). + When True, per-turn side effects that are redundant inside a batch + are skipped: currently `store.record_corroboration` for duplicate + sentences. The canonical `beliefs` row and the `ingest_log` write + (the v2.0 source-of-truth ingest record) are always produced — + `bulk=True` produces the same final belief state as `bulk=False` + for the same input sequence; only the corroboration audit table + differs. Reserved for future per-turn-cost reductions as new side + effects land. Default-off preserves current behavior. + Returns the number of beliefs inserted (or that would have been inserted if not already present). """ return len(_ingest_turn_ids( store=store, text=text, source=source, session_id=session_id, created_at=created_at, + bulk=bulk, )) @@ -100,6 +114,8 @@ def _ingest_turn_ids( source: str, session_id: str | None = None, created_at: str | None = None, + *, + bulk: bool = False, ) -> list[str]: """Internal variant of ingest_turn returning the inserted belief ids. @@ -122,11 +138,15 @@ def _ingest_turn_ids( if store.get_belief(belief_id) is not None: # Exact (source, sentence) duplicate: record a corroboration # so re-assertions are observable. Canonical row unchanged. - store.record_corroboration( - belief_id, - source_type=CORROBORATION_SOURCE_TRANSCRIPT_INGEST, - session_id=session_id, - ) + # bulk=True skips this — corroboration noise is uninteresting + # in batch ingest and is the only currently-redundant per-turn + # side effect. + if not bulk: + store.record_corroboration( + belief_id, + source_type=CORROBORATION_SOURCE_TRANSCRIPT_INGEST, + session_id=session_id, + ) continue # v2.0 #205 parallel-write: log the classifier input before # materializing the belief. belief_id is deterministic on diff --git a/tests/test_ingest_bulk.py b/tests/test_ingest_bulk.py new file mode 100644 index 000000000..8f7d3e935 --- /dev/null +++ b/tests/test_ingest_bulk.py @@ -0,0 +1,99 @@ +"""Tests for the `bulk=` parameter on `ingest_turn` (issue #194). + +`bulk=True` is a research-line / batch-ingest hint. Per-turn side +effects that are redundant inside a multi-turn batch are skipped: +currently `store.record_corroboration` for duplicate sentences. The +canonical `beliefs` row and the v2.0 `ingest_log` write are always +produced — `bulk=True` and `bulk=False` produce the same final belief +state; only the corroboration audit table differs. + +In-memory store, deterministic, no LLM/IO. +""" +from __future__ import annotations + +from aelfrice.ingest import ingest_turn +from aelfrice.store import MemoryStore + + +def _fresh_store() -> MemoryStore: + return MemoryStore(":memory:") + + +def test_bulk_default_is_false() -> None: + """`bulk` defaults to False — preserves prior behaviour for all + existing callers (signature is keyword-only, additive). + """ + store = _fresh_store() + text = "The sky is blue. Water is wet." + n = ingest_turn(store, text=text, source="user") + assert n >= 1 + + +def test_bulk_keyword_only() -> None: + """`bulk` is keyword-only — positional pass should fail. Guards + against a future arg being inserted before it. + """ + store = _fresh_store() + import pytest + + with pytest.raises(TypeError): + # 7 positional args (bulk would be the 7th) — should be rejected + # because bulk is keyword-only. + ingest_turn(store, "Hello world.", "user", None, None, "", True) # type: ignore[misc] + + +def test_bulk_unique_sentences_same_final_belief_state() -> None: + """For unique-sentence input, bulk=True and bulk=False produce + identical final belief state. (No duplicates means no corroboration + writes either way — this is a sanity baseline.) + """ + text = "The sky is blue. Water is wet. Fire is hot." + + store_bulk = _fresh_store() + store_normal = _fresh_store() + n_bulk = ingest_turn(store_bulk, text=text, source="user", bulk=True) + n_normal = ingest_turn(store_normal, text=text, source="user", bulk=False) + + assert n_bulk == n_normal + assert sorted(store_bulk.list_belief_ids()) == sorted(store_normal.list_belief_ids()) + + +def test_bulk_skips_corroboration_on_duplicates() -> None: + """Re-ingesting a duplicate sentence in bulk mode does NOT record a + corroboration. In non-bulk mode it does. Belief table is identical + in both cases. + """ + text = "The sky is blue." + source = "user" + + store_bulk = _fresh_store() + ingest_turn(store_bulk, text=text, source=source, bulk=True) + ingest_turn(store_bulk, text=text, source=source, bulk=True) + ingest_turn(store_bulk, text=text, source=source, bulk=True) + + store_normal = _fresh_store() + ingest_turn(store_normal, text=text, source=source, bulk=False) + ingest_turn(store_normal, text=text, source=source, bulk=False) + ingest_turn(store_normal, text=text, source=source, bulk=False) + + bulk_ids = sorted(store_bulk.list_belief_ids()) + normal_ids = sorted(store_normal.list_belief_ids()) + assert bulk_ids == normal_ids + assert len(bulk_ids) == 1 + + bid = bulk_ids[0] + # bulk=False records a corroboration on each duplicate (2 corroborations) + assert store_normal.count_corroborations(bid) == 2 + # bulk=True records none + assert store_bulk.count_corroborations(bid) == 0 + + +def test_bulk_writes_ingest_log() -> None: + """bulk=True must still write the v2.0 ingest_log row — that is the + source-of-truth record for the parallel-write phase, NOT a per-turn + side effect to skip. + """ + store = _fresh_store() + n = ingest_turn(store, text="The sky is blue.", source="user", bulk=True) + assert n == 1 + assert store.count_ingest_log() == 1