"""Adversarial scale / pagination / concurrency tests for SibylStore. Lane: scale Target: data completeness at the _POOL enumeration bound, pagination math, multi-instance cross-visibility, WAL concurrency, resource leaks, max_depth edge cases, and Cap/Validation error propagation. History: the original adversarial pass found SILENT DATA LOSS at _POOL=1000 (browse / list_namespaces / _categories_under all truncated at 1000 with no signal). store.py was then fixed: * _POOL raised 1000 -> 10_000. * All three paths route through _list_capped(), which LOGS A WARNING when the result hits the cap (no longer fully silent). * Negative max_depth now raises ValueError. These tests now assert the CORRECTED behavior: * POSITIVE regression guards: at 1500 entities / 1100 namespaces (< cap), everything is returned, nothing dropped. * ONE xfail (strict=False) documents the RESIDUAL truncation above 10_000 (client MAX_LIMIT + no cursor — an architectural limit, not silent). * A caplog test proves the warning fires at the cap. """ from __future__ import annotations import asyncio import logging import os import tempfile import threading import time from typing import Any import pytest from sibyl_memory_client import CapExceededError, ValidationError as SibylValidationError from sibyl_memory_langgraph import SibylStore from sibyl_memory_langgraph.store import _POOL # bound under test (10_000 post-fix) from langgraph.store.base import ( GetOp, ListNamespacesOp, MatchCondition, PutOp, SearchOp, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _fresh_path() -> str: return os.path.join(tempfile.mkdtemp(), "t.db") def _new_store(path: str | None = None) -> SibylStore: if path is None: path = _fresh_path() return SibylStore(path=path, tier="free") def _put_op(ns: tuple, key: str, value: dict) -> PutOp: return PutOp(namespace=ns, key=key, value=value, index=None, ttl=None) def _seed_many(store: SibylStore, ns: tuple, n: int, *, batch_size: int = 500) -> None: """Insert n entities under one namespace via batched put ops (fast).""" ops: list[PutOp] = [] for i in range(n): ops.append(_put_op(ns, f"k{i:06d}", {"n": i})) if len(ops) >= batch_size: store.batch(ops) ops = [] if ops: store.batch(ops) def _search_op(prefix, *, query=None, filter=None, limit=10, offset=0) -> SearchOp: return SearchOp( namespace_prefix=prefix, filter=filter, limit=limit, offset=offset, query=query, refresh_ttl=True, ) def _ls_op( *, max_depth=None, limit=100, offset=0, prefix=None, suffix=None, ) -> ListNamespacesOp: conditions: list[MatchCondition] = [] if prefix is not None: conditions.append(MatchCondition(match_type="prefix", path=prefix)) if suffix is not None: conditions.append(MatchCondition(match_type="suffix", path=suffix)) return ListNamespacesOp( match_conditions=tuple(conditions) if conditions else None, max_depth=max_depth, limit=limit, offset=offset, ) # --------------------------------------------------------------------------- # 1. Browse completeness below the _POOL cap (regression guard for fix #1) # --------------------------------------------------------------------------- class TestBrowseFullResultsBelowCap: """REGRESSION GUARD (was: HIGH silent data loss at _POOL=1000). After the fix (_POOL=10_000), browsing 1500 entities in one namespace returns ALL 1500 — no truncation below the cap, no dropped rows. """ N = 1500 # < _POOL (10_000) NS = ("pool-browse",) @pytest.fixture(autouse=True) def _setup(self): assert _POOL > self.N, f"test assumes N({self.N}) < _POOL({_POOL})" path = _fresh_path() self.store = _new_store(path) _seed_many(self.store, self.NS, self.N) yield self.store.close() def test_browse_returns_all_entities_below_cap(self): """Browse (query=None) returns every entity when count < _POOL.""" results = self.store.search(self.NS, limit=self.N + 500) actual = len(results) assert actual == self.N, ( f"REGRESSION: browse at {self.N} (< _POOL={_POOL}) must return all; " f"got {actual} (dropped {self.N - actual})." ) def test_pagination_below_cap_is_correct(self): """offset=1000 on 1500 items now returns the remaining 500 (no dead zone).""" page_after_1000 = self.store.search(self.NS, limit=self.N, offset=1000) assert len(page_after_1000) == self.N - 1000, ( f"offset=1000 on {self.N} items should return {self.N - 1000}; " f"got {len(page_after_1000)} (this used to be a permanent dead zone)." ) full = self.store.search(self.NS, limit=self.N + 500) assert len(full) == self.N def test_full_pagination_union_covers_everything(self): """Stepping through pages of 250 must reach exactly N distinct keys.""" page_size = 250 seen: set[str] = set() off = 0 while True: page = self.store.search(self.NS, limit=page_size, offset=off) if not page: break seen.update(item.key for item in page) off += page_size if off > self.N + page_size: break assert len(seen) == self.N, ( f"Paginated union covers {len(seen)} keys, expected {self.N}." ) def test_subtree_browse_returns_all_below_cap(self): """Subtree browse via a parent prefix also returns all 1500.""" path = _fresh_path() s = _new_store(path) try: _seed_many(s, ("tree", "u1"), self.N) results = s.search(("tree",), limit=self.N + 500) assert len(results) == self.N, ( f"REGRESSION: subtree browse must return all {self.N}; " f"got {len(results)}." ) finally: s.close() # --------------------------------------------------------------------------- # 2. list_namespaces completeness below the cap (regression guard for fix #2) # --------------------------------------------------------------------------- class TestListNamespacesFullResultsBelowCap: """REGRESSION GUARD (was: HIGH silent data loss at _POOL=1000). 1100 distinct namespaces (< _POOL) are ALL present in list_namespaces(). """ N_NS = 1100 # < _POOL @pytest.fixture(autouse=True) def _setup(self): assert _POOL > self.N_NS, f"test assumes N_NS({self.N_NS}) < _POOL({_POOL})" path = _fresh_path() self.store = _new_store(path) ops = [_put_op(("ns", f"s{i:04d}"), "k", {"n": i}) for i in range(self.N_NS)] # batch in chunks for speed for start in range(0, len(ops), 500): self.store.batch(ops[start:start + 500]) yield self.store.close() def test_list_namespaces_returns_all_below_cap(self): result = self.store.list_namespaces(limit=self.N_NS + 200) actual = len(result) assert actual == self.N_NS, ( f"REGRESSION: {self.N_NS} distinct namespaces (< _POOL={_POOL}) must " f"all appear; got {actual} (dropped {self.N_NS - actual})." ) def test_list_namespaces_paginated_union_complete(self): seen: set[tuple] = set() for off in range(0, self.N_NS + 250, 250): page = self.store.list_namespaces(limit=250, offset=off) if not page: break seen.update(page) assert len(seen) == self.N_NS, ( f"Paginated list_namespaces union has {len(seen)}, expected {self.N_NS}." ) # --------------------------------------------------------------------------- # 3. FTS _categories_under completeness below the cap (regression guard #3) # --------------------------------------------------------------------------- class TestFTSCategoriesFullBelowCap: """REGRESSION GUARD (was: MEDIUM silent data loss — oldest categories evicted from the 1000-row pool, FTS returned 0 instead of 50). With _POOL=10_000 and 1050 total categories, the keyword entities inserted FIRST (the oldest) are still inside the pool, so FTS finds all 50. """ def test_fts_finds_all_matching_categories_below_cap(self): path = _fresh_path() s = _new_store(path) try: # 50 keyword entities FIRST (the oldest) ... for i in range(50): s.put((f"cat{i:04d}",), "k", {"text": "zzzunique"}) # ... then 1000 ordinary ones. Total 1050 << _POOL, so nothing evicted. for i in range(50, 1050): s.put((f"cat{i:04d}",), "k", {"text": "ordinary"}) results = s.search((), query="zzzunique", limit=100) assert len(results) == 50, ( f"REGRESSION: FTS should find all 50 'zzzunique' categories " f"(1050 total < _POOL={_POOL}); got {len(results)}. The oldest " f"categories used to be evicted from the 1000-row pool." ) finally: s.close() # --------------------------------------------------------------------------- # 4. Residual truncation ABOVE the cap + warning signal # # NOTE on methodology: the 2 MB free-tier cap makes >10_000 real rows # IMPOSSIBLE to insert (the cap is on the true DB footprint, page_count * # page_size, which is exhausted at ~3,900 entities even with empty bodies; a # paid/uncapped tier needs offline-unavailable server verification). So the # only feasible way to exercise the adapter's >_POOL behavior on the free tier # is to stub the client's data source (list_entities) while keeping ALL the # real adapter code: _list_capped()'s cap detection + warning, the prefix # filter, and the offset/limit slicing. This isolates exactly the adapter logic # the fix changed. # --------------------------------------------------------------------------- def _synthetic_rows(n: int, category: str = "big") -> list[dict[str, Any]]: """n entity rows shaped like the real client's list_entities output.""" return [ { "id": f"id{i:07d}", "tenant_id": "t", "category": category, "name": f"k{i:07d}", "status": None, "body": {"n": i}, "created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-01T00:00:00Z", } for i in range(n) ] def _stub_list_entities(store: SibylStore, rows: list[dict[str, Any]]) -> None: """Replace store._client.list_entities with one that clamps to `limit`, exactly like the real client's MAX_LIMIT clamp (_clamp_limit -> 10_000). """ def fake(category: str | None = None, *, status: str | None = None, limit: int = 100): scoped = rows if category is None else [r for r in rows if r["category"] == category] return scoped[:limit] # mimic the client clamp; >limit rows are dropped here store._client.list_entities = fake # type: ignore[attr-defined] @pytest.fixture() def store_over_pool(): """Real SibylStore whose data source reports _POOL+50 rows in one namespace.""" n = _POOL + 50 path = _fresh_path() s = _new_store(path) _stub_list_entities(s, _synthetic_rows(n, category="big")) try: yield s, n finally: s.close() @pytest.mark.xfail( reason="architectural: client MAX_LIMIT=10_000 + no cursor; full fix needs a " "client-side enumeration API — pending operator decision", strict=False, ) def test_browse_over_pool_still_truncates(store_over_pool): """RESIDUAL LIMIT — above _POOL the browse pool is still bounded. With _POOL+50 rows available, browsing returns only _POOL. This is no longer SILENT (a warning is logged — see test_enumeration_warns_at_cap), but the LangGraph return type carries no has_more flag, so rows past the cap are unreachable in one pass. xfail(strict=False): documents the residual hole. """ s, n = store_over_pool results = s.search(("big",), limit=n + 1000) assert len(results) == n, ( f"residual truncation: {n} rows available (> _POOL={_POOL}), " f"browse returned {len(results)}; {n - len(results)} unreachable in one pass." ) def test_enumeration_warns_at_cap(store_over_pool, caplog): """The fix's key improvement: hitting the enumeration cap LOGS A WARNING (no longer fully silent), even though the return type has no has_more flag. """ s, _n = store_over_pool caplog.clear() with caplog.at_level(logging.WARNING, logger="sibyl_memory_langgraph.store"): s.search(("big",), limit=10) # browse path -> _list_capped() hits the cap warned = [r for r in caplog.records if r.levelno >= logging.WARNING] assert warned, ( "Expected a WARNING when enumeration hits the _POOL cap; none logged. " "The fix is supposed to make truncation non-silent." ) joined = " ".join(r.getMessage().lower() for r in warned) assert "cap" in joined or str(_POOL) in joined, ( f"Warning fired but did not mention the cap: {[r.getMessage() for r in warned]}" ) def test_list_namespaces_warns_at_cap(store_over_pool, caplog): """list_namespaces shares the _list_capped() path, so it warns at the cap too.""" s, _n = store_over_pool caplog.clear() with caplog.at_level(logging.WARNING, logger="sibyl_memory_langgraph.store"): s.list_namespaces(limit=10) assert any(r.levelno >= logging.WARNING for r in caplog.records), ( "Expected a WARNING from list_namespaces when enumeration hits the cap." ) def test_below_pool_does_not_warn(caplog): """Negative control: at well under _POOL rows, NO cap warning is logged.""" path = _fresh_path() s = _new_store(path) try: _stub_list_entities(s, _synthetic_rows(100, category="big")) caplog.clear() with caplog.at_level(logging.WARNING, logger="sibyl_memory_langgraph.store"): s.search(("big",), limit=10) s.list_namespaces(limit=10) assert not [r for r in caplog.records if r.levelno >= logging.WARNING], ( "A cap warning fired below _POOL — the cap detection is too eager." ) finally: s.close() # --------------------------------------------------------------------------- # 5. Pagination math edge cases (unchanged — all passing) # --------------------------------------------------------------------------- class TestPaginationEdgeCases: N = 12 # small, deterministic @pytest.fixture() def store_with_data(self): path = _fresh_path() s = _new_store(path) ns = ("pag",) ops = [_put_op(ns, f"k{i:02d}", {"n": i}) for i in range(self.N)] s.batch(ops) try: yield s finally: s.close() def test_limit_zero_returns_empty_not_crash(self, store_with_data): """limit=0 must not crash; must return empty list.""" try: results = store_with_data.search(("pag",), limit=0, offset=0) assert len(results) == 0, f"limit=0 returned {len(results)} items" except (ValueError, TypeError): pass # also acceptable def test_limit_larger_than_total_returns_all(self, store_with_data): """limit >> N should return exactly N items without error.""" results = store_with_data.search(("pag",), limit=100_000, offset=0) assert len(results) == self.N, ( f"limit=100000 returned {len(results)}, expected {self.N}" ) def test_offset_beyond_end_returns_empty(self, store_with_data): """offset > N should return empty without wrapping or crashing.""" results = store_with_data.search(("pag",), limit=10, offset=self.N + 100) assert len(results) == 0, ( f"offset past end returned {len(results)} items (expected 0)" ) def test_offset_exactly_at_end_returns_empty(self, store_with_data): """offset == N (one past last item) should return empty.""" results = store_with_data.search(("pag",), limit=10, offset=self.N) assert len(results) == 0 def test_negative_limit_does_not_return_unbounded_set(self, store_with_data): """Negative limit must not silently return all rows (unbounded scan). Python slice [::-1] with a negative limit could theoretically reverse or do surprising things; the client clamps negative limits. We verify the result is not larger than N and no unchecked exception escapes. """ try: results = store_with_data.search(("pag",), limit=-1, offset=0) assert len(results) <= self.N, ( f"Negative limit=-1 returned {len(results)} items > N={self.N} — " f"possible unbounded scan." ) except (ValueError, TypeError): pass # ideal: reject negative limits explicitly def test_negative_offset_does_not_wrap_or_crash(self, store_with_data): """Negative offset — Python slicing wraps around; should raise or return <=N items. `rows[-2:-2+limit]` for large lists returns wrong results (not a tail-anchor window). We assert no crash and result count <= N. """ try: results = store_with_data.search(("pag",), limit=5, offset=-1) # Document actual count — likely 0 or 1 (wrong), never an error assert len(results) <= self.N, ( f"Negative offset=-1 returned {len(results)} items > N" ) except (ValueError, TypeError): pass # ideal: reject negative offsets def test_full_pagination_covers_all_items(self, store_with_data): """Step through all pages; union must equal the full item set.""" page_size = 5 seen_keys: set[str] = set() off = 0 while True: page = store_with_data.search(("pag",), limit=page_size, offset=off) if not page: break for item in page: seen_keys.add(item.key) off += page_size if off > self.N + page_size: break assert len(seen_keys) == self.N, ( f"Paginated union covers {len(seen_keys)} items, expected {self.N}. " f"Possible off-by-one or pagination gap." ) def test_list_namespaces_offset_beyond_end_returns_empty(self): """list_namespaces: offset past end returns empty without wrapping.""" path = _fresh_path() s = _new_store(path) try: for i in range(5): s.put((f"ns{i}",), "k", {"n": i}) result = s.list_namespaces(limit=10, offset=1000) assert len(result) == 0, ( f"offset=1000 (well past 5 namespaces) returned {len(result)} items" ) finally: s.close() def test_list_namespaces_paginated_union_is_complete(self): """Paginating through list_namespaces must cover every namespace.""" path = _fresh_path() s = _new_store(path) try: for i in range(20): s.put((f"ns{i:02d}",), "k", {"n": i}) seen: set[tuple] = set() for off in range(0, 25, 5): page = s.list_namespaces(limit=5, offset=off) seen.update(page) assert len(seen) == 20, ( f"Paginated list_namespaces union has {len(seen)} namespaces, expected 20." ) finally: s.close() # --------------------------------------------------------------------------- # 6. max_depth edge cases # --------------------------------------------------------------------------- class TestMaxDepthEdgeCases: def test_max_depth_zero_truncates_to_empty_tuple(self): """max_depth=0 truncates every namespace to () — should produce [()] after dedup, or nothing. Must not crash with IndexError. """ path = _fresh_path() s = _new_store(path) try: s.put(("a", "b"), "k", {"x": 1}) s.put(("c",), "k", {"x": 2}) try: result = s.list_namespaces(max_depth=0) # All namespaces truncate to (); deduplication leaves [()] for ns in result: assert len(ns) == 0, f"max_depth=0 gave non-empty tuple: {ns}" except (ValueError, TypeError, IndexError) as e: pytest.fail(f"max_depth=0 raised unexpected exception: {type(e).__name__}: {e}") finally: s.close() def test_max_depth_exceeds_deepest_namespace_returns_full(self): """max_depth >> deepest depth: ns[:very_large] = ns — no padding, no crash.""" path = _fresh_path() s = _new_store(path) try: s.put(("a",), "k", {"x": 1}) s.put(("a", "b", "c"), "k", {"x": 2}) result = s.list_namespaces(max_depth=999) ns_set = set(result) assert ("a",) in ns_set assert ("a", "b", "c") in ns_set finally: s.close() def test_negative_max_depth_raises_value_error(self): """FIXED (#4): negative max_depth now raises ValueError instead of silently truncating the last namespace element via Python's ns[:-1]. """ path = _fresh_path() s = _new_store(path) try: s.put(("a", "b", "c"), "k", {"x": 1}) with pytest.raises(ValueError): s.list_namespaces(max_depth=-1) # a more-negative value must also raise with pytest.raises(ValueError): s.list_namespaces(max_depth=-5) finally: s.close() # --------------------------------------------------------------------------- # 7. Multi-instance cross-visibility (same DB, WAL) # --------------------------------------------------------------------------- class TestMultiInstance: """Two SibylStore objects on the same file path. WAL mode means readers never block writers and commits are visible immediately. """ @pytest.fixture() def shared_path(self) -> str: return _fresh_path() def test_write_on_instance1_visible_to_instance2(self, shared_path): s1 = _new_store(shared_path) s2 = _new_store(shared_path) try: s1.put(("shared",), "k1", {"msg": "from s1"}) item = s2.get(("shared",), "k1") assert item is not None, "Instance 2 could not see write from instance 1" assert item.value == {"msg": "from s1"} finally: s1.close(); s2.close() def test_delete_on_instance1_visible_to_instance2(self, shared_path): s1 = _new_store(shared_path) s2 = _new_store(shared_path) try: s1.put(("shared",), "k1", {"x": 1}) assert s2.get(("shared",), "k1") is not None s1.delete(("shared",), "k1") assert s2.get(("shared",), "k1") is None, ( "Instance 2 still sees item deleted by instance 1" ) finally: s1.close(); s2.close() def test_overwrite_on_instance1_not_stale_on_instance2(self, shared_path): s1 = _new_store(shared_path) s2 = _new_store(shared_path) try: s1.put(("shared",), "k1", {"v": 1}) s1.put(("shared",), "k1", {"v": 2}) item = s2.get(("shared",), "k1") assert item is not None assert item.value == {"v": 2}, ( f"Instance 2 got stale value {item.value!r}, expected {{'v': 2}}" ) finally: s1.close(); s2.close() def test_concurrent_writes_both_instances_no_corruption(self, shared_path): """50 writes from each instance concurrently — WAL + busy_timeout=5000ms should prevent data corruption or deadlock. """ s1 = _new_store(shared_path) s2 = _new_store(shared_path) errors: list[str] = [] def write_s1(): for i in range(50): try: s1.put(("conc",), f"s1_{i:02d}", {"v": i}) except Exception as e: errors.append(f"s1 write {i}: {type(e).__name__}: {e}") def write_s2(): for i in range(50): try: s2.put(("conc",), f"s2_{i:02d}", {"v": i}) except Exception as e: errors.append(f"s2 write {i}: {type(e).__name__}: {e}") t1 = threading.Thread(target=write_s1, daemon=True) t2 = threading.Thread(target=write_s2, daemon=True) t1.start(); t2.start() t1.join(timeout=15); t2.join(timeout=15) try: assert not errors, f"Concurrent writes from two instances produced errors:\n" + "\n".join(errors) total = s1.search(("conc",), limit=200) assert len(total) == 100, ( f"Expected 100 items after concurrent writes from 2 instances, got {len(total)}" ) finally: s1.close(); s2.close() def test_concurrent_abatch_same_store_no_deadlock(self, shared_path): """Multiple asyncio tasks calling abatch() on the same store instance.""" s = _new_store(shared_path) try: for i in range(10): s.put(("ab",), f"k{i}", {"n": i}) async def run(): op = SearchOp( namespace_prefix=("ab",), filter=None, limit=10, offset=0, query=None, refresh_ttl=True, ) tasks = [s.abatch([op]) for _ in range(8)] return await asyncio.gather(*tasks) results = asyncio.run(run()) assert len(results) == 8 for r in results: assert len(r[0]) == 10 finally: s.close() # --------------------------------------------------------------------------- # 8. Resource / connection leaks # --------------------------------------------------------------------------- def test_open_close_many_stores_no_exception(): """Open and close 60 separate SibylStore instances — each should open cleanly and release without OS errors. Uses a shared DB to exercise WAL contention in the open/close cycle. """ path = _fresh_path() # Seed some data s0 = _new_store(path) s0.put(("leak",), "k", {"x": 1}) s0.close() for i in range(60): s = _new_store(path) try: item = s.get(("leak",), "k") assert item is not None, f"Iteration {i}: data lost after open" finally: s.close() # Final sanity: one more open should still work s_final = _new_store(path) try: assert s_final.get(("leak",), "k") is not None finally: s_final.close() def test_abatch_worker_threads_close_cleanly(): """abatch() dispatches to run_in_executor (thread pool). Many sequential abatch calls must not exhaust file descriptors or leave zombie threads. """ path = _fresh_path() s = _new_store(path) try: for i in range(5): s.put(("ab",), f"k{i}", {"n": i}) async def run_many(): op = SearchOp( namespace_prefix=("ab",), filter=None, limit=5, offset=0, query=None, refresh_ttl=True, ) for _ in range(20): await s.abatch([op]) asyncio.run(run_many()) finally: s.close() # --------------------------------------------------------------------------- # 9. Cap / Validation error surface # --------------------------------------------------------------------------- def test_cap_exceeded_error_propagates_not_swallowed(): """Writes past the 2MB free-tier cap must raise CapExceededError — not silently succeed, not crash with an internal SQLite or StorageError, and not corrupt the DB (existing data must still be readable after the error). """ path = _fresh_path() s = _new_store(path) try: # ~250 KB per write; 2MB / 250KB ≈ 8 writes before cap. # We allow up to 12 writes and assert cap is hit before 12. payload = {"data": "x" * 250_000} cap_hit_at: int | None = None for i in range(12): try: s.put(("cap",), f"large{i}", payload) except CapExceededError: cap_hit_at = i break except Exception as e: pytest.fail( f"Unexpected exception type at write {i}: " f"{type(e).__name__}: {e}" ) assert cap_hit_at is not None, ( "Expected CapExceededError before 12 × 250KB writes (3MB > 2MB cap). " "Cap may not be enforced, or the error is swallowed inside SibylStore." ) # Existing data must survive the cap hit item = s.get(("cap",), "large0") assert item is not None, ( "Entity written before cap hit is gone after CapExceededError — " "possible DB corruption." ) finally: s.close() def test_single_value_over_per_value_limit_raises(): """A single value exceeding the per-value 1024 KB limit raises SibylValidationError. The adapter has two independent size gates: - Per-value: 1024 KB max body → ValidationError - Total DB: 2 MB free-tier → CapExceededError A 1.5MB value hits the per-value limit first and must raise ValidationError (not crash silently, not corrupt the DB). """ path = _fresh_path() s = _new_store(path) try: # 1.5 MB — above the 1024 KB per-value limit, below the 2MB DB cap large_payload = {"data": "z" * 1_500_000} with pytest.raises(SibylValidationError): s.put(("huge",), "single", large_payload) # DB must still be usable after the rejection s.put(("huge",), "small_ok", {"ok": True}) assert s.get(("huge",), "small_ok") is not None finally: s.close() def test_cap_exceeded_then_delete_allows_new_write(): """After hitting the cap, deleting items should allow new writes to succeed (cap gate re-evaluates committed size, not a latching error). """ path = _fresh_path() s = _new_store(path) try: payload = {"data": "x" * 250_000} keys_written = [] # Fill to cap for i in range(12): try: s.put(("cap2",), f"k{i}", payload) keys_written.append(f"k{i}") except CapExceededError: break assert keys_written, "Should have written at least one entity before cap" # Delete all written keys to free up space for k in keys_written: s.delete(("cap2",), k) # Now a new write should succeed (cap freed) try: s.put(("cap2",), "new_after_delete", {"small": "value"}) except CapExceededError: pytest.fail( "CapExceededError after deleting all prior entities — " "cap gate does not re-evaluate freed space." ) finally: s.close()