diff --git a/agent_memory_leaderboard/README.md b/agent_memory_leaderboard/README.md index 08f1133..03f06ec 100644 --- a/agent_memory_leaderboard/README.md +++ b/agent_memory_leaderboard/README.md @@ -219,6 +219,35 @@ outlasts that short timeout still reports not-ready; BUSY is not cached as healt This checks readability, not durable-write capacity. The independent upstream health probe remains unchanged. +### Durable transaction boundaries + +The Add path combines request registration and the first atomic send intent in +one transaction. Each subsequent chunk also validates its state and persists +any required send intent in one journal operation. Already-inflight atomic +receipts can be replayed without rewriting identical intent state. Unknown +legacy outcomes retain the existing query-only recovery policy. + +The backend acknowledgment (memory ID) is still committed before visibility +verification. Once the final chunk is verified, its stored marker and request +completion are committed together. A new one-chunk Add therefore uses three +durable commits instead of five; a new N-chunk Add uses 3N instead of 3N+2. +Completed replays perform no write transaction. Legacy journals whose last +chunk is stored but whose request is incomplete still complete on retry. + +Run the isolated synthetic benchmark with +`uv run python scripts/benchmark_journal.py`. Use `--commit-delay-ms 10` to +inject a fixed delay per commit, or `--source-root /path/to/checkout/agent_memory_leaderboard` +to compare another checkout with the same harness. It uses temporary storage +and an in-memory synthetic backend, never calls a service, and reports only +aggregate counts and timings. It measures explicit `db.commit()` calls; an +implicit SQLite context-manager commit aborts the run instead of producing an +undercounted result. Its results are not deployment capacity estimates. + +These changes retain `synchronous=FULL`, the existing journal mode and schema, +and the per-request and cross-process locks. No SQLite transaction or global +journal gate spans a backend HTTP call. This reduces commit overhead but does +not remove SQLite's single-writer limit or establish deployment throughput. + ### Existing journals and rollback Startup upgrades the four-column `writes` table transactionally, adding a diff --git a/agent_memory_leaderboard/scripts/benchmark_journal.py b/agent_memory_leaderboard/scripts/benchmark_journal.py new file mode 100644 index 0000000..225074c --- /dev/null +++ b/agent_memory_leaderboard/scripts/benchmark_journal.py @@ -0,0 +1,146 @@ +"""Isolated journal benchmark with synthetic data; never contacts a backend. + +Run with `uv run python scripts/benchmark_journal.py`. An optional injected +commit delay models slow storage but is not a measurement of any deployment. +Use --source-root to run this same harness against another source checkout. +""" + +import argparse +import json +import logging +import math +import sqlite3 +import sys +import tempfile +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace + + +class UnsupportedCommitError(RuntimeError): + """The source checkout uses a commit path this harness cannot measure.""" + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--source-root", type=Path, default=Path(__file__).resolve().parents[1]) + parser.add_argument("--requests", type=int, default=64) + parser.add_argument("--concurrency", type=int, nargs="+", default=[1, 16, 32, 64]) + parser.add_argument("--commit-delay-ms", type=float, default=0) + args = parser.parse_args() + if ( + args.requests < 1 + or any(c < 1 for c in args.concurrency) + or not math.isfinite(args.commit_delay_ms) + or args.commit_delay_ms < 0 + ): + parser.error("counts must be positive and commit delay finite and nonnegative") + sys.path.insert(0, str(args.source_root)) + from src.config import Settings + from src.errors import AdapterError + from src.journal import Journal + from src.models import AddRequest + from src.service import MemoryAdapter + + logging.disable(logging.CRITICAL) + + class Backend: + def __init__(self): + self.lock = threading.Lock() + self.receipts = {} + + def store_idempotent(self, owner, body, marker): + with self.lock: + return self.receipts.setdefault((owner, marker), SimpleNamespace(memory_id=marker)) + + def wait_for_visibility(self, owner, memory_id, marker): + with self.lock: + memory = self.receipts[owner, marker] + assert memory.memory_id == memory_id + return memory + + original_connect = sqlite3.connect + for concurrency in args.concurrency: + with tempfile.TemporaryDirectory(prefix="journal-benchmark-") as directory: + settings = Settings( + memoria_url="http://synthetic.invalid", + memoria_master_key="synthetic", + state_dir=Path(directory), + ) + journal = Journal(settings.state_dir) + backend = Backend() + adapter = MemoryAdapter(settings, backend, journal) + commits = 0 + commit_lock = threading.Lock() + + class Connection(sqlite3.Connection): + def __exit__(self, exc_type, exc_value, traceback): + if exc_type is None and self.in_transaction: + # sqlite3's context-manager commit bypasses this subclass's + # commit() override, so neither its count nor delay is valid. + self.rollback() + raise UnsupportedCommitError( + "Benchmark aborted: implicit context-manager commits are not " + "measured. Use a source checkout with explicit db.commit() calls." + ) + return super().__exit__(exc_type, exc_value, traceback) + + def commit(self): + nonlocal commits + if args.commit_delay_ms: + time.sleep(args.commit_delay_ms / 1000) + super().commit() + with commit_lock: + commits += 1 + + def connect(*a, **kw): + return original_connect(*a, factory=Connection, **kw) + + def add(index): + request = AddRequest( + request_id=f"request-{index}", + user_id=f"user-{index}", + session_id="session", + messages=[{"role": "user", "content": "A synthetic fact for journal timing."}], + ) + started = time.perf_counter() + status = 200 + try: + adapter.add(request) + except AdapterError as exc: + status = exc.status + return status, (time.perf_counter() - started) * 1000 + + sqlite3.connect = connect + started = time.perf_counter() + try: + with ThreadPoolExecutor(max_workers=concurrency) as pool: + results = list(pool.map(add, range(args.requests))) + except UnsupportedCommitError as exc: + parser.exit(1, f"{exc}\n") + finally: + elapsed = time.perf_counter() - started + sqlite3.connect = original_connect + latencies = sorted(ms for _, ms in results) + statuses = {str(s): sum(code == s for code, _ in results) for s, _ in results} + print( + json.dumps( + { + "concurrency": concurrency, + "requests": args.requests, + "injected_commit_delay_ms": args.commit_delay_ms, + "statuses": statuses, + "commits": commits, + "unique_backend_records": len(backend.receipts), + "success_rps": round(statuses.get("200", 0) / elapsed, 3), + "p95_ms": round(latencies[math.ceil(len(latencies) * 0.95) - 1], 3), + } + ), + flush=True, + ) + + +if __name__ == "__main__": + main() diff --git a/agent_memory_leaderboard/src/journal.py b/agent_memory_leaderboard/src/journal.py index 5b726a5..a667666 100644 --- a/agent_memory_leaderboard/src/journal.py +++ b/agent_memory_leaderboard/src/journal.py @@ -193,30 +193,71 @@ def lock(self, key: str): timing.error_category = "lock_io_error" raise + @staticmethod + def _begin(db, key: str, fingerprint: str) -> bool: + row = db.execute( + "SELECT fingerprint, complete FROM requests WHERE key = ?", (key,) + ).fetchone() + if row: + if row[0] != fingerprint: + raise AdapterError( + 400, + "request_id was already used with a different payload; " + "restore the original payload", + ) + return bool(row[1]) + db.execute("INSERT INTO requests(key, fingerprint) VALUES (?, ?)", (key, fingerprint)) + return False + def begin(self, key: str, fingerprint: str) -> bool: with self.connect("begin") as db: - row = db.execute( - "SELECT fingerprint, complete FROM requests WHERE key = ?", (key,) - ).fetchone() - if row: - if row[0] != fingerprint: - raise AdapterError( - 400, - "request_id was already used with a different payload; " - "restore the original payload", - ) - return bool(row[1]) - db.execute("INSERT INTO requests(key, fingerprint) VALUES (?, ?)", (key, fingerprint)) - return False + return self._begin(db, key, fingerprint) + + @staticmethod + def _state(db, key: str, position: int) -> tuple[str, str | None, str]: + row = db.execute( + "SELECT state, memory_id, protocol FROM writes WHERE request_key = ? AND position = ?", + (key, position), + ).fetchone() + return row if row else ("pending", None, IDEMPOTENT_PROTOCOL) def state(self, key: str, position: int) -> tuple[str, str | None, str]: with self.connect("state") as db: - row = db.execute( - "SELECT state, memory_id, protocol FROM writes " - "WHERE request_key = ? AND position = ?", - (key, position), - ).fetchone() - return row if row else ("pending", None, IDEMPOTENT_PROTOCOL) + return self._state(db, key, position) + + @staticmethod + def _record(db, key, position, state, memory_id, protocol): + db.execute( + "INSERT INTO writes(request_key, position, state, memory_id, protocol) " + "VALUES (?, ?, ?, ?, ?) " + "ON CONFLICT(request_key, position) DO UPDATE SET " + "state = excluded.state, memory_id = excluded.memory_id, " + "protocol = excluded.protocol", + (key, position, state, memory_id, protocol), + ) + + def prepare_write( + self, key: str, fingerprint: str, position: int + ) -> tuple[str, str | None, str] | None: + """Register the request and persist send intent in one durable transaction. + + Caller holds lock(key) throughout Add. None denotes a completed replay. + Legacy unknown outcomes must never be converted to atomic receipts. + """ + with self.connect("prepare_write") as db: + if self._begin(db, key, fingerprint): + return None + state, memory_id, protocol = self._state(db, key, position) + if protocol not in {LEGACY_PROTOCOL, IDEMPOTENT_PROTOCOL}: + raise AdapterError(503, "Unsupported journal write protocol; check adapter version") + if state not in {"pending", "inflight", "stored"}: + raise AdapterError(503, "Unsupported journal write state; check adapter version") + if state == "pending": + # Pending legacy writes are known not to have been sent. Persist + # the atomic protocol before HTTP, including on the first chunk. + state, memory_id, protocol = "inflight", None, IDEMPOTENT_PROTOCOL + self._record(db, key, position, state, memory_id, protocol) + return state, memory_id, protocol def record( self, @@ -226,16 +267,14 @@ def record( memory_id: str | None = None, *, protocol: str, + complete_request: bool = False, ) -> None: + if complete_request and state != "stored": + raise ValueError("Only a verified stored write can complete a request") with self.connect("record") as db: - db.execute( - "INSERT INTO writes(request_key, position, state, memory_id, protocol) " - "VALUES (?, ?, ?, ?, ?) " - "ON CONFLICT(request_key, position) DO UPDATE SET " - "state = excluded.state, memory_id = excluded.memory_id, " - "protocol = excluded.protocol", - (key, position, state, memory_id, protocol), - ) + self._record(db, key, position, state, memory_id, protocol) + if complete_request: + db.execute("UPDATE requests SET complete = 1 WHERE key = ?", (key,)) def complete(self, key: str) -> None: with self.connect("complete") as db: diff --git a/agent_memory_leaderboard/src/journal_metrics.py b/agent_memory_leaderboard/src/journal_metrics.py index 81b191b..d334411 100644 --- a/agent_memory_leaderboard/src/journal_metrics.py +++ b/agent_memory_leaderboard/src/journal_metrics.py @@ -15,7 +15,17 @@ # Uvicorn configures this logger at INFO in the adapter's existing entrypoint. LOGGER = logging.getLogger("uvicorn.error") OPERATIONS = frozenset( - {"initialize", "probe", "begin", "state", "record", "complete", "lock_acquire", "lock_release"} + { + "initialize", + "probe", + "begin", + "state", + "prepare_write", + "record", + "complete", + "lock_acquire", + "lock_release", + } ) diff --git a/agent_memory_leaderboard/src/service.py b/agent_memory_leaderboard/src/service.py index d4344e5..d9ba92f 100644 --- a/agent_memory_leaderboard/src/service.py +++ b/agent_memory_leaderboard/src/service.py @@ -1,10 +1,11 @@ import hashlib import json +from itertools import chain, pairwise from .config import Settings from .errors import AdapterError from .ingestion import message_parts -from .journal import IDEMPOTENT_PROTOCOL, LEGACY_PROTOCOL, Journal +from .journal import LEGACY_PROTOCOL, Journal from .memoria import Memoria from .models import AddRequest, AddResponse, Evidence, SearchRequest, SearchResponse, text_content @@ -32,10 +33,14 @@ def add(self, request: AddRequest) -> AddResponse: key = digest([self.settings.memoria_url.rstrip("/"), owner, request.request_id]) fingerprint = digest(request.model_dump()) with self.journal.lock(key): - if not self.journal.begin(key, fingerprint): - for position, body in enumerate(self.write_bodies(request, key)): - self.write(owner, key, position, body) - self.journal.complete(key) + # One-item lookahead identifies the last chunk without buffering + # the entire request, so its stored marker and completion commit together. + bodies = chain(self.write_bodies(request, key), [None]) + for position, (body, following) in enumerate(pairwise(bodies)): + prepared = self.journal.prepare_write(key, fingerprint, position) + if prepared is None: + break + self.write(owner, key, position, body, prepared, last=following is None) return AddResponse( request_id=request.request_id, user_id=request.user_id, @@ -68,14 +73,22 @@ def write_bodies(self, request: AddRequest, key: str): body["observed_at"] = observed_at yield body - def write(self, owner: str, key: str, position: int, body: dict) -> None: - state, memory_id, protocol = self.journal.state(key, position) - if protocol not in {LEGACY_PROTOCOL, IDEMPOTENT_PROTOCOL}: - raise AdapterError(503, "Unsupported journal write protocol; check adapter version") + def write( + self, + owner: str, + key: str, + position: int, + body: dict, + prepared: tuple[str, str | None, str], + *, + last: bool = False, + ) -> None: + state, memory_id, protocol = prepared if state == "stored": + # Older journals may have persisted the final chunk but not complete. + if last: + self.journal.complete(key) return - if state not in {"pending", "inflight"}: - raise AdapterError(503, "Unsupported journal write state; check adapter version") marker = body["extra_metadata"]["aml_write"] if state == "inflight" and memory_id: # Verification below handles temporary unreadability without a POST. @@ -87,13 +100,8 @@ def write(self, owner: str, key: str, position: int, body: dict) -> None: raise AdapterError(503, "Prior write outcome is unknown; retry the same Add later") memory_id = memory.memory_id else: - # Pending rows and atomic inflight rows without an ID can both replay - # the receipt key. Unknown legacy inflight rows are handled above. - # A pending legacy write is known not to have been sent/committed. - # Persist the new protocol before POST so even a pre-send crash can - # safely replay it. Unknown legacy inflight writes never enter here. - protocol = IDEMPOTENT_PROTOCOL - self.journal.record(key, position, "inflight", protocol=protocol) + # prepare_write committed the atomic intent before this call. Reusing + # an existing atomic inflight row needs no additional SQLite commit. try: memory = self.memoria.store_idempotent(owner, body, marker) except AdapterError as exc: @@ -104,7 +112,14 @@ def write(self, owner: str, key: str, position: int, body: dict) -> None: self.journal.record(key, position, "inflight", memory_id, protocol=protocol) persisted = self.memoria.wait_for_visibility(owner, memory_id, marker) - self.journal.record(key, position, "stored", persisted.memory_id, protocol=protocol) + self.journal.record( + key, + position, + "stored", + persisted.memory_id, + protocol=protocol, + complete_request=last, + ) def search(self, request: SearchRequest) -> SearchResponse: memories = self.memoria.search( diff --git a/agent_memory_leaderboard/tests/test_idempotent_writes.py b/agent_memory_leaderboard/tests/test_idempotent_writes.py index 2758bc9..ce62721 100644 --- a/agent_memory_leaderboard/tests/test_idempotent_writes.py +++ b/agent_memory_leaderboard/tests/test_idempotent_writes.py @@ -87,17 +87,17 @@ def _exit_after_inflight(directory): state_dir=directory, ) journal = Journal(directory) - record = journal.record + prepare = journal.prepare_write def interrupt(*args, **kwargs): - record(*args, **kwargs) + prepare(*args, **kwargs) os._exit(23) def unexpected_post(request): (directory / "unexpected-post").touch() raise AssertionError("Process should exit before HTTP") - journal.record = interrupt + journal.prepare_write = interrupt with httpx.Client(transport=httpx.MockTransport(unexpected_post)) as http: adapter = MemoryAdapter(settings, Memoria(settings, http), journal) adapter.add(AddRequest.model_validate(addition())) diff --git a/agent_memory_leaderboard/tests/test_journal_benchmark.py b/agent_memory_leaderboard/tests/test_journal_benchmark.py new file mode 100644 index 0000000..0e2da81 --- /dev/null +++ b/agent_memory_leaderboard/tests/test_journal_benchmark.py @@ -0,0 +1,54 @@ +"""The benchmark must fail closed when its commit instrumentation is bypassed.""" + +import json +import shutil +import subprocess +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] + + +@pytest.mark.parametrize("implicit_commit", [False, True]) +def test_benchmark_rejects_unmeasured_context_manager_commits(tmp_path, implicit_commit): + shutil.copytree(ROOT / "src", tmp_path / "src", ignore=shutil.ignore_patterns("__pycache__")) + if implicit_commit: + journal = tmp_path / "src/journal.py" + lines = journal.read_text().splitlines(keepends=True) + replaced = False + for i, line in enumerate(lines): + if 'timing.run("commit", db.commit)' in line: + indent = line[: len(line) - len(line.lstrip())] + lines[i] = indent + "with db:\n" + indent + " pass\n" + replaced = True + assert replaced + journal.write_text("".join(lines)) + result = subprocess.run( + [ + sys.executable, + str(ROOT / "scripts/benchmark_journal.py"), + "--source-root", + str(tmp_path), + "--requests", + "2", + "--concurrency", + "1", + "--commit-delay-ms", + "1", + ], + capture_output=True, + text=True, + timeout=20, + ) + if implicit_commit: + assert result.returncode != 0 + assert "implicit context-manager commits are not measured" in result.stderr + assert not result.stdout.strip() # No misleading result for the rejected case. + else: + assert result.returncode == 0, result.stderr + summary = json.loads(result.stdout) + assert summary["statuses"] == {"200": 2} + assert summary["commits"] == 6 + assert summary["unique_backend_records"] == 2 diff --git a/agent_memory_leaderboard/tests/test_journal_metrics.py b/agent_memory_leaderboard/tests/test_journal_metrics.py index 35bc2c2..81f589b 100644 --- a/agent_memory_leaderboard/tests/test_journal_metrics.py +++ b/agent_memory_leaderboard/tests/test_journal_metrics.py @@ -50,8 +50,8 @@ def test_default_add_summary_includes_idempotent_replay_without_payloads(tmp_pat ) summaries = events(caplog, "journal_add_summary") assert [e["http_status"] for e in summaries] == [200, 200, 400] - assert summaries[0]["operations"]["record"]["count"] == 3 - assert summaries[1]["operations"]["begin"]["count"] == 1 + assert summaries[0]["operations"]["record"]["count"] == 2 + assert summaries[1]["operations"]["prepare_write"]["count"] == 1 assert "record" not in summaries[1]["operations"] assert len({e["add_trace"] for e in summaries}) == 3 assert all(0 <= e["journal_fraction"] <= 1.001 for e in summaries) diff --git a/agent_memory_leaderboard/tests/test_journal_transactions.py b/agent_memory_leaderboard/tests/test_journal_transactions.py new file mode 100644 index 0000000..c3a6d6a --- /dev/null +++ b/agent_memory_leaderboard/tests/test_journal_transactions.py @@ -0,0 +1,157 @@ +"""Durable transaction boundaries exercised with synthetic payloads and failures.""" + +import sqlite3 + +import httpx +import pytest +from test_endpoints import FakeMemoria, addition + +from src.config import Settings +from src.errors import AdapterError +from src.journal import IDEMPOTENT_PROTOCOL, Journal +from src.memoria import Memoria +from src.models import AddRequest +from src.service import MemoryAdapter, digest + + +@pytest.fixture +def adapter(tmp_path): + settings = Settings( + memoria_url="http://memoria.test", memoria_master_key="upstream-secret", state_dir=tmp_path + ) + journal = Journal(tmp_path) + upstream = FakeMemoria() + with httpx.Client(transport=httpx.MockTransport(upstream)) as http: + yield MemoryAdapter(settings, Memoria(settings, http), journal), upstream + + +def snapshot(adapter): + with sqlite3.connect(adapter.journal.path) as db: + requests = db.execute("SELECT complete FROM requests").fetchall() + writes = db.execute( + "SELECT state, memory_id, protocol FROM writes ORDER BY position" + ).fetchall() + return requests, writes + + +@pytest.mark.parametrize("chunks", [1, 3]) +def test_three_durable_commits_per_chunk_and_read_only_replay(adapter, monkeypatch, chunks): + service, upstream = adapter + original = sqlite3.connect + commits = [] + + class Counted(sqlite3.Connection): + def commit(self): + commits.append(True) + return super().commit() + + monkeypatch.setattr(sqlite3, "connect", lambda *a, **kw: original(*a, factory=Counted, **kw)) + request = AddRequest.model_validate( + addition( + messages=[{"role": "user", "content": f"Synthetic fact {i}"} for i in range(chunks)] + ) + ) + service.add(request) + assert len(commits) == 3 * chunks + assert len(upstream.memories) == chunks + assert snapshot(service)[0] == [(1,)] + service.add(request) + assert len(commits) == 3 * chunks + assert len(upstream.writes) == chunks + + +@pytest.mark.parametrize("failed_commit", [1, 2, 3]) +def test_failed_commit_preserves_recoverable_state(adapter, monkeypatch, failed_commit): + service, upstream = adapter + original = sqlite3.connect + attempts = 0 + + class Failing(sqlite3.Connection): + def commit(self): + nonlocal attempts + attempts += 1 + if attempts == failed_commit: + raise sqlite3.OperationalError("synthetic commit failure") + return super().commit() + + monkeypatch.setattr(sqlite3, "connect", lambda *a, **kw: original(*a, factory=Failing, **kw)) + request = AddRequest.model_validate(addition()) + with pytest.raises(AdapterError) as error: + service.add(request) + assert error.value.status == 503 + requests, writes = snapshot(service) + if failed_commit == 1: + assert requests == writes == [] + assert not upstream.writes + else: + assert requests == [(0,)] + assert writes[0][0] == "inflight" and writes[0][2] == IDEMPOTENT_PROTOCOL + assert (writes[0][1] is not None) == (failed_commit == 3) + # Reopen the journal as on restart. A lost ACK can replay the same receipt; + # a durable known ID must recover without issuing another write. + service.journal = Journal(service.settings.state_dir) + service.add(request) + assert snapshot(service)[0] == [(1,)] + assert len(upstream.memories) == 1 + assert len(upstream.writes) == (2 if failed_commit == 2 else 1) + + +def test_completion_sql_failure_rolls_back_last_stored_marker(adapter): + service, upstream = adapter + with sqlite3.connect(service.journal.path) as db: + db.execute("""CREATE TRIGGER reject_complete BEFORE UPDATE OF complete ON requests + BEGIN SELECT RAISE(ABORT, 'synthetic completion failure'); END""") + request = AddRequest.model_validate(addition()) + with pytest.raises(AdapterError): + service.add(request) + requests, writes = snapshot(service) + assert requests == [(0,)] + assert writes[0][0] == "inflight" and writes[0][1] is not None + with sqlite3.connect(service.journal.path) as db: + db.execute("DROP TRIGGER reject_complete") + service.add(request) + assert snapshot(service)[0] == [(1,)] + assert len(upstream.writes) == len(upstream.memories) == 1 + + +def test_interruption_after_final_commit_replays_without_http(adapter, monkeypatch): + service, upstream = adapter + original = service.journal.record + + def interrupted(*args, **kwargs): + original(*args, **kwargs) + if kwargs.get("complete_request"): + raise KeyboardInterrupt("synthetic process interruption") + + monkeypatch.setattr(service.journal, "record", interrupted) + request = AddRequest.model_validate(addition()) + with pytest.raises(KeyboardInterrupt): + service.add(request) + assert snapshot(service)[0] == [(1,)] + service.journal = Journal(service.settings.state_dir) + call_count = len(upstream.calls) + service.add(request) + assert len(upstream.calls) == call_count + + +def test_old_final_stored_row_completes_without_http(adapter): + service, upstream = adapter + request = AddRequest.model_validate(addition()) + key = digest([service.settings.memoria_url, service.owner(request.user_id), request.request_id]) + service.journal.begin(key, digest(request.model_dump())) + service.journal.record(key, 0, "stored", "synthetic-id", protocol=IDEMPOTENT_PROTOCOL) + service.add(request) + assert snapshot(service)[0] == [(1,)] + assert not upstream.calls + + +def test_network_calls_do_not_hold_journal_gate(adapter): + service, upstream = adapter + other = Journal(service.settings.state_dir, queue_timeout_seconds=0.05) + + def independent_journal_operation(*_): + other.state("independent-synthetic-request", 0) + + upstream.hook = independent_journal_operation + service.add(AddRequest.model_validate(addition())) + assert snapshot(service)[0] == [(1,)] diff --git a/agent_memory_leaderboard/tests/test_write_visibility.py b/agent_memory_leaderboard/tests/test_write_visibility.py index b29f762..8a5bae9 100644 --- a/agent_memory_leaderboard/tests/test_write_visibility.py +++ b/agent_memory_leaderboard/tests/test_write_visibility.py @@ -104,7 +104,9 @@ def lag(request, body, owner): return httpx.Response(200, content="null") upstream.hook = lag - MemoryAdapter(boundary.settings, boundary, journal).write(owner, "synthetic", 0, body) + MemoryAdapter(boundary.settings, boundary, journal).write( + owner, "synthetic", 0, body, journal.state("synthetic", 0) + ) assert len(reads) == 2 and len(upstream.writes) == 1 assert journal.state("synthetic", 0) == ("stored", memory.memory_id, protocol)