perf(adapter): reduce durable SQLite commits per Add

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

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()

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:

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",
+    }
 )