feat: bulk= keyword param on ingest_turn (#194)

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

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