From d793f7b9fb09bc4d89508bf7cdd1dcd0bc3b49ec Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 15 Sep 2026 11:48:19 +0530 Subject: [PATCH] fix(supermemory): explicit failure sentinel and bounded pending-turn buffer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings on the salvage stack: - _write_turns' _quietly consolidation keyed failure on a None result, implicitly assuming add_memory never legitimately returns None. A None- returning stub (the most common mock idiom) would mark every successful write failed and re-append the batch forever. Module-level _FAILED sentinel: only a raised exception re-queues. - _pending_turns had no bound: a persistently failing service accumulated one entry per turn for the process lifetime (gateway runs never re-initialize), and every retry re-sent the whole accumulated payload — O(n^2) upload bytes, and a size-rejected batch could never shrink. Cap the buffer (50 turns / 256 KiB, drop oldest, warn once per trim). --- plugins/memory/supermemory/__init__.py | 31 ++++++++++++++++++++++++-- 1 file changed, 29 insertions(+), 2 deletions(-) diff --git a/plugins/memory/supermemory/__init__.py b/plugins/memory/supermemory/__init__.py index c7e77b3e3e..1a3fa222c4 100644 --- a/plugins/memory/supermemory/__init__.py +++ b/plugins/memory/supermemory/__init__.py @@ -26,6 +26,9 @@ _API_KEY_URL = "http://app.supermemory.ai/integrations?connect=hermes" _INJECTED_BLOCK_RE = re.compile(r"[\s\S]*?\s*", re.DOTALL) _DATA_URI_RE = re.compile(r"data:[^;,\s]+;base64,[A-Za-z0-9+/=]+") # pasted inline images are useless as memory text _CAPTURE_BUCKET_HOURS = 4 # one capture document per session per 4h window (matches the other Supermemory agent integrations) +_FAILED = object() # _quietly default for capture writes: an explicit failure marker (a client returning None still counts as success) +_MAX_PENDING_TURNS = 50 # a down service must not accumulate an unbounded retry buffer +_MAX_PENDING_BYTES = 256 * 1024 _DEFAULT_ENTITY_CONTEXT = ( "User-assistant conversation. Format: [role: user]...[user:end] and [role: assistant]...[assistant:end].\n\n" "Only extract things useful in future conversations. Most messages are not worth remembering.\n\n" @@ -430,8 +433,8 @@ class SupermemoryMemoryProvider(MemoryProvider): result = _quietly(lambda: self._client.add_memory(content, metadata=metadata, entity_context=self._entity_context, custom_id=_capture_custom_id(sid, now)), "Supermemory capture failed (%s, session=%s, %d turns pending)", mode, sid, len(batch), - level=logging.WARNING if mode != "turn" else logging.DEBUG) - if result is None: # add_memory always returns a dict on success; None = it raised + level=logging.WARNING if mode != "turn" else logging.DEBUG, default=_FAILED) + if result is _FAILED: # only a raised exception re-queues the batch failed += batch self._pending_turns = failed @@ -443,10 +446,34 @@ class SupermemoryMemoryProvider(MemoryProvider): "session_id": session_id or self._session_id} if turn["user"] or turn["assistant"]: self._write_turns("turn", turn) + self._bound_pending_turns() def _flush_pending(self, mode: str) -> None: if self._can_write(): self._write_turns(mode) + self._bound_pending_turns() + + def _bound_pending_turns(self) -> None: + """Keep the retry buffer bounded: drop OLDEST entries past the turn/byte caps. + + Without this, a persistently failing service accumulates one entry per turn for the + process lifetime (gateway runs never re-initialize) and every retry re-sends the + whole accumulated payload.""" + with self._capture_lock: + if len(self._pending_turns) <= _MAX_PENDING_TURNS and \ + sum(len(t["user"]) + len(t["assistant"]) for t in self._pending_turns) <= _MAX_PENDING_BYTES: + return + kept: List[Dict[str, str]] = list(self._pending_turns) + total = sum(len(t["user"]) + len(t["assistant"]) for t in kept) + while len(kept) > _MAX_PENDING_TURNS or total > _MAX_PENDING_BYTES: + if not kept: + break + dropped = kept.pop(0) + total -= len(dropped["user"]) + len(dropped["assistant"]) + if len(kept) != len(self._pending_turns): + logger.warning("Supermemory: dropped %d oldest pending turn(s) to keep the retry buffer bounded", + len(self._pending_turns) - len(kept)) + self._pending_turns = kept def on_session_end(self, messages: List[Dict[str, Any]]) -> None: # Turns were already written as they completed; only retry what failed.