fix(memory): avoid replaying Honcho checkpoint history (#7687)

* fix(memory): avoid replaying Honcho checkpoint history

* fix(memory): harden Honcho capture lookups

---------

Co-authored-by: haotianqi666 <250384991+haotianqi666@users.noreply.github.com>

backend/packages/harness/deerflow/agents/memory/AGENTS.md

@@ -383,6 +383,17 @@ Do not add embedded OpenViking imports, root-key access, or trusted identity hea
 Honcho is a remote HTTP adapter for user-model memory.
 It creates one workspace per resolved `user_id`.
 A missing user fails closed to no memory.
+Capture stable-ID messages incrementally using remote `deerflow_source_id`
+metadata, scoped by session and including owner/role/text in the opaque identity.
+Never deduplicate distinct source IDs by text. Consume every filtered lookup page
+before appending a batch; failed/malformed lookups must not become blind replays.
+Cap each lookup at 10 pages, aborting over-budget captures without partial results;
+duplicate stored identities may span pages. Distinguish lookup/write warnings.
+Bound batches to Honcho's 100-message limit. Keep lookup + append under the
+process-local session lock; do not claim cross-process exactly-once delivery.
+ID-less direct callers remain append-only; legacy records lack capture identities.
+Pin cumulative graph turns, restart/compaction, failures and pagination through
+the real HTTP client in `tests/test_honcho_incremental_capture.py`.
 Its async methods offload synchronous HTTP work with `asyncio.to_thread`.
 Honcho's `from_config` consumes the `should_keep_hidden_message` host hook.
 Its write path drops hidden human messages unless that hook accepts them;

backend/packages/harness/deerflow/agents/memory/backends/honcho/README.md

@@ -71,6 +71,39 @@ cannot see each other's memory by construction.
   workspace-scoped search, while the passive per-turn write middleware is
   retained because Honcho's deriver learns from `add()` writes.
 
+## Incremental conversation capture
+
+Checkpoint history is cumulative. For messages with stable source IDs, the
+adapter stores an opaque `deerflow_source_id` in Honcho message metadata and
+checks the session's metadata-filtered `messages/list` endpoint before appending
+new messages. This requires session-read access as well as message-write access.
+Lookups and writes use batches of at most 100 messages. Each lookup requests
+100 records per page and consumes at most 10 pages before publishing that batch.
+All advertised pages must be consumed; a larger page count aborts capture rather
+than using partial results. The budget allows multiple stored records per source
+identity, but exceptionally duplicate-heavy histories may exceed it.
+
+The identity includes the owner, source message ID, role, peer and normalized
+text. Distinct messages with identical text are retained; edited source messages
+are captured as new versions, not replacements of old Honcho records. Stored
+identities survive manager/process restarts and overlapping compacted history.
+An unavailable, malformed or over-budget lookup aborts that capture with a
+`capture lookup failed` warning instead of assuming the history is new.
+This is distinct from `write failed` for setup or message-write errors; a later
+capture can retry normally once the lookup succeeds.
+
+Lookup and append are serialized per endpoint/workspace/session within one
+process, including across manager instances. This is **not exactly-once delivery**:
+independent processes must serialize access to a session upstream. A retry can
+recognize an earlier write whose response was lost once the committed records
+are visible to the lookup, but Honcho does not provide an atomic deduplication
+constraint here.
+
+Direct callers without stable message IDs retain append-on-every-call behavior.
+Records written before this metadata was introduced cannot be matched reliably:
+the first capture after upgrading may replay that older history once. This
+change does not remove historical duplicates or migrate existing records.
+
 ## Limitations
 
 - No DeerMem-style fact CRUD: `create_fact`/`delete_fact`/`update_fact`,
@@ -81,8 +114,10 @@ cannot see each other's memory by construction.
   unsupported-operation error; conversation writes still happen through the
   retained middleware. DeerMem remains the default backend.
 - No migration of existing DeerMem data.
-- Writes are fire-and-forget per call: a failed write is logged and dropped
-  (at-most-once); nothing is buffered locally, so `shutdown_flush` is a no-op.
+- Writes run synchronously within each call. Failed captures are logged; no
+  retry is scheduled or buffered locally, so `shutdown_flush` is a no-op.
+  A later call can retry messages still present in its history, while stored
+  source identities suppress already committed messages.
 - `agent_name` is not mapped — Honcho models the user, not per-agent facts.
 
 ## Async execution and failure behavior

backend/packages/harness/deerflow/agents/memory/backends/honcho/client.py

@@ -1,7 +1,7 @@
 """Minimal synchronous Honcho v3 HTTP client for the memory backend.
 
-Deliberately dependency-light: plain httpx against the v3 REST API (peers and
-sessions are get-or-create server-side, so every call here is idempotent).
+Deliberately dependency-light: plain httpx against the v3 REST API. Peers and
+sessions are get-or-create server-side; message creation is append-only.
 Moving to the official ``honcho-ai`` SDK is a possible follow-up, mirroring the
 OpenViking custom-HTTP -> official-adapter arc.
 """
@@ -14,6 +14,10 @@ import httpx
 
 from .config import HonchoConfig
 
+SOURCE_ID_KEY = "deerflow_source_id"
+MESSAGE_BATCH_SIZE = 100
+MAX_LOOKUP_PAGES = 10
+
 
 class HonchoRequestError(RuntimeError):
     """A Honcho API call failed (transport error or non-2xx response)."""
@@ -57,9 +61,40 @@ class HonchoClient:
     def set_session_peers(self, workspace: str, session_id: str, peer_ids: list[str]) -> None:
         self._post(f"/v3/workspaces/{workspace}/sessions/{session_id}/peers", {peer_id: {} for peer_id in peer_ids})
 
-    def add_messages(self, workspace: str, session_id: str, messages: list[dict[str, str]]) -> None:
+    def add_messages(self, workspace: str, session_id: str, messages: list[dict[str, Any]]) -> None:
         self._post(f"/v3/workspaces/{workspace}/sessions/{session_id}/messages", {"messages": messages})
 
+    def existing_source_ids(self, workspace: str, session_id: str, source_ids: list[str]) -> set[str]:
+        """Consume all lookup pages within a bounded budget, or fail the batch."""
+        if not source_ids:
+            return set()
+        requested = set(source_ids)
+        payload = {"filters": {"OR": [{"metadata": {SOURCE_ID_KEY: key}} for key in source_ids]}}
+        found: set[str] = set()
+        page = 1
+        while True:
+            data = self._post(f"/v3/workspaces/{workspace}/sessions/{session_id}/messages/list?page={page}&size=100", payload)
+            if not isinstance(data, dict) or not isinstance(data.get("items"), list):
+                raise HonchoRequestError("Honcho message lookup returned an invalid page")
+            pages = data.get("pages")
+            if type(pages) is not int or pages < 0 or type(data.get("page")) is not int or data["page"] != page:
+                raise HonchoRequestError("Honcho message lookup returned invalid pagination")
+            if pages < page and not (page == 1 and pages == 0 and not data["items"]):
+                raise HonchoRequestError("Honcho message lookup returned inconsistent pagination")
+            # Duplicate stored identities can span multiple pages even for one
+            # requested key. Bound requests, not the number of matching rows.
+            if pages > MAX_LOOKUP_PAGES:
+                raise HonchoRequestError(f"Honcho message lookup exceeded the {MAX_LOOKUP_PAGES}-page limit")
+            for item in data["items"]:
+                metadata = item.get("metadata") if isinstance(item, dict) else None
+                key = metadata.get(SOURCE_ID_KEY) if isinstance(metadata, dict) else None
+                if not isinstance(key, str) or key not in requested:
+                    raise HonchoRequestError("Honcho message lookup returned an invalid source identity")
+                found.add(key)
+            if page >= pages:
+                return found
+            page += 1
+
     def working_representation(self, workspace: str, peer_id: str, *, max_conclusions: int = 25) -> str:
         data = self._post(f"/v3/workspaces/{workspace}/peers/{peer_id}/representation", {"max_conclusions": max_conclusions})
         if isinstance(data, dict):

backend/packages/harness/deerflow/agents/memory/backends/honcho/honcho_manager.py

@@ -35,7 +35,10 @@ from __future__ import annotations
 
 import asyncio
 import hashlib
+import json
 import logging
+import threading
+import weakref
 from typing import Any, ClassVar, Literal
 
 from pydantic import PrivateAttr
@@ -43,12 +46,27 @@ from pydantic import PrivateAttr
 # ABC contract -- the ONE allowed `from deerflow` import in this backend folder.
 from deerflow.agents.memory.manager import MemoryManager, MemoryManagerError, MemoryReadError
 
-from .client import HonchoClient
+from .client import MESSAGE_BATCH_SIZE, SOURCE_ID_KEY, HonchoClient
 from .config import HonchoConfig, sanitize_id
 
 logger = logging.getLogger(__name__)
 
 _UTC_NOW_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
+_CaptureKey = tuple[str, str, str]
+_capture_locks: weakref.WeakValueDictionary[_CaptureKey, threading.RLock] = weakref.WeakValueDictionary()
+_capture_locks_guard = threading.Lock()
+
+
+def _capture_lock(base_url: str, workspace: str, session_id: str) -> threading.RLock:
+    # Serialize lookup + append across managers in this process without retaining
+    # inactive sessions. Independent processes still need upstream admission.
+    key = (base_url.rstrip("/"), workspace, session_id)
+    with _capture_locks_guard:
+        lock = _capture_locks.get(key)
+        if lock is None:
+            lock = threading.RLock()
+            _capture_locks[key] = lock
+        return lock
 
 
 def _now_iso() -> str:
@@ -92,10 +110,10 @@ class HonchoMemoryManager(MemoryManager):
     """MemoryManager backed by a Honcho v3 instance (self-hosted or hosted).
 
     Known limitation: ``search`` scopes ``category`` server-side to messages
-    stored with ``metadata.category``, but ``add()`` writes plain metadata-less
-    messages and fact CRUD is unsupported, so nothing this backend stores ever
+    stored with ``metadata.category``, but ``add()`` sets only source-identity
+    metadata and fact CRUD is unsupported, so nothing this backend stores ever
     carries a category — category queries return empty results here until the
-    write path gains metadata support (see ``search``).
+    write path gains category support (see ``search``).
     """
 
     _config: HonchoConfig = PrivateAttr(default=None)
@@ -185,7 +203,8 @@ class HonchoMemoryManager(MemoryManager):
             return
         user_peer = self._user_peer(user_id)
         assistant_peer = self._config.assistant_peer
-        outgoing: list[dict[str, str]] = []
+        outgoing: list[dict[str, Any]] = []
+        seen: set[str] = set()
         for message in messages or []:
             msg_type = getattr(message, "type", None)
             if msg_type == "human":
@@ -202,22 +221,46 @@ class HonchoMemoryManager(MemoryManager):
             if not text:
                 continue
             if msg_type == "human":
-                outgoing.append({"peer_id": user_peer, "content": text[: self._config.message_char_limit]})
+                peer_id = user_peer
             elif msg_type in ("ai", "AIMessageChunk"):
-                outgoing.append({"peer_id": assistant_peer, "content": text[: self._config.message_char_limit]})
+                peer_id = assistant_peer
+            else:
+                continue
+            item: dict[str, Any] = {"peer_id": peer_id, "content": text[: self._config.message_char_limit]}
+            message_id = getattr(message, "id", None)
+            if isinstance(message_id, str) and message_id:
+                # Include owner and role even when operators deliberately share
+                # workspace/peer overrides; edits of the same source remain new.
+                source = json.dumps([user_id, message_id, "human" if msg_type == "human" else "ai", peer_id, text], ensure_ascii=True)
+                source_id = hashlib.sha256(source.encode("utf-8")).hexdigest()
+                if source_id in seen:
+                    continue
+                seen.add(source_id)
+                item["metadata"] = {SOURCE_ID_KEY: source_id}
+            outgoing.append(item)
         if not outgoing:
             return
         session_id = f"df-{_stable_id(thread_id)}"
+        operation = "write"
         try:
-            self._client.get_or_create_peer(workspace, user_peer)
-            self._client.get_or_create_peer(workspace, assistant_peer)
-            self._client.get_or_create_session(workspace, session_id)
-            self._client.set_session_peers(workspace, session_id, [user_peer, assistant_peer])
-            self._client.add_messages(workspace, session_id, outgoing)
+            with _capture_lock(self._config.base_url, workspace, session_id):
+                self._client.get_or_create_peer(workspace, user_peer)
+                self._client.get_or_create_peer(workspace, assistant_peer)
+                self._client.get_or_create_session(workspace, session_id)
+                self._client.set_session_peers(workspace, session_id, [user_peer, assistant_peer])
+                for offset in range(0, len(outgoing), MESSAGE_BATCH_SIZE):
+                    batch = outgoing[offset : offset + MESSAGE_BATCH_SIZE]
+                    source_ids = [item["metadata"][SOURCE_ID_KEY] for item in batch if "metadata" in item]
+                    operation = "capture lookup"
+                    existing = self._client.existing_source_ids(workspace, session_id, source_ids) if source_ids else set()
+                    operation = "write"
+                    pending = [item for item in batch if item.get("metadata", {}).get(SOURCE_ID_KEY) not in existing]
+                    if pending:
+                        self._client.add_messages(workspace, session_id, pending)
         except MemoryManagerError:
             raise
         except Exception as exc:
-            logger.warning("honcho memory: write failed for thread %s: %s", thread_id, exc)
+            logger.warning("honcho memory: %s failed for thread %s: %s", operation, thread_id, exc)
 
     # ── Tier 1: read ─────────────────────────────────────────────────────
     def get_context(
@@ -251,10 +294,10 @@ class HonchoMemoryManager(MemoryManager):
 
         Known limitation: that filter only matches messages stored with
         ``metadata.category``, and nothing this backend writes has it — ``add()``
-        sends plain ``{"peer_id", "content"}`` messages with no metadata, and
+        sends source-identity metadata but no category, and
         fact CRUD is unsupported, so no other write path can set it either. A
         ``category`` query therefore returns empty results in this backend
-        until the write path gains metadata support."""
+        until the write path gains category support."""
         workspace = self._workspace(user_id)
         if workspace is None:
             return []

backend/tests/blocking_io/test_honcho_memory_backend.py

@@ -16,6 +16,8 @@ class _BlockingFakeClient:
 
     def __init__(self, probe_dir: Path):
         self._probe = probe_dir / "probe.txt"
+        self.looked_up = False
+        self.written = False
 
     def _block(self) -> None:
         self._probe.write_text("blocked io")
@@ -31,6 +33,12 @@ class _BlockingFakeClient:
 
     def add_messages(self, ws: str, sid: str, msgs: list[Any]) -> None:
         self._block()
+        self.written = True
+
+    def existing_source_ids(self, ws: str, sid: str, source_ids: list[str]) -> set[str]:
+        self._block()
+        self.looked_up = True
+        return set()
 
     def working_representation(self, ws: str, pid: str, *, max_conclusions: int = 25) -> str:
         self._block()
@@ -65,6 +73,8 @@ async def test_async_honcho_operations_do_not_block_event_loop(
         messages,
         user_id="alice",
     )
+    assert manager._client.looked_up
+    assert manager._client.written
     assert await manager.aget_context("alice") is not None
     assert (
         await manager.asearch(

backend/tests/blocking_io/test_summarization_memory_flush.py

@@ -65,6 +65,10 @@ class _FakeHonchoClient:
     def set_session_peers(self, ws: str, sid: str, pids: list[str]) -> None:
         pass
 
+    def existing_source_ids(self, ws: str, sid: str, source_ids: list[str]) -> set[str]:
+        self._writes.write([])
+        return set()
+
     def add_messages(self, ws: str, sid: str, msgs: list[dict[str, str]]) -> None:
         self._writes.write([message["content"] for message in msgs])
 

backend/tests/test_honcho_incremental_capture.py

@@ -0,0 +1,340 @@
+"""Honcho captures checkpoint history incrementally through the real HTTP client."""
+
+import asyncio
+import json
+import math
+from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor
+from threading import Barrier
+
+import httpx
+import pytest
+from langchain.agents import create_agent
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import AIMessage, HumanMessage, messages_to_dict
+from langgraph.checkpoint.memory import InMemorySaver
+
+from deerflow.agents.memory.backends.honcho import honcho_manager
+from deerflow.agents.memory.backends.honcho.client import HonchoClient
+from deerflow.agents.middlewares import memory_middleware
+from deerflow.agents.middlewares.memory_middleware import MemoryMiddleware
+from deerflow.config.memory_config import MemoryConfig
+
+SOURCE_ID = "deerflow_source_id"
+
+
+class MessageStore:
+    """Emulate append-only messages and Honcho's metadata-filtered offset pages."""
+
+    def __init__(self):
+        self.messages = defaultdict(list)
+        self.batches = []
+        self.lookups = []
+        self.fail_write = None
+        self.write_count = 0
+        self.lose_response = False
+        self.list_error = None
+        self.page_size = 2
+
+    def receive(self, request):
+        path = request.url.path
+        body = json.loads(request.content)
+        if path.endswith("/messages/list"):
+            self.lookups.append(request)
+            if self.list_error is not None:
+                return self.list_error
+            filters = body["filters"]["OR"]
+            assert 0 < len(filters) <= 100
+            keys = {entry["metadata"][SOURCE_ID] for entry in filters}
+            items = [item for item in self.messages[path.removesuffix("/list")] if item.get("metadata", {}).get(SOURCE_ID) in keys]
+            page = int(request.url.params["page"])
+            size = min(int(request.url.params["size"]), self.page_size)
+            return httpx.Response(200, json={"items": items[(page - 1) * size : page * size], "page": page, "pages": math.ceil(len(items) / size), "total": len(items)})
+        if path.endswith("/messages"):
+            incoming = body["messages"]
+            if not 0 < len(incoming) <= 100:
+                return httpx.Response(422, json={"detail": "message batch size"})
+            self.write_count += 1
+            if self.write_count == self.fail_write and not self.lose_response:
+                return httpx.Response(503, json={"detail": "unavailable"})
+            self.messages[path].extend(incoming)
+            self.batches.append(incoming)
+            if self.write_count == self.fail_write:
+                raise httpx.ReadTimeout("response lost after commit", request=request)
+        return httpx.Response(200, json={})
+
+    def contents(self):
+        return [item["content"] for batch in self.batches for item in batch]
+
+
+@pytest.fixture
+def capture(monkeypatch):
+    store = MessageStore()
+    managers = []
+    monkeypatch.setattr(honcho_manager, "HonchoClient", lambda config: HonchoClient(config, transport=httpx.MockTransport(store.receive)))
+
+    def create(**config):
+        manager = honcho_manager.HonchoMemoryManager.from_config({"base_url": "https://honcho.invalid", **config})
+        managers.append(manager)
+        monkeypatch.setattr(memory_middleware, "get_memory_manager", lambda: manager)
+        return manager
+
+    create()
+    yield create, store
+    for manager in managers:
+        manager.close()
+
+
+def run_turns(*, asynchronous, repeat_text=False, separate_threads=False, turns=3):
+    agent = create_agent(
+        model=FakeMessagesListChatModel(responses=[AIMessage(content=f"Answer {n}") for n in range(turns)]),
+        middleware=[MemoryMiddleware(memory_config=MemoryConfig(enabled=True))],
+        checkpointer=InMemorySaver(),
+    )
+
+    async def run():
+        for n in range(turns):
+            thread_id = f"thread-{n}" if separate_threads else "thread-1"
+            state = {"messages": [HumanMessage(content="Please continue" if repeat_text else f"Question {n}")]}
+            config = {"configurable": {"thread_id": thread_id}}
+            context = {"thread_id": thread_id, "user_id": "repro-user"}
+            if asynchronous:
+                await agent.ainvoke(state, config, context=context)
+            else:
+                agent.invoke(state, config, context=context)
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"])
+def test_acknowledged_history_is_not_resubmitted(capture, asynchronous):
+    _, store = capture
+    run_turns(asynchronous=asynchronous)
+    assert [len(batch) for batch in store.batches] == [2, 2, 2]
+
+
+@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"])
+def test_same_text_in_distinct_turns_is_preserved_once_per_turn(capture, asynchronous):
+    _, store = capture
+    run_turns(asynchronous=asynchronous, repeat_text=True)
+    assert store.contents().count("Please continue") == 3
+
+
+@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"])
+def test_first_turn_control(capture, asynchronous):
+    _, store = capture
+    run_turns(asynchronous=asynchronous, turns=1)
+    assert [len(batch) for batch in store.batches] == [2]
+
+
+@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"])
+def test_independent_threads_control(capture, asynchronous):
+    _, store = capture
+    run_turns(asynchronous=asynchronous, separate_threads=True, repeat_text=True)
+    assert [len(batch) for batch in store.batches] == [2, 2, 2]
+    assert len(store.messages) == 3
+
+
+def test_restart_and_compacted_overlap_use_remote_identity(capture):
+    create, store = capture
+    messages = [HumanMessage("question", id="h1"), AIMessage("answer", id="a1"), HumanMessage("next", id="h2")]
+    original = messages_to_dict(messages)
+    first = create()
+    first.add("thread", messages[:2], user_id="alice")
+    first.close()
+    second = create()
+    second.add("thread", messages[1:], user_id="alice")
+    second.add_nowait("thread", messages, user_id="alice")
+    assert store.contents() == ["question", "answer", "next"]
+    assert any(request.url.params["page"] == "2" for request in store.lookups)
+    assert messages_to_dict(messages) == original
+
+
+def test_distinct_sources_edits_and_roles_are_not_text_deduplicated(capture):
+    create, store = capture
+    manager = create()
+    h1 = HumanMessage("same", id="h1")
+    messages = [h1, h1, HumanMessage("same", id="h2"), HumanMessage("edited", id="h1"), AIMessage("same", id="h1")]
+    manager.add("thread", messages, user_id="alice")
+    manager.add("thread", messages, user_id="alice")
+    assert store.contents() == ["same", "same", "edited", "same"]
+    assert len({item["metadata"][SOURCE_ID] for item in store.batches[0]}) == 4
+
+
+def test_shared_workspace_peer_and_message_ids_still_preserve_each_user(capture):
+    create, store = capture
+    manager = create(workspace_overrides={"alice": "shared", "bob": "shared"}, user_peer_overrides={"alice": "shared-peer", "bob": "shared-peer"})
+    messages = [HumanMessage("same", id="same")]
+    for user in ("alice", "bob", "alice", "bob"):
+        manager.add("thread", messages, user_id=user)
+    assert store.contents() == ["same", "same"]
+
+
+def test_idless_callers_keep_append_behavior(capture):
+    create, store = capture
+    manager = create()
+    for _ in range(2):
+        manager.add("thread", [HumanMessage("same")], user_id="alice")
+    assert store.contents() == ["same", "same"]
+    assert not store.lookups
+
+
+def test_same_source_id_in_independent_threads_is_not_suppressed(capture):
+    create, store = capture
+    manager = create()
+    message = HumanMessage("same", id="h1")
+    for thread_id in ("first", "second", "first", "second"):
+        manager.add(thread_id, [message], user_id="alice")
+    assert store.contents() == ["same", "same"]
+
+
+def test_malformed_later_lookup_page_does_not_publish_partial_results(capture, monkeypatch):
+    create, store = capture
+    manager = create()
+    messages = [HumanMessage(str(n), id=str(n)) for n in range(4)]
+    manager.add("thread", messages[:3], user_id="alice")
+    original = manager._client._post
+
+    def post(path, payload):
+        if "/messages/list?page=2" in path:
+            return {"items": [None], "page": 2, "pages": 2}
+        return original(path, payload)
+
+    monkeypatch.setattr(manager._client, "_post", post)
+    manager.add("thread", messages, user_id="alice")
+    assert store.contents() == ["0", "1", "2"]
+    monkeypatch.setattr(manager._client, "_post", original)
+    manager.add("thread", messages, user_id="alice")
+    assert store.contents() == ["0", "1", "2", "3"]
+
+
+@pytest.mark.parametrize("lost_response", [False, True])
+def test_retry_reconciles_failed_or_committed_writes(capture, lost_response, caplog):
+    create, store = capture
+    messages = [HumanMessage("first", id="h1"), AIMessage("reply", id="a1"), HumanMessage("next", id="h2")]
+    store.fail_write = 1
+    store.lose_response = lost_response
+    create().add("thread", messages[:2], user_id="alice")
+    assert "write failed for thread thread" in caplog.text
+    assert "capture lookup failed" not in caplog.text
+    create().add("thread", messages, user_id="alice")
+    assert store.contents() == ["first", "reply", "next"]
+
+
+def test_batched_retry_preserves_already_committed_prefix(capture):
+    create, store = capture
+    store.page_size = 10
+    messages = [HumanMessage(f"message {n}", id=f"h{n}") for n in range(105)]
+    store.fail_write = 2
+    create().add("thread", messages, user_id="alice")
+    assert len(store.contents()) == 100
+    create().add("thread", messages, user_id="alice")
+    assert store.contents() == [message.content for message in messages]
+    assert [len(batch) for batch in store.batches] == [100, 5]
+
+
+@pytest.mark.parametrize("growing", [False, True], ids=["excessive-total", "growing-total"])
+def test_lookup_page_limit_aborts_capture_without_partial_results(capture, monkeypatch, caplog, growing):
+    create, store = capture
+    manager = create()
+    message = HumanMessage("private-history", id="h1")
+    manager.add("thread", [message], user_id="alice")
+    original = manager._client._post
+    requested_pages = []
+
+    def post(path, payload):
+        if "/messages/list?" not in path:
+            return original(path, payload)
+        page = int(httpx.URL(path).params["page"])
+        requested_pages.append(page)
+        # Keep the pre-fix implementation finite without providing its bound.
+        if len(requested_pages) > 10:
+            raise AssertionError("lookup exceeded the test's request budget")
+        return {"items": store.batches[0], "page": page, "pages": page + 1 if growing else 1_000_000}
+
+    monkeypatch.setattr(manager._client, "_post", post)
+    manager.add("thread", [message, HumanMessage("next", id="h2")], user_id="alice")
+    assert requested_pages == (list(range(1, 11)) if growing else [1])
+    assert store.write_count == 1
+    assert store.contents() == ["private-history"]
+    assert "capture lookup failed" in caplog.text
+    assert "page limit" in caplog.text
+    assert "private-history" not in caplog.text
+    monkeypatch.setattr(manager._client, "_post", original)
+    manager.add("thread", [message, HumanMessage("next", id="h2")], user_id="alice")
+    assert store.contents() == ["private-history", "next"]
+
+
+def test_lookup_accepts_duplicate_source_records_through_last_allowed_page(capture):
+    create, store = capture
+    manager = create()
+    message = HumanMessage("first", id="h1")
+    manager.add("thread", [message], user_id="alice")
+    records = next(iter(store.messages.values()))
+    records.extend([records[0]] * 9)
+    store.page_size = 1
+    store.lookups.clear()
+    manager.add("thread", [message, HumanMessage("next", id="h2")], user_id="alice")
+    assert [int(request.url.params["page"]) for request in store.lookups] == list(range(1, 11))
+    assert store.contents() == ["first", "next"]
+
+
+@pytest.mark.parametrize(
+    "response",
+    [
+        httpx.Response(403, json={"detail": "session read permission required"}),
+        httpx.Response(503, json={"detail": "unavailable"}),
+        httpx.Response(200, json={}),
+        httpx.Response(200, json={"items": "invalid", "page": 1, "pages": 1}),
+        httpx.Response(200, json={"items": [], "page": 1, "pages": True}),
+        httpx.Response(200, json={"items": [], "page": 2, "pages": 2}),
+        httpx.Response(200, json={"items": [{"metadata": {SOURCE_ID: "not-requested"}}], "page": 1, "pages": 1}),
+    ],
+)
+def test_lookup_failure_does_not_blindly_replay(capture, response, caplog):
+    create, store = capture
+    manager = create()
+    messages = [HumanMessage("private-history", id="h1")]
+    manager.add("thread", messages, user_id="alice")
+    store.list_error = response
+    manager.add("thread", messages + [HumanMessage("next", id="h2")], user_id="alice")
+    assert store.contents() == ["private-history"]
+    assert store.write_count == 1
+    assert "capture lookup failed for thread thread" in caplog.text
+    assert "write failed" not in caplog.text
+    assert "private-history" not in caplog.text
+    store.list_error = None
+    manager.add("thread", messages + [HumanMessage("next", id="h2")], user_id="alice")
+    assert store.contents() == ["private-history", "next"]
+
+
+def test_concurrent_managers_share_session_capture_lock(capture, monkeypatch):
+    create, store = capture
+    managers = [create(), create()]
+    start = Barrier(2)
+    messages = [HumanMessage("hello", id="h1")]
+    original = HonchoClient.add_messages
+
+    def write(client, *args):
+        # The publication must still own the same session lock as the lookup.
+        lock = honcho_manager._capture_lock("https://honcho.invalid", args[0], args[1])
+        with ThreadPoolExecutor(max_workers=1) as probe:
+
+            def acquire_elsewhere():
+                acquired = lock.acquire(blocking=False)
+                if acquired:
+                    lock.release()
+                return acquired
+
+            assert not probe.submit(acquire_elsewhere).result(timeout=5)
+        original(client, *args)
+
+    monkeypatch.setattr(HonchoClient, "add_messages", write)
+
+    def run(manager):
+        start.wait(timeout=5)
+        manager.add("thread", messages, user_id="alice")
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        list(pool.map(run, managers))
+    assert store.contents() == ["hello"]

backend/tests/test_honcho_memory_backend.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import json
 from types import SimpleNamespace
 
@@ -393,6 +394,7 @@ class _FakeClient:
 
     def __init__(self):
         self.calls: list[tuple[str, tuple]] = []
+        self.message_batches: list[list[dict]] = []
         self.representation_text = "ajayr likes concise answers"
         self.raise_on: str | None = None
         # Which exception type _maybe_raise raises; default matches the real
@@ -422,6 +424,12 @@ class _FakeClient:
     def add_messages(self, ws, sid, msgs):
         self._maybe_raise("messages")
         self.calls.append(("messages", (ws, sid, tuple((m["peer_id"], m["content"]) for m in msgs))))
+        self.message_batches.append(msgs)
+
+    def existing_source_ids(self, ws, sid, source_ids):
+        self._maybe_raise("lookup")
+        self.calls.append(("lookup", (ws, sid, tuple(source_ids))))
+        return set()
 
     def working_representation(self, ws, pid, *, max_conclusions=25):
         self._maybe_raise("representation")
@@ -465,6 +473,18 @@ class TestHonchoClientTransport:
 
 
 class TestHonchoManagerWrite:
+    def test_add_with_source_id_looks_up_before_writing_metadata(self):
+        mgr, fake = _manager(workspace_overrides={"u1": "shared"}, user_peer_overrides={"u1": "alice"})
+        mgr.add("t-1", [SimpleNamespace(type="human", content=" hello ", id="h1")], user_id="u1")
+        sid = f"df-{_stable_id('t-1')}"
+        source_id = hashlib.sha256(json.dumps(["u1", "h1", "human", "alice", "hello"]).encode("utf-8")).hexdigest()
+        assert ("messages", ("shared", sid, (("alice", "hello"),))) in fake.calls
+        assert fake.calls[-2:] == [
+            ("lookup", ("shared", sid, (source_id,))),
+            ("messages", ("shared", sid, (("alice", "hello"),))),
+        ]
+        assert fake.message_batches == [[{"peer_id": "alice", "content": "hello", "metadata": {"deerflow_source_id": source_id}}]]
+
     def test_add_maps_messages_to_peers(self):
         mgr, fake = _manager(workspace_overrides={"u1": "shared"}, user_peer_overrides={"u1": "alice"})
         mgr.add("t-1", [_msg("human", "please check the deploy"), _msg("ai", "deploy is green"), _msg("tool", "ignored")], user_id="u1")