fix(plugins): land a slice despite a dropped oversize message, normalise keys, skip landed adds on retry

Three findings from the second review pass. A slice holding one message over
the 250 KB ceiling answered False after the other messages had landed, and
both retrying callers (StorePipeline, the importer) re-sent the whole slice on
every attempt: the drop is deterministic, so the rest was stored again and an
import source never completed. The slice now reports landed; the warning that
names the session and the message index stays. resolve_api_key strips
whitespace and refuses a key with a control character, because the HTTP layer
quotes the whole Authorization value in its error and that text reached the
health hint, the settings card and the gateway log. A retry of the same slice
skips the adds the service already accepted (their digests are kept until the
slice lands), so a transient failure mid-slice stores nothing twice; the digest
ignores the fill-in timestamp. A lone surrogate in a message is replaced so the
size check and the send never raise. A parity test holds convert_messages to
the local plugin's.

Co-authored-by: Claude (claude-fable-5-1) <noreply@anthropic.com>

Author: gloryfromca

plugins-dist/everos-cloud-memory/raven_everos_cloud/backend.py

@@ -28,6 +28,7 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import json
 import logging
 import math
@@ -96,11 +97,14 @@ def resolve_api_key(config: Mapping[str, Any]) -> tuple[str, str | None]:
     one the settings page edits; a key that came from the environment is never
     written back, so the file records only what a person typed.
     """
-    key = str(config.get("api_key") or "")
-    if key:
-        return key, "file"
-    key = os.environ.get(ENV_KEY, "")
-    return (key, "env") if key else ("", None)
+    for raw, source in ((config.get("api_key"), "file"), (os.environ.get(ENV_KEY), "env")):
+        key = str(raw or "").strip()
+        if key:
+            # A control character cannot travel in a header: the HTTP layer refuses
+            # it by quoting the whole Authorization value in its error, and every
+            # hint and log line that prints that error would show the key.
+            return (key, source) if key.isprintable() else ("", None)
+    return "", None
 
 
 # ---------------------------------------------------------------------------
@@ -161,6 +165,9 @@ def convert_messages(
             content = ""  # a tool-call-only turn; "None" would be read by the extraction model as a reply
         elif not isinstance(content, str):
             content = str(content)
+        # A lone surrogate (a Claude Code transcript can carry one) cannot be encoded;
+        # it becomes U+FFFD here so the slice is sendable and ``store`` never raises.
+        content = content.encode("utf-8", errors="replace").decode("utf-8")
         tool_calls = m.get("tool_calls") if role == "assistant" else None
         if not content and not tool_calls:
             continue
@@ -243,6 +250,21 @@ def _score(row: dict[str, Any], default: float = 0.0) -> float:
         return default
 
 
+def _digest(body: dict[str, Any]) -> str:
+    """Identity of one add for the retry memo: the content, not the clock.
+
+    A row without a timestamp is stamped with "now" on every conversion, so a
+    retry of the same slice would never match if the stamp took part.
+    """
+    stable = {
+        "session_id": body.get("session_id"),
+        "messages": [{k: v for k, v in m.items() if k != "timestamp"} for m in body.get("messages", [])],
+    }
+    return hashlib.sha256(
+        json.dumps(stable, ensure_ascii=False, sort_keys=True).encode("utf-8", errors="replace")
+    ).hexdigest()
+
+
 def _joined(*parts: Any) -> str:
     return " - ".join(str(part).strip() for part in parts if part)
 
@@ -370,6 +392,7 @@ def __init__(self, ctx: PluginContext, *, client: httpx.AsyncClient | None = Non
         self._client: httpx.AsyncClient | None = client
         self._owns_client = client is None
         self._unflushed: set[str] = set()
+        self._landed: dict[str, set[str]] = {}
         self._stopping = False
         self._no_key_logged = False
 
@@ -527,32 +550,44 @@ async def store(
             return True
         if not self._has_key("store"):
             return False
-        batches, dropped = self._batches(session_id, payload)
+        batches, _dropped = self._batches(session_id, payload)
         explicit_end = bool(metadata and (metadata.get("flush") or metadata.get("is_final")))
         self._unflushed.add(session_id)
         for batch in batches:
             body = {"session_id": session_id, "mode": "agent", "messages": batch}
+            # Both callers re-send a slice that answered False (contracts/memory.py);
+            # a batch the service accepted on the failed attempt is not sent twice.
+            # The memo lives only until the slice lands, so a later store of the
+            # same content is a new write, not a retry.
+            digest = _digest(body)
+            if digest in self._landed.get(session_id, ()):
+                continue
             if not await self._send("add", "/api/v2/memory/add", body, ADD_TIMEOUT_S):
                 return False
+            self._landed.setdefault(session_id, set()).add(digest)
         if explicit_end:
             if not await self._send("flush", "/api/v2/memory/flush", {"session_id": session_id}, flush_timeout_s()):
                 return False
             self._unflushed.discard(session_id)
-        return not dropped
+        self._landed.pop(session_id, None)
+        # A dropped oversize message is logged by _batches and is not a reason to
+        # answer False: no retry can make it fit, and a False here would only make
+        # both callers store the rest of the slice again on every attempt.
+        return True
 
     def _batches(self, session_id: str, payload: list[dict[str, Any]]) -> tuple[list[list[dict[str, Any]]], bool]:
         """Split a slice at the service's limits; a single oversize message is dropped and named.
 
         A message over the byte ceiling cannot be made to fit without storing
         words the user never said, so it is left out and the rest goes on; the
