Harden input validation and error handling across CLI, adapters, and runner

.github/workflows/ci.yml

@@ -12,15 +12,19 @@ jobs:
       - uses: actions/setup-python@v5
         with:
           python-version: "3.12"
-      - run: pip install "ruff>=0.8,<1"
+      - run: pip install -e ".[dev]"
       - run: ruff check .
 
   test:
     runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
         with:
-          python-version: "3.12"
-      - run: pip install -e ".[all]" pytest pytest-asyncio
+          python-version: ${{ matrix.python-version }}
+      - run: pip install -e ".[all,dev]"
       - run: pytest -q

README.md

@@ -28,7 +28,7 @@ That's 119 paraphrased queries, and the two rank-1 misses are the dataset doing
 
 **Adversarial.** Retain a small set of facts chosen to invite a bad merge, trigger the store's consolidation, and check whether recall now returns a statement no source made. Then feed corrections and check whether the fabrication goes away. A store that invents a fact and won't retract it is worse than one that misses. A check that errors is reported as an error, never as "clean".
 
-**Silent write loss.** After retain, memeval polls until the stored docs are actually recallable, instead of trusting the write acknowledgment. This caught a real case during this repo's own release verification — the store returned success and the fact never became findable:
+**Silent write loss.** After retain, memeval polls a sample of docs spread across the whole corpus until each is actually recallable, instead of trusting the write acknowledgment. This caught a real case during this repo's own release verification — the store returned success and the fact never became findable:
 
 ```
 WARNING: 1 doc(s) never became recallable (silent write loss?): nw-06
@@ -41,9 +41,9 @@ Rank scores can come out perfect while a store quietly drops writes. This check
     git clone https://github.com/nuk3s/memeval
     cd memeval
     pip install -e .            # core: stdlib only
-    pip install -e .[tui]       # + replay TUI (Textual)
-    pip install -e .[charts]    # + SVG/PNG charts (matplotlib)
-    pip install -e .[all]
+    pip install -e ".[tui]"     # + replay TUI (Textual)
+    pip install -e ".[charts]"  # + SVG/PNG charts (matplotlib)
+    pip install -e ".[all]"
 
 Python 3.10 or newer.
 
@@ -95,7 +95,9 @@ numbers from it, and send a fix if the docs lied.
 Writing an adapter is ~40 lines: subclass `MemoryAdapter` from
 `memeval/adapters/base.py`, implement `retain(items)` and `recall(query, k)`, optionally
 `consolidate()`, `supersede()`, `prepare()`, and `config()`, then register it in
-`memeval/adapters/__init__.py`. Copy `example_rest.py` and start from there.
+`memeval/adapters/__init__.py`. Copy `example_rest.py` and start from there; the bundled adapters share
+`memeval/adapters/_http.py` for JSON requests, which folds the server's error body into
+the exception message so a 4xx in the run log says why.
 `memeval doctor` is the conformance check: it writes three sentinel docs, polls the recall
 roundtrip, and probes the optional capabilities. Partial visibility is an exit-1 PARTIAL,
 not a pass.

docs/run-log.md

@@ -19,7 +19,8 @@ line, in the order things happened. Same-second collisions get a `-2`, `-3`… s
 Notes.
 
 The `ready` event is the readiness poll that replaced a fixed sleep: after retain, memeval
-polls recall on a sample of corpus docs (word-overlap match, ≥50% of the doc's content
+polls recall on up to 10 docs spread evenly across the corpus, first and last always
+included so a dropped tail batch is caught (word-overlap match, ≥50% of the doc's content
 words, because extracting stores rewrite text) until each appears or `--ready-timeout`
 runs out. A doc in `missing` was accepted by the store but never became recallable —
 silent write loss. The run continues; the event keeps the evidence.
@@ -28,5 +29,9 @@ Hit text is truncated in the log for size; fabrication verdicts are computed on
 recalled text before truncation, so a `verdict` can reference a match you can't fully see
 in the neighbouring `query` event.
 
+`--retries N` re-issues a recall only on transport errors and HTTP 5xx/408/425/429. Any
+other 4xx (bad token, unknown route) fails identically every time, so it becomes an error row
+on the first attempt.
+
 Latency percentiles count only error-free queries; `errors` counts the rest. Error
 messages are never empty strings — a bare exception records its class name.

memeval/__main__.py

@@ -0,0 +1,5 @@
+"""`python -m memeval` entry point; the console script is memeval.cli:main."""
+from .cli import main
+
+if __name__ == "__main__":
+    main()

memeval/adapters/_http.py

@@ -0,0 +1,40 @@
+"""Shared urllib plumbing for the bundled REST adapters. Stdlib only.
+
+Every adapter used to carry its own copy of this, and only some of them folded the
+server's error body into the exception. Keeping it here means a 4xx in the run log
+says *why* for all of them.
+"""
+import json
+import urllib.error
+import urllib.request
+
+ERROR_BODY_LIMIT = 200
+
+
+def request_json(url, body=None, method=None, headers=None, timeout=120, context=None):
+    """Send `body` as JSON (POST by default when a body is given, GET otherwise) and
+    decode the JSON reply; an empty reply decodes to {}.
+
+    An HTTPError is re-raised with the first ERROR_BODY_LIMIT chars of the response
+    body in its message, so the error row reads "400 Bad Request — {detail...}"
+    instead of just the status line.
+    """
+    data = json.dumps(body).encode() if body is not None else None
+    if method is None:
+        method = "POST" if data is not None else "GET"
+    req = urllib.request.Request(
+        url, data=data, method=method,
+        headers={"Content-Type": "application/json", **(headers or {})})
+    try:
+        with urllib.request.urlopen(req, timeout=timeout, context=context) as r:
+            raw = r.read().decode()
+            return json.loads(raw) if raw else {}
+    except urllib.error.HTTPError as e:
+        detail = ""
+        try:
+            detail = e.read().decode()[:ERROR_BODY_LIMIT]
+        except Exception:
+            pass
+        raise urllib.error.HTTPError(
+            e.url, e.code, f"{e.reason} — {detail}" if detail else e.reason,
+            e.headers, None) from None

memeval/adapters/example_rest.py

@@ -1,8 +1,7 @@
-import json
 import os
-import urllib.request
 from typing import Dict, List
 
+from ._http import request_json
 from .base import MemoryAdapter
 
 
@@ -27,12 +26,8 @@ def __init__(self, url=None, token=None, user=None):
         self.user = user or os.environ.get("MEMEVAL_REST_USER", "memeval")
 
     def _post(self, path: str, body: Dict, timeout: int = 120):
-        req = urllib.request.Request(
-            self.url + path, data=json.dumps(body).encode(), method="POST",
-            headers={"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"})
-        with urllib.request.urlopen(req, timeout=timeout) as r:
-            raw = r.read().decode()
-            return json.loads(raw) if raw else {}
+        return request_json(self.url + path, body=body, method="POST", timeout=timeout,
+                            headers={"Authorization": f"Bearer {self.token}"})
 
     def retain(self, items: List[Dict]) -> None:
         for i in items:

memeval/adapters/hindsight.py

@@ -1,9 +1,8 @@
-import json
 import os
 import ssl
-import urllib.request
 from typing import Dict, List, Optional
 
+from ._http import request_json
 from .base import MemoryAdapter
 
 
@@ -15,7 +14,7 @@ class HindsightAdapter(MemoryAdapter):
       HINDSIGHT_TOKEN    bearer token       (default empty)
       HINDSIGHT_BANK     bank name          (default memeval-scratch)
       HINDSIGHT_INSECURE any non-empty value skips TLS verification (for internal CAs)
-      HINDSIGHT_BATCH    items per retain POST (default 20)
+      HINDSIGHT_BATCH    items per retain POST (default 20, minimum 1)
       HINDSIGHT_TIMEOUT  seconds per retain POST (default 600)
 
     Point HINDSIGHT_BANK at a throwaway bank. The eval writes into it.
@@ -31,21 +30,18 @@ def __init__(self, url: Optional[str] = None, token: Optional[str] = None,
         self.bank = bank or os.environ.get("HINDSIGHT_BANK", "memeval-scratch")
         if insecure is None:
             insecure = bool(os.environ.get("HINDSIGHT_INSECURE", ""))
-        self.batch = int(os.environ.get("HINDSIGHT_BATCH", "20"))
+        # a batch of 0 would make range() raise on the first retain
+        self.batch = max(1, int(os.environ.get("HINDSIGHT_BATCH", "20")))
         self.timeout = int(os.environ.get("HINDSIGHT_TIMEOUT", "600"))
         self.ctx = ssl.create_default_context()
         if insecure:
             self.ctx.check_hostname = False
             self.ctx.verify_mode = ssl.CERT_NONE
 
     def _call(self, method: str, path: str, body=None, timeout: int = 300):
-        data = json.dumps(body).encode() if body is not None else None
-        req = urllib.request.Request(
-            self.url + path, data=data, method=method,
-            headers={"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"})
-        with urllib.request.urlopen(req, timeout=timeout, context=self.ctx) as r:
-            raw = r.read().decode()
-            return json.loads(raw) if raw else {}
+        return request_json(self.url + path, body=body, method=method, timeout=timeout,
+                            headers={"Authorization": f"Bearer {self.token}"},
+                            context=self.ctx)
 
     def _bank(self, suffix: str) -> str:
         return f"/v1/default/banks/{self.bank}{suffix}"
@@ -82,10 +78,10 @@ def recall(self, query: str, k: int = 10) -> List[Dict]:
         return out
 
     def consolidate(self) -> None:
-        try:
-            self._call("POST", self._bank("/consolidate"), {})
-        except Exception:
-            pass  # bank may have observations off; nothing to consolidate
+        # Deliberately unguarded. Runner.consolidate records a failure in the run log
+        # and continues, and doctor reports it; swallowing it here would log ok=True
+        # for a consolidation that never ran (observations off, server down, 4xx).
+        self._call("POST", self._bank("/consolidate"), {})
 
     def supersede(self, doc_id: str, content: str, tags: Optional[list] = None) -> None:
         self.retain([{"content": content, "id": doc_id, "tags": tags or []}])

memeval/adapters/letta.py

@@ -1,9 +1,7 @@
-import json
 import os
-import urllib.error
 import urllib.parse
-import urllib.request
 
+from ._http import request_json
 from .base import MemoryAdapter
 
 
@@ -36,26 +34,8 @@ def _agent_path(self, suffix=""):
         return f"{self.url}/v1/agents/{agent}/archival-memory{suffix}"
 
     def _call(self, url, body=None, timeout=120):
-        headers = {"Content-Type": "application/json"}
-        if self.token:
-            headers["Authorization"] = f"Bearer {self.token}"
-        data = json.dumps(body).encode() if body is not None else None
-        req = urllib.request.Request(url, data=data,
-                                     method="POST" if body is not None else "GET",
-                                     headers=headers)
-        try:
-            with urllib.request.urlopen(req, timeout=timeout) as r:
-                raw = r.read().decode()
-                return json.loads(raw) if raw else {}
-        except urllib.error.HTTPError as e:
-            detail = ""
-            try:
-                detail = e.read().decode()[:200]
-            except Exception:
-                pass
-            raise urllib.error.HTTPError(
-                e.url, e.code, f"{e.reason} — {detail}" if detail else e.reason,
-                e.headers, None) from None
+        headers = {"Authorization": f"Bearer {self.token}"} if self.token else {}
+        return request_json(url, body=body, headers=headers, timeout=timeout)
 
     def retain(self, items):
         for i in items:

memeval/adapters/mem0.py

@@ -1,8 +1,6 @@
-import json
 import os
-import urllib.error
-import urllib.request
 
+from ._http import request_json
 from .base import MemoryAdapter
 
 
@@ -30,24 +28,9 @@ def __init__(self):
         self.infer = os.environ.get("MEM0_INFER", "1") not in ("0", "false", "no")
 
     def _call(self, path, body, timeout=120):
-        headers = {"Content-Type": "application/json"}
-        if self.api_key:
-            headers["X-API-Key"] = self.api_key
-        req = urllib.request.Request(self.url + path, data=json.dumps(body).encode(),
-                                     method="POST", headers=headers)
-        try:
-            with urllib.request.urlopen(req, timeout=timeout) as r:
-                raw = r.read().decode()
-                return json.loads(raw) if raw else {}
-        except urllib.error.HTTPError as e:
-            detail = ""
-            try:
-                detail = e.read().decode()[:200]
-            except Exception:
-                pass
-            raise urllib.error.HTTPError(
-                e.url, e.code, f"{e.reason} — {detail}" if detail else e.reason,
-                e.headers, None) from None
+        headers = {"X-API-Key": self.api_key} if self.api_key else {}
+        return request_json(self.url + path, body=body, method="POST",
+                            headers=headers, timeout=timeout)
 
     def retain(self, items):
         for i in items:

memeval/adapters/zep.py

@@ -1,9 +1,8 @@
-import json
 import os
 import urllib.error
 import urllib.parse
-import urllib.request
 
+from ._http import request_json
 from .base import MemoryAdapter
 
 
@@ -34,24 +33,9 @@ def __init__(self):
         self.session = os.environ.get("ZEP_SESSION", "memeval-scratch")
 
     def _call(self, path, body, timeout=120):
-        headers = {"Content-Type": "application/json"}
-        if self.api_key:
-            headers["Authorization"] = f"Api-Key {self.api_key}"
-        req = urllib.request.Request(self.url + path, data=json.dumps(body).encode(),
-                                     method="POST", headers=headers)
-        try:
-            with urllib.request.urlopen(req, timeout=timeout) as r:
-                raw = r.read().decode()
-                return json.loads(raw) if raw else {}
-        except urllib.error.HTTPError as e:
-            detail = ""
-            try:
-                detail = e.read().decode()[:200]
-            except Exception:
-                pass
-            raise urllib.error.HTTPError(
-                e.url, e.code, f"{e.reason} — {detail}" if detail else e.reason,
-                e.headers, None) from None
+        headers = {"Authorization": f"Api-Key {self.api_key}"} if self.api_key else {}
+        return request_json(self.url + path, body=body, method="POST",
+                            headers=headers, timeout=timeout)
 
     def prepare(self):
         for path, body in (("/api/v2/users", {"user_id": self.user}),

memeval/adversarial.py

@@ -14,7 +14,46 @@
 
 def load_probe(path):
     with open(path, encoding="utf-8") as f:
-        return json.load(f)
+        return validate_probe(json.load(f))
+
+
+def _nonempty_str(v):
+    return isinstance(v, str) and bool(v.strip())
+
+
+def validate_probe(probe):
+    """Shape-check the probe at load time, so a typo fails before retain writes into
+    the store rather than as a KeyError or re.error halfway through the run."""
+    if not isinstance(probe, dict):
+        raise ValueError("probe must be a JSON object")
+    corpus = probe.get("corpus")
+    if not isinstance(corpus, list) or not corpus:
+        raise ValueError("probe 'corpus' must be a non-empty list")
+    for n, d in enumerate(corpus, 1):
+        if not isinstance(d, dict) or not _nonempty_str(d.get("content")):
+            raise ValueError(f"probe corpus item {n}: missing or empty 'content'")
+    checks = probe.get("checks")
+    if not isinstance(checks, list) or not checks:
+        raise ValueError("probe 'checks' must be a non-empty list")
+    for n, c in enumerate(checks, 1):
+        if not isinstance(c, dict) or not _nonempty_str(c.get("query")):
+            raise ValueError(f"probe check {n}: missing or empty 'query'")
+        for key in ("forbidden", "expected"):
+            pats = c.get(key, [])
+            if not isinstance(pats, list) or not all(isinstance(p, str) for p in pats):
+                raise ValueError(f"probe check {n}: {key!r} must be a list of regex strings")
+            for p in pats:
+                try:
+                    re.compile(p)
+                except re.error as e:
+                    raise ValueError(f"probe check {n}: bad {key} regex {p!r}: {e}") from None
+    corrections = probe.get("corrections", [])
+    if not isinstance(corrections, list) or not all(isinstance(x, str) for x in corrections):
+        raise ValueError("probe 'corrections' must be a list of strings")
+    wait = probe.get("consolidate_wait", 2)
+    if isinstance(wait, bool) or not isinstance(wait, (int, float)) or wait < 0:
+        raise ValueError("probe 'consolidate_wait' must be a non-negative number of seconds")
+    return probe
 
 
 def _check(runner, checks, phase):