xmemory: opt-in mirroring of memorization-sweep transcripts (text-only)

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
 

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`).
 

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(

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}")

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),
         )
 
 

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())

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