From f7da737118ee3dc68ab51ca9ec2e0a9c931d6bbc Mon Sep 17 00:00:00 2001 From: haotianqi666 Date: Fri, 30 Oct 2026 07:33:41 +0800 Subject: [PATCH] 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> --- .../harness/deerflow/agents/memory/AGENTS.md | 11 + .../agents/memory/backends/honcho/README.md | 39 +- .../agents/memory/backends/honcho/client.py | 41 ++- .../memory/backends/honcho/honcho_manager.py | 73 +++- .../blocking_io/test_honcho_memory_backend.py | 10 + .../test_summarization_memory_flush.py | 4 + .../tests/test_honcho_incremental_capture.py | 340 ++++++++++++++++++ backend/tests/test_honcho_memory_backend.py | 20 ++ 8 files changed, 518 insertions(+), 20 deletions(-) create mode 100644 backend/tests/test_honcho_incremental_capture.py diff --git a/backend/packages/harness/deerflow/agents/memory/AGENTS.md b/backend/packages/harness/deerflow/agents/memory/AGENTS.md index 462fcb4a7e..25193bf2a0 100644 --- a/backend/packages/harness/deerflow/agents/memory/AGENTS.md +++ b/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; diff --git a/backend/packages/harness/deerflow/agents/memory/backends/honcho/README.md b/backend/packages/harness/deerflow/agents/memory/backends/honcho/README.md index 29e2de7f94..031bc7437a 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/honcho/README.md +++ b/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 diff --git a/backend/packages/harness/deerflow/agents/memory/backends/honcho/client.py b/backend/packages/harness/deerflow/agents/memory/backends/honcho/client.py index ccac6ee40e..8524a08a62 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/honcho/client.py +++ b/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): diff --git a/backend/packages/harness/deerflow/agents/memory/backends/honcho/honcho_manager.py b/backend/packages/harness/deerflow/agents/memory/backends/honcho/honcho_manager.py index d8a2c73e48..98cb0195f9 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/honcho/honcho_manager.py +++ b/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 [] diff --git a/backend/tests/blocking_io/test_honcho_memory_backend.py b/backend/tests/blocking_io/test_honcho_memory_backend.py index e175ac0900..4b221a9f05 100644 --- a/backend/tests/blocking_io/test_honcho_memory_backend.py +++ b/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( diff --git a/backend/tests/blocking_io/test_summarization_memory_flush.py b/backend/tests/blocking_io/test_summarization_memory_flush.py index 9f0363d0ec..60d31f5a68 100644 --- a/backend/tests/blocking_io/test_summarization_memory_flush.py +++ b/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]) diff --git a/backend/tests/test_honcho_incremental_capture.py b/backend/tests/test_honcho_incremental_capture.py new file mode 100644 index 0000000000..44a1c6f015 --- /dev/null +++ b/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"] diff --git a/backend/tests/test_honcho_memory_backend.py b/backend/tests/test_honcho_memory_backend.py index 109369c917..d302e170dd 100644 --- a/backend/tests/test_honcho_memory_backend.py +++ b/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")