-        caller hears ``False`` for the slice.
+        slice still reports landed (see ``store``).
         """
         batches: list[list[dict[str, Any]]] = []
         current: list[dict[str, Any]] = []
         size = 0
         dropped = False
         for index, message in enumerate(payload):
-            item_size = len(json.dumps(message, ensure_ascii=False).encode("utf-8"))
+            item_size = len(json.dumps(message, ensure_ascii=False).encode("utf-8", errors="replace"))
             if item_size > ADD_MAX_BYTES:
                 # ponytail: a single message over the ceiling is dropped whole; chunk one
                 # message across adds if real turns ever hit this.

tests/test_everos_cloud_backend.py

@@ -13,7 +13,7 @@
 from raven.contracts.memory import BackendHealth, Memory
 from raven.plugins import PluginContext, ServiceLocator
 from raven_everos_cloud import backend as mod
-from raven_everos_cloud.backend import EverosCloudBackend, resolve_api_key
+from raven_everos_cloud.backend import EverosCloudBackend, convert_messages, resolve_api_key
 from tests._everos_cloud_fake import FakeCloud
 
 pytestmark = pytest.mark.everos_cloud
@@ -40,6 +40,14 @@ def _requests(fake: FakeCloud, path_end: str) -> list[dict]:
 # ── key resolution (C7) ──────────────────────────────────────────────
 
 
+def test_key_resolution_strips_whitespace_and_refuses_a_control_character(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A trailing newline would make the HTTP layer quote the whole Authorization value in its error."""
+    monkeypatch.setenv("EVEROS_CLOUD_API_KEY", "from-env\n")
+    assert resolve_api_key({"api_key": " from-file \n"}) == ("from-file", "file")
+    assert resolve_api_key({}) == ("from-env", "env")
+    assert resolve_api_key({"api_key": "from\x00file"}) == ("", None)
+
+
 def test_key_resolution_prefers_the_file(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("EVEROS_CLOUD_API_KEY", "from-env")
     assert resolve_api_key({"api_key": "from-file"}) == ("from-file", "file")
@@ -193,7 +201,9 @@ async def test_oversize_message_fails_its_own_add_and_logs_index(tmp_path: Path,
         + [{"role": "user", "content": "small"}] * 2
     )
     with caplog.at_level(logging.WARNING, logger="t"):
-        assert await b.store("over", msgs) is False
+        # The slice lands: no retry can make the dropped message fit, and a False
+        # would make both callers store the other four again on every attempt.
+        assert await b.store("over", msgs) is True
     adds = _requests(fake, "/add")
     assert len(adds) == 1 and len(adds[0]["body"]["messages"]) == 4
     assert "message 2 of session over" in caplog.text
@@ -216,7 +226,7 @@ async def test_batches_are_sized_in_utf8_bytes_not_code_points(tmp_path: Path, c
     b2 = _backend(tmp_path, fake2)
     big = [{"role": "user", "content": "\u00e9" * 150_000}, {"role": "user", "content": "small"}]
     with caplog.at_level(logging.WARNING, logger="t"):
-        assert await b2.store("over-utf8", big) is False
+        assert await b2.store("over-utf8", big) is True
     adds2 = _requests(fake2, "/add")
     assert len(adds2) == 1 and len(adds2[0]["body"]["messages"]) == 1
     assert "message 0 of session over-utf8 is 300" in caplog.text
@@ -253,6 +263,60 @@ async def test_a_tool_call_only_turn_is_stored_with_empty_content(tmp_path: Path
     assert sent[1]["content"] == "" and sent[1]["role"] == "assistant"
 
 
+async def test_a_lone_surrogate_is_replaced_and_the_slice_still_lands(tmp_path: Path) -> None:
+    fake = FakeCloud()
+    b = _backend(tmp_path, fake)
+    assert await b.store("surrogate", [{"role": "user", "content": "half \ud83d emoji"}]) is True
+    sent = _requests(fake, "/add")[0]["body"]["messages"][0]["content"]
+    assert "\ud83d" not in sent and "half" in sent and "emoji" in sent
+
+
+async def test_a_retry_after_a_transient_failure_skips_the_batches_that_landed(tmp_path: Path) -> None:
+    """StorePipeline re-sends a slice that answered False; the batches the service
+    already accepted must not be stored twice for it."""
+    fake = FakeCloud()
+    inner = fake.transport()
+    adds = {"n": 0}
+
+    async def flaky(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("/add"):
+            adds["n"] += 1
+            if adds["n"] == 2:
+                return httpx.Response(500, json={"error": {"message": "ECM-TRANSIENT"}})
+        return await inner.handle_async_request(request)
+
+    b = EverosCloudBackend(_ctx(tmp_path), client=httpx.AsyncClient(transport=httpx.MockTransport(flaky)))
+    msgs = [{"role": "user", "content": f"m{i}"} for i in range(600)]
+    assert await b.store("retry", msgs) is False
+    assert await b.store("retry", msgs) is True
+    sent = [len(a["body"]["messages"]) for a in _requests(fake, "/add")]
+    assert sent == [500, 100]  # the 500-message batch once, the 100-message batch once (its first try never landed)
+
+
+async def test_message_translation_matches_the_local_plugin() -> None:
+    """The cloud copy of convert_messages must not drift from raven_everos (the boundary forbids importing it)."""
+    from raven_everos.backend import convert_messages as local_convert
+
+    rows = [
+        {"role": "user", "content": "hello", "timestamp": 1_760_000_000_000},
+        {
+            "role": "assistant",
+            "content": None,
+            "tool_calls": [{"id": "c1", "type": "function", "function": {"name": "x", "arguments": "{}"}}],
+            "timestamp": 1_760_000_000_001,
+        },
+        {
+            "role": "tool",
+            "content": [{"type": "text", "text": "ok"}, {"type": "image", "url": "x"}],
+            "tool_call_id": "c1",
+            "timestamp": 1_760_000_000_002,
+        },
+        {"role": "system", "content": "dropped", "timestamp": 1_760_000_000_003},
+        {"role": "assistant", "content": 42, "timestamp": "2026-10-09T00:00:00Z"},
+    ]
+    assert convert_messages(rows, agent_id="a", user_id="u") == local_convert(rows, agent_id="a", user_id="u")
+
+
 # ── recall_session (A15) ────────────────────────────────────────────