diff --git a/docs/config.md b/docs/config.md index 99715107..762f3483 100644 --- a/docs/config.md +++ b/docs/config.md @@ -1152,16 +1152,17 @@ Sources pull data from external services on a schedule. See [sources.md](sources When active: - The `memorize` tool **dual-writes**: memU (as always) plus an async `write_async` to xmemory. Failures on the xmemory side never fail the tool. - `memory_recall` appends xmemory's read result (serialized as JSON) to memU's N items, run concurrently so the dual lookup is one round-trip. Read behavior is controlled via `xmemory.read_mode` (defaults to `single-answer`). -- The memorization **sweep** (session-close / cron) stays memU-only — it does not go through the `memorize` tool handler. +- The memorization **sweep** (session-close / cron) stays memU-only by default — it does not go through the `memorize` tool handler. Opting in via `xmemory.index_conversations` mirrors each swept message window to xmemory as a text-only transcript (role + content only — no thinking, no tool blocks/results), chunked and written with fast extraction, fire-and-forget alongside the memU pass. | Key | Type | Default | Description | |-----|------|---------|-------------| | `xmemory.api_key` | string | *(empty)* | xmemory bearer token (invite-only). Secret → `config.local.yaml`. | | `xmemory.instance_id` | string | *(empty)* | The xmemory instance to bind. Both this and `api_key` are required to activate. | | `xmemory.api_url` | string | `https://api.xmemory.ai` | API base URL. | -| `xmemory.extraction_logic` | string | `deep` | Write extraction mode: `deep` (accurate) or `fast` (high-volume). | +| `xmemory.extraction_logic` | string | `deep` | Write extraction mode for `memorize`-tool facts: `deep` (accurate) or `fast` (high-volume). Sweep transcripts always use `fast`. | | `xmemory.read_mode` | string | `single-answer` | Read mode for recall, whose result is appended as JSON: `single-answer` (synthesized answer envelope), `raw-tables` (table columns + rows), or `xresponse` (objects + relations). | | `xmemory.timeout` | float | `60.0` | Per-request timeout in seconds. | +| `xmemory.index_conversations` | bool | `false` | Mirror the memorization sweep's session transcripts (text only) to xmemory. Best-effort: a failed write is logged, never retried (the sweep watermark is memU's). Off by default — full transcripts leave the machine only when explicitly enabled. | ## Docker diff --git a/docs/memory.md b/docs/memory.md index 75c28fc5..7ba424ab 100644 --- a/docs/memory.md +++ b/docs/memory.md @@ -192,7 +192,7 @@ When active: - **`memorize` dual-writes** — the fact goes to memU (file + extraction, as always) **and** is enqueued to xmemory via async `write_async`. xmemory failures are swallowed (logged) so the tool never fails on them. The result message shows `(+ xmemory)` when the xmemory enqueue succeeded. - **`memory_recall` adds xmemory read output** — alongside memU's N items + category breadcrumbs, recall runs xmemory **concurrently** and appends a `[xmemory]` section holding the read result serialized as JSON. Read behavior is controlled by `xmemory.read_mode` (`single-answer` by default, a synthesized answer envelope; `raw-tables` returns table columns + rows, `xresponse` returns objects + relations). The bridge does not parse the result — the shape differs per mode, so it serializes whatever the reader returns. When xmemory is disabled or returns nothing, recall output is byte-for-byte the memU-only shape. -- **The memorization sweep stays memU-only** — session-close / cron indexing goes through the bridge directly, not the `memorize` tool handler, so it is unaffected. +- **The memorization sweep stays memU-only by default** — session-close / cron indexing goes through the bridge directly, not the `memorize` tool handler. Setting `xmemory.index_conversations: true` additionally mirrors every message window the sweep indexes into memU to xmemory as a **text-only transcript**: each message contributes role + content (+ timestamp) only — thinking and tool blocks/results never leave the machine, exactly like the memU sweep payload. Transcripts are chunked (~64 KB per write) and enqueued with **fast** extraction regardless of `xmemory.extraction_logic` (which still governs `memorize`-tool writes), as a fire-and-forget background task that never blocks or fails the memU pass. Delivery is best-effort: the sweep watermark belongs to memU, so a window whose xmemory write fails is not retried. Implementation: `nerve/memory/xmemory_bridge.py` (`XmemoryBridge`), wired into `ToolContext.xmemory_bridge` next to `memory_bridge`. Backed by the `xmemory-ai` SDK (`AsyncXmemoryClient`). diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index a5376933..6ca9f599 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -794,6 +794,7 @@ async def _memorize_session( await self._memory_bridge.memorize_conversation( session_id, context_msgs, ) + self.schedule_xmemory_transcript(session_id, context_msgs) logger.info( "Indexed %d messages from session %s into memU", len(context_msgs), session_id, @@ -850,6 +851,37 @@ def _done(t: asyncio.Task) -> None: task.add_done_callback(_done) + def schedule_xmemory_transcript( + self, session_id: str, messages: list[dict], + ) -> None: + """Mirror a just-memorized message window to xmemory (fire-and-forget). + + Inert unless ``xmemory.index_conversations`` is opted in. Runs as a + background task so a slow or failing xmemory never extends the global + memorize lock; the bridge isolates its own errors and sends text only + (role + content — never thinking or tool blocks/results). Best-effort + by design: the sweep watermark is memU's, so a window lost here (task + failure or shutdown cancellation) is not retried for xmemory. + """ + bridge = self._xmemory_bridge + if bridge is None or not bridge.indexes_conversations or not messages: + return + + task = asyncio.create_task( + bridge.memorize_conversation(session_id, messages), + ) + self._memorize_bg_tasks.add(task) + + def _done(t: asyncio.Task) -> None: + self._memorize_bg_tasks.discard(t) + if not t.cancelled() and t.exception() is not None: + logger.warning( + "xmemory transcript write failed for session %s: %s", + session_id, t.exception(), + ) + + task.add_done_callback(_done) + async def _memorize_incremental(self, session_id: str) -> int: """Index only messages newer than last_memorized_at into memU. @@ -886,6 +918,7 @@ async def _memorize_incremental(self, session_id: str) -> int: await self._memory_bridge.memorize_conversation( session_id, new_msgs, ) + self.schedule_xmemory_transcript(session_id, new_msgs) if latest_ts: await self.db.update_session_fields( diff --git a/nerve/agent/tools/handlers/memory.py b/nerve/agent/tools/handlers/memory.py index fe36b3b9..1d1f1cd6 100644 --- a/nerve/agent/tools/handlers/memory.py +++ b/nerve/agent/tools/handlers/memory.py @@ -425,8 +425,9 @@ def _write_memorize_file() -> None: # Optional dual-write to xmemory (async, fire-and-forget). Independent of # the memU outcome and never fails the tool; inert when xmemory is - # disabled. The memorization *sweep* does not go through this handler, so - # it stays memU-only as intended. + # disabled. The memorization *sweep* does not go through this handler — + # its transcripts reach xmemory only via the separate opt-in path + # (``xmemory.index_conversations`` → XmemoryBridge.memorize_conversation). xmem_written = False if ctx.xmemory_bridge is not None and ctx.xmemory_bridge.available: xmem_written = await ctx.xmemory_bridge.memorize(f"{memory_type}: {content}") diff --git a/nerve/config.py b/nerve/config.py index feefae8f..7db36d3b 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -2496,7 +2496,11 @@ class XmemoryConfig: Activated only when both ``api_key`` (the bearer token) and ``instance_id`` are set. When active, the ``memorize`` tool dual-writes to xmemory (async) and ``memory_recall`` appends xmemory's synthesized - answer to the memU results. The memorization sweep stays memU-only. + answer to the memU results. The memorization sweep stays memU-only + unless ``index_conversations`` is additionally set: then every message + window the sweep indexes into memU is also mirrored to xmemory as a + text-only transcript (role + content only — thinking and tool + blocks/results are never sent). Empty keys = no-op, zero overhead, no SDK calls. The instance and its schema are created out of band (by the operator) on xmemory's side. @@ -2508,6 +2512,10 @@ class XmemoryConfig: extraction_logic: str = "deep" # "deep" (default) or "fast" read_mode: str = "single-answer" # "single-answer" | "raw-tables" | "xresponse" timeout: float = 60.0 + # Opt-in: mirror the memorization sweep's session transcripts to xmemory. + # Off by default — full transcripts leave the machine only when the + # operator explicitly enables it. + index_conversations: bool = False @property def enabled(self) -> bool: @@ -2524,6 +2532,7 @@ def from_dict(cls, d: dict) -> "XmemoryConfig": extraction_logic=d.get("extraction_logic", "deep"), read_mode=d.get("read_mode", "single-answer"), timeout=d.get("timeout", 60.0), + index_conversations=d.get("index_conversations", False), ) diff --git a/nerve/gateway/routes/sessions.py b/nerve/gateway/routes/sessions.py index bfaabb10..c49e0c90 100644 --- a/nerve/gateway/routes/sessions.py +++ b/nerve/gateway/routes/sessions.py @@ -381,6 +381,8 @@ async def _bg_memorize(): context_msgs.append(msg) if context_msgs: await engine._memory_bridge.memorize_conversation(session_id, context_msgs) + # Same window to xmemory (opt-in, fire-and-forget). + engine.schedule_xmemory_transcript(session_id, context_msgs) except Exception as e: logger.warning("Background memorize for deleted session %s failed: %s", session_id, e) asyncio.create_task(_bg_memorize()) diff --git a/nerve/memory/xmemory_bridge.py b/nerve/memory/xmemory_bridge.py index 36895c71..d2fa3bb0 100644 --- a/nerve/memory/xmemory_bridge.py +++ b/nerve/memory/xmemory_bridge.py @@ -15,8 +15,13 @@ query. The read mode is configurable (``xmemory.read_mode``): a synthesized natural-language answer by default (``single-answer``), or the structured ``raw-tables`` / ``xresponse`` payloads. -* The memorization *sweep* (session-close, cron) stays memU-only — it - never goes through the ``memorize`` tool handler, so it's untouched. +* The memorization *sweep* (session-close, cron) is memU-only by default. + With ``xmemory.index_conversations`` set, every message window the sweep + indexes into memU is also mirrored here as a **text-only** transcript + (:meth:`XmemoryBridge.memorize_conversation`): role + content only — + thinking and tool blocks/results never leave the box. Transcripts are + chunked and written with FAST extraction (they are high-volume; the + configured ``extraction_logic`` still governs the memorize tool). The bridge is inert unless ``config.xmemory.enabled`` (both an API token and an ``instance_id`` are set). Every xmemory call is wrapped so a slow @@ -34,6 +39,11 @@ logger = logging.getLogger(__name__) +# Soft byte budget per transcript write job (headers may push a chunk a few +# dozen bytes over). Sized so each xmemory extraction sees a coherent slice +# of conversation (~16K tokens) while staying well under request-size caps. +_TRANSCRIPT_CHUNK_BYTES = 64_000 + class XmemoryBridge: """Thin async wrapper around the ``xmemory-ai`` SDK. @@ -127,6 +137,12 @@ def available(self) -> bool: """True when xmemory is configured, imported, and bound.""" return self._available and self._instance is not None + @property + def indexes_conversations(self) -> bool: + """True when the bridge is available AND transcript mirroring is + opted in via ``xmemory.index_conversations``.""" + return self.available and self._config.index_conversations + # ------------------------------------------------------------------ # # Data ops # ------------------------------------------------------------------ # @@ -171,6 +187,52 @@ async def memorize(self, text: str) -> bool: logger.warning("xmemory write_async failed: %s", e) return False + async def memorize_conversation(self, session_id: str, messages: list[dict]) -> int: + """Mirror a session-transcript window to xmemory as free-text writes. + + Same text-only contract as the memU sweep: each message contributes + ``role`` + ``content`` (+ ``created_at`` when present) — ``thinking`` + and ``blocks`` (tool calls/results, images) are never sent. Long + transcripts are split at message boundaries into + ~``_TRANSCRIPT_CHUNK_BYTES`` chunks, each enqueued via ``write_async`` + with FAST extraction (transcripts are high-volume; the configured + ``extraction_logic`` still governs the memorize tool's writes). + + Opt-in via ``xmemory.index_conversations`` and best-effort by design: + a failed chunk is logged, the remaining chunks are abandoned, and the + window is never retried for xmemory (the sweep watermark is memU's). + Returns the number of chunks successfully enqueued (0 when disabled, + empty, or on an immediate failure). + """ + if not self.indexes_conversations or not messages: + return 0 + chunks = _transcript_chunks( + session_id, messages, chunk_bytes=_TRANSCRIPT_CHUNK_BYTES, + ) + if not chunks: + return 0 + sent = 0 + for chunk in chunks: + try: + await self._instance.write_async( + chunk, extraction_logic=self._ExtractionLogic.FAST, + ) + sent += 1 + except Exception as e: + logger.warning( + "xmemory transcript write failed for session %s " + "(chunk %d/%d): %s — abandoning remaining chunks", + session_id, sent + 1, len(chunks), e, + ) + break + if sent: + logger.info( + "xmemory: enqueued transcript for session %s " + "(%d message(s), %d/%d chunk(s))", + session_id, len(messages), sent, len(chunks), + ) + return sent + async def recall_answer(self, query: str) -> str | None: """Query xmemory and return its read result serialized as JSON. @@ -233,3 +295,73 @@ def _json_default(value: Any) -> Any: if hasattr(value, "__dict__"): return value.__dict__ return str(value) + + +def _transcript_lines(messages: list[dict]) -> list[str]: + """Flatten message rows into text-only transcript lines. + + Mirrors the memU sweep's payload contract (see + ``MemUBridge.memorize_conversation``): only ``role`` + ``content`` + (+ ``created_at`` when present) survive. ``thinking`` and ``blocks`` + (tool calls/results, images) are deliberately dropped, and messages + with empty content are skipped. + """ + lines: list[str] = [] + for msg in messages: + content = str(msg.get("content") or "").strip() + if not content: + continue + role = msg.get("role") or "unknown" + created_at = msg.get("created_at") + prefix = f"[{created_at}] {role}" if created_at else str(role) + lines.append(f"{prefix}: {content}") + return lines + + +def _transcript_chunks( + session_id: str, + messages: list[dict], + chunk_bytes: int = _TRANSCRIPT_CHUNK_BYTES, +) -> list[str]: + """Split a transcript into write-sized chunks of ~``chunk_bytes`` each. + + Splits at message boundaries so each extraction sees whole messages; a + single message larger than the budget is hard-split on byte boundaries + (multibyte characters straddling a cut are dropped, matching the recall + handler's clipping). Each chunk opens with a one-line header carrying + the session id and, for multi-chunk transcripts, its position — enough + context for xmemory's extraction to relate the parts. + """ + lines = _transcript_lines(messages) + if not lines: + return [] + + # Message-boundary pieces, hard-splitting any single oversized line. + parts: list[str] = [] + for line in lines: + data = line.encode("utf-8") + if len(data) <= chunk_bytes: + parts.append(line) + else: + parts.extend( + data[i : i + chunk_bytes].decode("utf-8", "ignore") + for i in range(0, len(data), chunk_bytes) + ) + + groups: list[list[str]] = [[]] + size = 0 + for part in parts: + n = len(part.encode("utf-8")) + 1 # +1 for the joining newline + if size and size + n > chunk_bytes: + groups.append([]) + size = 0 + groups[-1].append(part) + size += n + + total = len(groups) + chunks: list[str] = [] + for i, group in enumerate(groups, start=1): + position = f", part {i}/{total}" if total > 1 else "" + header = f"Conversation transcript (session {session_id}{position}):\n" + chunks.append(header + "\n".join(group)) + return chunks diff --git a/tests/test_xmemory_bridge.py b/tests/test_xmemory_bridge.py index 61a8f58e..ee49abaf 100644 --- a/tests/test_xmemory_bridge.py +++ b/tests/test_xmemory_bridge.py @@ -1,17 +1,22 @@ """Tests for the optional xmemory.ai memory layer. xmemory runs *alongside* memU, never replacing it: -* ``memorize`` dual-writes (memU + xmemory async), and -* ``memory_recall`` appends xmemory's read output to memU's hits. +* ``memorize`` dual-writes (memU + xmemory async), +* ``memory_recall`` appends xmemory's read output to memU's hits, and +* with ``index_conversations`` opted in, the memorization sweep mirrors + its text-only session transcripts to xmemory. -These tests lock in three contracts: (1) the bridge is inert unless both a +These tests lock in four contracts: (1) the bridge is inert unless both a token and an instance_id are configured, (2) every xmemory failure is -isolated so memU recall/memorize still works, and (3) the handlers combine -both sources without regressing the memU-only output shape. +isolated so memU recall/memorize still works, (3) the handlers combine +both sources without regressing the memU-only output shape, and (4) +transcript mirroring is opt-in, text-only (no thinking / tool blocks), +chunked, and best-effort. """ from __future__ import annotations +import asyncio import json import sys from pathlib import Path @@ -20,13 +25,19 @@ import pytest +from nerve.agent.engine import AgentEngine from nerve.agent.tools.handlers.memory import ( memorize_handler, memory_recall_handler, ) from nerve.agent.tools.registry import ToolContext from nerve.config import NerveConfig, XmemoryConfig -from nerve.memory.xmemory_bridge import XmemoryBridge, _serialize_read_payload +from nerve.memory.xmemory_bridge import ( + XmemoryBridge, + _serialize_read_payload, + _transcript_chunks, + _transcript_lines, +) # --------------------------------------------------------------------------- # @@ -48,6 +59,7 @@ def test_config_from_dict_defaults_and_overrides() -> None: assert c.extraction_logic == "deep" assert c.read_mode == "single-answer" assert c.timeout == 60.0 + assert c.index_conversations is False # transcripts are strictly opt-in c2 = XmemoryConfig.from_dict({ "api_key": "tok", @@ -56,10 +68,12 @@ def test_config_from_dict_defaults_and_overrides() -> None: "extraction_logic": "fast", "read_mode": "raw-tables", "timeout": 30, + "index_conversations": True, }) assert c2.enabled and c2.api_url == "https://example.test" assert c2.extraction_logic == "fast" and c2.read_mode == "raw-tables" assert c2.timeout == 30.0 + assert c2.index_conversations is True def test_nerveconfig_wires_xmemory_block() -> None: @@ -167,6 +181,7 @@ async def test_bridge_disabled_when_package_missing(monkeypatch) -> None: async def _enabled_bridge( extraction_logic: str = "deep", read_mode: str = "single-answer", # mirrors the production default + index_conversations: bool = False, # mirrors the production default ) -> XmemoryBridge: """Build a bridge bound to a (fake-token) real client, then mock the instance handle so reads/writes never hit the network.""" @@ -175,6 +190,7 @@ async def _enabled_bridge( instance_id="inst_1", extraction_logic=extraction_logic, read_mode=read_mode, + index_conversations=index_conversations, ) bridge = XmemoryBridge(cfg) await bridge.initialize() # client + .instance() are network-free @@ -448,3 +464,306 @@ async def test_memorize_handler_succeeds_even_if_xmemory_write_fails(monkeypatch text = result.content[0]["text"] assert "Memorized: fact" in text assert "(+ xmemory)" not in text # memU still succeeded, xmemory silently skipped + + +# --------------------------------------------------------------------------- # +# Transcript helpers — text-only flattening and chunking +# --------------------------------------------------------------------------- # + + +def test_transcript_lines_are_text_only() -> None: + """Only role + content (+ timestamp) survive — the same contract as the + memU sweep. Thinking and tool blocks/results must never be included.""" + msgs = [ + { + "role": "user", + "content": "hello", + "created_at": "2026-01-01 00:00:00", + "thinking": "PRIVATE-REASONING", + "blocks": [{"type": "tool_result", "text": "RAW-TOOL-DUMP"}], + }, + {"role": "assistant", "content": ""}, # empty → skipped + {"role": "assistant", "content": "hi there"}, # no timestamp → bare role + ] + lines = _transcript_lines(msgs) + assert lines == [ + "[2026-01-01 00:00:00] user: hello", + "assistant: hi there", + ] + joined = "\n".join(lines) + assert "PRIVATE-REASONING" not in joined + assert "RAW-TOOL-DUMP" not in joined + + +def test_transcript_chunks_single_chunk_header() -> None: + chunks = _transcript_chunks("s-1", [{"role": "user", "content": "hello"}]) + assert len(chunks) == 1 + header = chunks[0].splitlines()[0] + assert header == "Conversation transcript (session s-1):" + assert "user: hello" in chunks[0] + + +def test_transcript_chunks_split_at_message_boundaries() -> None: + msgs = [{"role": "user", "content": f"m{i} " + "x" * 30} for i in range(10)] + chunks = _transcript_chunks("s-1", msgs, chunk_bytes=80) + assert len(chunks) > 1 + total = len(chunks) + for i, chunk in enumerate(chunks, start=1): + header, body = chunk.split("\n", 1) + assert header == f"Conversation transcript (session s-1, part {i}/{total}):" + assert len(body.encode("utf-8")) <= 80 # body respects the budget + combined = "\n".join(c.split("\n", 1)[1] for c in chunks) + for i in range(10): + assert f"m{i} " in combined # every message survives, exactly once each + + +def test_transcript_chunks_hard_split_oversized_message() -> None: + """A single message larger than the budget is split rather than dropped.""" + msgs = [{"role": "user", "content": "A" * 200}] + chunks = _transcript_chunks("s-1", msgs, chunk_bytes=80) + assert len(chunks) >= 3 + rejoined = "".join(c.split("\n", 1)[1] for c in chunks) + assert "A" * 200 in rejoined # nothing lost (ASCII → no boundary drops) + + +def test_transcript_chunks_empty_transcript() -> None: + assert _transcript_chunks("s-1", []) == [] + assert _transcript_chunks("s-1", [{"role": "user", "content": ""}]) == [] + + +# --------------------------------------------------------------------------- # +# Bridge — conversation mirroring (opt-in, FAST extraction, best-effort) +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_memorize_conversation_requires_opt_in() -> None: + """An available bridge without ``index_conversations`` never writes.""" + bridge = await _enabled_bridge() # index_conversations defaults to False + bridge._instance.write_async = AsyncMock() + assert bridge.indexes_conversations is False + sent = await bridge.memorize_conversation( + "s-1", [{"role": "user", "content": "hello"}], + ) + assert sent == 0 + bridge._instance.write_async.assert_not_called() + await bridge.aclose() + + +@pytest.mark.asyncio +async def test_memorize_conversation_noop_when_bridge_unavailable() -> None: + """Opt-in without credentials stays inert (no SDK calls, no errors).""" + bridge = XmemoryBridge(XmemoryConfig(index_conversations=True)) # no keys + await bridge.initialize() + assert bridge.indexes_conversations is False + assert await bridge.memorize_conversation( + "s-1", [{"role": "user", "content": "hello"}], + ) == 0 + + +@pytest.mark.asyncio +async def test_memorize_conversation_sends_text_only_with_fast_extraction() -> None: + """Transcripts always use FAST extraction — even when the configured + ``extraction_logic`` (which governs the memorize tool) is ``deep``.""" + bridge = await _enabled_bridge(extraction_logic="deep", index_conversations=True) + bridge._instance.write_async = AsyncMock(return_value=SimpleNamespace(write_id="w1")) + msgs = [ + { + "role": "user", + "content": "hello", + "created_at": "2026-01-01 00:00:00", + "thinking": "PRIVATE-REASONING", + "blocks": [{"type": "tool_result", "text": "RAW-TOOL-DUMP"}], + }, + {"role": "assistant", "content": "hi!"}, + ] + assert await bridge.memorize_conversation("s-1", msgs) == 1 + args, kwargs = bridge._instance.write_async.call_args + payload = args[0] + assert payload.startswith("Conversation transcript (session s-1)") + assert "[2026-01-01 00:00:00] user: hello" in payload + assert "assistant: hi!" in payload + assert "PRIVATE-REASONING" not in payload + assert "RAW-TOOL-DUMP" not in payload + assert kwargs["extraction_logic"] == bridge._ExtractionLogic.FAST + await bridge.aclose() + + +@pytest.mark.asyncio +async def test_memorize_conversation_chunks_large_transcripts(monkeypatch) -> None: + monkeypatch.setattr("nerve.memory.xmemory_bridge._TRANSCRIPT_CHUNK_BYTES", 64) + bridge = await _enabled_bridge(index_conversations=True) + bridge._instance.write_async = AsyncMock(return_value=SimpleNamespace(write_id="w")) + msgs = [ + {"role": "user", "content": f"message number {i} padded " + "x" * 20} + for i in range(6) + ] + sent = await bridge.memorize_conversation("s-1", msgs) + assert sent > 1 + assert sent == bridge._instance.write_async.await_count + payloads = [c.args[0] for c in bridge._instance.write_async.call_args_list] + for payload in payloads: + assert payload.splitlines()[0].startswith( + "Conversation transcript (session s-1, part ", + ) + combined = "\n".join(payloads) + for i in range(6): + assert f"message number {i} " in combined + await bridge.aclose() + + +@pytest.mark.asyncio +async def test_memorize_conversation_stops_on_first_failure(monkeypatch) -> None: + """Best-effort: a failed chunk abandons the rest instead of hammering a + down service — and never raises into the caller.""" + monkeypatch.setattr("nerve.memory.xmemory_bridge._TRANSCRIPT_CHUNK_BYTES", 64) + bridge = await _enabled_bridge(index_conversations=True) + bridge._instance.write_async = AsyncMock( + side_effect=[ + SimpleNamespace(write_id="w1"), + RuntimeError("quota exceeded"), + SimpleNamespace(write_id="w3"), + ], + ) + msgs = [ + {"role": "user", "content": f"chunk filler {i} " + "y" * 40} + for i in range(8) + ] + sent = await bridge.memorize_conversation("s-1", msgs) + assert sent == 1 # first chunk enqueued… + assert bridge._instance.write_async.await_count == 2 # …second failed, rest abandoned + await bridge.aclose() + + +@pytest.mark.asyncio +async def test_memorize_conversation_empty_transcript_is_noop() -> None: + bridge = await _enabled_bridge(index_conversations=True) + bridge._instance.write_async = AsyncMock() + assert await bridge.memorize_conversation("s-1", []) == 0 + assert await bridge.memorize_conversation( + "s-1", [{"role": "assistant", "content": ""}], + ) == 0 + bridge._instance.write_async.assert_not_called() + await bridge.aclose() + + +# --------------------------------------------------------------------------- # +# Engine — the sweep mirrors its memU window to xmemory +# --------------------------------------------------------------------------- # + + +def _bare_engine(xmem, memu=None, db=None) -> AgentEngine: + """AgentEngine with only the attributes the memorize paths touch.""" + engine = AgentEngine.__new__(AgentEngine) + engine._xmemory_bridge = xmem + engine._memory_bridge = memu + engine._memorize_bg_tasks = set() + engine.db = db + return engine + + +async def _drain_bg_tasks(engine: AgentEngine) -> None: + while engine._memorize_bg_tasks: + await asyncio.gather( + *list(engine._memorize_bg_tasks), return_exceptions=True, + ) + await asyncio.sleep(0) # let done-callbacks run and discard + + +def _xmem_mirror(opted_in: bool = True) -> MagicMock: + xmem = MagicMock() + xmem.indexes_conversations = opted_in + xmem.memorize_conversation = AsyncMock(return_value=1) + return xmem + + +def test_schedule_xmemory_transcript_inert_without_bridge() -> None: + engine = _bare_engine(xmem=None) + engine.schedule_xmemory_transcript("s-1", [{"role": "user", "content": "x"}]) + assert engine._memorize_bg_tasks == set() # nothing scheduled, no crash + + +@pytest.mark.asyncio +async def test_schedule_xmemory_transcript_inert_without_opt_in() -> None: + xmem = _xmem_mirror(opted_in=False) + engine = _bare_engine(xmem=xmem) + engine.schedule_xmemory_transcript("s-1", [{"role": "user", "content": "x"}]) + assert engine._memorize_bg_tasks == set() + xmem.memorize_conversation.assert_not_called() + + +@pytest.mark.asyncio +async def test_schedule_xmemory_transcript_fires_bridge_write() -> None: + xmem = _xmem_mirror() + engine = _bare_engine(xmem=xmem) + msgs = [{"role": "user", "content": "hello", "created_at": "2026-01-01 00:00:00"}] + engine.schedule_xmemory_transcript("s-1", msgs) + assert len(engine._memorize_bg_tasks) == 1 + await _drain_bg_tasks(engine) + xmem.memorize_conversation.assert_awaited_once_with("s-1", msgs) + assert engine._memorize_bg_tasks == set() # done-callback cleaned up + + +@pytest.mark.asyncio +async def test_schedule_xmemory_transcript_isolates_task_failure() -> None: + """A crashing mirror task is logged by the done-callback, never raised.""" + xmem = _xmem_mirror() + xmem.memorize_conversation = AsyncMock(side_effect=RuntimeError("boom")) + engine = _bare_engine(xmem=xmem) + engine.schedule_xmemory_transcript("s-1", [{"role": "user", "content": "x"}]) + await _drain_bg_tasks(engine) # must not raise + assert engine._memorize_bg_tasks == set() + + +@pytest.mark.asyncio +async def test_incremental_sweep_mirrors_new_messages_to_xmemory() -> None: + """The periodic sweep sends the exact memU window to xmemory too.""" + old = {"role": "user", "content": "old", "created_at": "2026-01-01 00:00:00"} + new_user = {"role": "user", "content": "newer", "created_at": "2026-01-02 10:00:00"} + new_asst = {"role": "assistant", "content": "reply", "created_at": "2026-01-02 10:00:05"} + + db = MagicMock() + db.get_session = AsyncMock( + return_value={"id": "s-1", "last_memorized_at": "2026-01-01 12:00:00"}, + ) + db.get_messages = AsyncMock(return_value=[old, new_user, new_asst]) + db.update_session_fields = AsyncMock() + + memu = MagicMock() + memu.available = True + memu.memorize_conversation = AsyncMock(return_value=True) + + xmem = _xmem_mirror() + engine = _bare_engine(xmem=xmem, memu=memu, db=db) + + count = await engine._memorize_incremental("s-1") + assert count == 2 # only the post-watermark window + + await _drain_bg_tasks(engine) + memu.memorize_conversation.assert_awaited_once_with("s-1", [new_user, new_asst]) + xmem.memorize_conversation.assert_awaited_once_with("s-1", [new_user, new_asst]) + db.update_session_fields.assert_awaited_once_with( + "s-1", {"last_memorized_at": "2026-01-02 10:00:05"}, + ) + + +@pytest.mark.asyncio +async def test_incremental_sweep_memu_only_when_not_opted_in() -> None: + db = MagicMock() + db.get_session = AsyncMock(return_value={"id": "s-1", "last_memorized_at": None}) + db.get_messages = AsyncMock( + return_value=[{"role": "user", "content": "hi", "created_at": "2026-01-02 10:00:00"}], + ) + db.update_session_fields = AsyncMock() + + memu = MagicMock() + memu.available = True + memu.memorize_conversation = AsyncMock(return_value=True) + + xmem = _xmem_mirror(opted_in=False) + engine = _bare_engine(xmem=xmem, memu=memu, db=db) + + assert await engine._memorize_incremental("s-1") == 1 + await _drain_bg_tasks(engine) + memu.memorize_conversation.assert_awaited_once() + xmem.memorize_conversation.assert_not_called()