xiaoxue_admin
/
mempalace
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(3.7.0): agent logstream — scope the slice to coordination only 

Narrows the 3.7.0 slice to the feature users actually get: RFC 003 agent
coordination (events, artifacts, patch handoffs) plus the multi-master
logstream sync that makes it work across machines.

Removed, deferred with the rest of RFC 004:
- replica_sync.py / vector_cache.py and the `mempalace replica` CLI
- the /snapshot/* hub endpoints they backed
- website/concepts/replicated-palace.md

Memory read replicas were a leaf on the dependency graph (nothing in the
logstream path imports them), and shipping them half-done meant documenting
a mesh whose memory does not actually converge. Dropping them lets the docs
say one true thing instead of two hedged ones: coordination syncs, memory
stays local, point every agent at one hub if you want shared recall.

Fixes a bug found by running it: _start_peer_sync_thread() read peers.json
once at startup and returned early when absent, so a hub started before
peers.json was written never synced — silently, forever. That is the order
the guide tells users to follow. Membership is now re-read every round.

Verified on two live hubs: delegation loop end-to-end, verbatim patch
round-trip by sha256, bidirectional sync, CLI sync alongside a live hub,
and automatic convergence 15s after writing peers.json with no restart.
This commit is contained in:
Igor Lins e Silva 2026-08-06 00:25:49 -03:00
parent f78fe31c92
commit 123d01a16e
11 changed files with 106 additions and 1433 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

134
mempalace/cli.py
View File

@ -1445,88 +1445,6 @@ def cmd_artifact(args):
ls.close()
def cmd_replica(args):
"""RFC 004 step 1: read-replica operations for memory (drawers + KG)."""
import json
as_json = getattr(args, "json", False)
palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path
if args.replica_action == "embed-cache":
from .embedding import get_embedder_identity
from .vector_cache import build_cache
model = args.model or get_embedder_identity().model_name
def progress(done, scanned):
if not as_json and done % 5120 < args.batch:
print(f" embedded {done} (scanned {scanned})", flush=True)
try:
stats = build_cache(
palace_path,
model,
batch_size=args.batch,
authored_only=not args.all,
progress=progress,
)
except Exception as exc:
_logstream_fail(str(exc), as_json)
if as_json:
print(json.dumps(stats, indent=2, ensure_ascii=False))
else:
print(
f" model={stats['model']}: embedded {stats['embedded']} "
f"(cached {stats['skipped_cached']}, total in cache {stats['cache_total']})"
)
return
if args.replica_action == "pull":
from .replica_sync import pull_from_peers, pull_memory
try:
if args.peer:
results = [
pull_memory(
palace_path,
args.peer,
args.token or "",
reconcile_deletes=not args.no_reconcile,
pull_kg=not getattr(args, "no_kg", False),
with_vectors=getattr(args, "with_vectors", False),
)
]
else:
results = pull_from_peers(
palace_path,
pull_kg=not getattr(args, "no_kg", False),
with_vectors=getattr(args, "with_vectors", False),
)
if not results:
_logstream_fail(
f"no peers configured ({palace_path}/peers.json) and no --peer given",
as_json,
)
except Exception as exc:
_logstream_fail(str(exc), as_json)
if as_json:
print(json.dumps(results, indent=2, ensure_ascii=False))
else:
for stats in results:
if stats.get("error"):
print(
f" {stats.get('peer_name', stats['origin_url'])}: ERROR {stats['error']}"
)
else:
print(
f" {stats.get('peer_name', stats['origin_url'])} "
f"({stats['origin_replica']}): {stats['drawers_upserted']} drawers folded, "
f"{stats['drawers_deleted']} reconciled away, "
f"KG +{stats['kg_entities']} entities / +{stats['kg_triples']} triples"
)
if any(s.get("error") for s in results):
sys.exit(1)
def cmd_palace_set_embedder(args):
"""Record (or force-override) a palace's embedder identity (RFC 001).
@ -2833,51 +2751,6 @@ def main():
"--json", action="store_true", help="Metadata as JSON (content omitted with --out)"
)
# replica (RFC 004 read replicas)
p_replica = sub.add_parser(
"replica", help="Memory read-replica operations — pull facts from an origin palace"
)
replica_sub = p_replica.add_subparsers(dest="replica_action")
p_rep_pull = replica_sub.add_parser(
"pull", help="Pull drawers + KG from the origin; derive the vector index locally"
)
p_rep_pull.add_argument(
"--peer", default=None, help="Origin base URL (default: all peers in peers.json)"
)
p_rep_pull.add_argument("--token", default=None, help="Bearer token for --peer")
p_rep_pull.add_argument(
"--no-reconcile",
action="store_true",
help="Skip deleting local copies whose upstream original is gone",
)
p_rep_pull.add_argument(
"--no-kg",
action="store_true",
help="Skip knowledge-graph rows (use when the peer replicates your own KG back)",
)
p_rep_pull.add_argument(
"--with-vectors",
action="store_true",
help="Use vectors precomputed by the origin under this palace's embedder "
"identity (insert-only fold; origin must have run embed-cache)",
)
p_rep_pull.add_argument("--json", action="store_true", help="Machine-readable output")
p_rep_embed = replica_sub.add_parser(
"embed-cache",
help="Bulk-embed local documents into the portable vector cache "
"(distributed derivation: compute here, fold anywhere)",
)
p_rep_embed.add_argument(
"--model", default=None, help="Embedder identity (default: this palace's identity)"
)
p_rep_embed.add_argument("--batch", type=int, default=256, help="Embedding batch size")
p_rep_embed.add_argument(
"--all",
action="store_true",
help="Embed everything (default: authored-only — the set peers pull from us)",
)
p_rep_embed.add_argument("--json", action="store_true", help="Machine-readable output")
p_palace = sub.add_parser("palace", help="Palace maintenance commands")
palace_sub = p_palace.add_subparsers(dest="palace_action")
p_set_embedder = palace_sub.add_parser(
@ -2948,13 +2821,6 @@ def main():
cmd_artifact(args)
return
if args.command == "replica":
if not getattr(args, "replica_action", None):
p_replica.print_help()
return
cmd_replica(args)
return
if args.command == "daemon":
if not getattr(args, "daemon_action", None):
p_daemon.print_help()

177
mempalace/mcp_server.py
View File

@ -6200,165 +6200,11 @@ def _http_handle_get(handler) -> None:
if not _http_serve_sync(handler, path):
handler.send_error(404, "Not Found")
return
if path.startswith("/snapshot/"):
if handler._request_rejected(require_auth=True):
return
if not _http_serve_snapshot(handler, path):
handler.send_error(404, "Not Found")
return
if handler._request_rejected(require_auth=False):
return
handler.send_error(404, "Not Found")
def _http_serve_snapshot(handler, path: str) -> bool:
"""RFC 004 step 1: content-snapshot endpoints for memory read replicas.
GET /snapshot/manifest → {replica_id, drawers, kg, collection}
GET /snapshot/drawers?offset=&limit= → {items: [{id, document, metadata}]}
GET /snapshot/ids?offset=&limit= → {ids: [...]} (delete reconciliation)
GET /snapshot/kg?table=&after=&limit= → {rows: [...]} (rowid-paged)
Facts only — verbatim documents, metadata, and KG rows. Vector indexes
are never served: the replica derives its own (RFC 004 layer 3). Chroma
access happens under the global request lock like every other
Chroma-touching request; pages are small enough to keep holds short.
"""
from urllib.parse import parse_qsl
query = dict(parse_qsl(urlparse(handler.path).query))
try:
limit = max(1, min(int(query.get("limit", "500") or 500), 1000))
offset = max(0, int(query.get("offset", "0") or 0))
after = max(0, int(query.get("after", "0") or 0))
except (TypeError, ValueError):
handler._send_json(400, {"error": "limit/offset/after must be integers"})
return True
if path == "/snapshot/manifest":
from .vector_cache import VECTOR_CACHE_FILENAME, VectorCache
with _HTTP_REQUEST_LOCK:
col = _get_collection()
drawer_count = col.count() if col else 0
kg_stats = _call_kg(lambda kg: kg.stats())
vector_models = {}
if _config.palace_path and os.path.exists(
os.path.join(os.path.expanduser(_config.palace_path), VECTOR_CACHE_FILENAME)
):
cache = VectorCache(_config.palace_path)
try:
with cache._lock:
rows = (
cache._conn()
.execute("SELECT embedder, count(*) AS n FROM vectors GROUP BY embedder")
.fetchall()
)
vector_models = {row["embedder"]: row["n"] for row in rows}
finally:
cache.close()
handler._send_json(
200,
{
"replica_id": _call_logstream(lambda ls: ls.replica_id),
"drawers": drawer_count,
"kg": {
"entities": kg_stats.get("entities", 0),
"triples": kg_stats.get("triples", 0),
},
"collection": _config.collection_name,
"vector_cache": vector_models,
},
)
return True
if path == "/snapshot/drawers":
from .replica_sync import REPLICA_ORIGIN_KEY
with _HTTP_REQUEST_LOCK:
col = _get_collection()
if col is None:
handler._send_json(503, {"error": "no palace collection available"})
return True
batch = col.get(limit=limit, offset=offset, include=["documents", "metadatas"])
ids = batch.get("ids") or []
docs = batch.get("documents") or []
metas = batch.get("metadatas") or []
# Authored-only: a snapshot serves the facts THIS replica authors.
# Drawers folded from another origin (replica_origin-stamped) are
# excluded, or bidirectional pull-pairs would echo each other's
# content back re-stamped, corrupting provenance and reconciliation.
items = [
{"id": i, "document": d, "metadata": m or {}}
for i, d, m in zip(ids, docs, metas)
if not (m or {}).get(REPLICA_ORIGIN_KEY)
]
# Distributed derivation (RFC 004): ship cached vectors alongside
# content when the caller names an embedder identity, so the fold
# side can insert without re-deriving. Missing vectors are simply
# omitted; the fold falls back to local embedding per item.
vectors_model = query.get("vectors") or ""
if vectors_model and items:
from .vector_cache import VectorCache, vector_to_b64
cache = VectorCache(_config.palace_path)
try:
found = cache.get_many(vectors_model, [item["id"] for item in items])
finally:
cache.close()
for item in items:
vector = found.get(item["id"])
if vector is not None:
item["vector_b64"] = vector_to_b64(vector)
items_with = sum(1 for item in items if "vector_b64" in item)
else:
items_with = 0
handler._send_json(
200,
{
"items": items,
"count": len(items),
"vectors_included": items_with,
"vectors_model": vectors_model or None,
"offset": offset,
# Raw advance: pages may return fewer items than scanned.
"next_offset": offset + len(ids) if ids else None,
},
)
return True
if path == "/snapshot/ids":
from .replica_sync import REPLICA_ORIGIN_KEY
with _HTTP_REQUEST_LOCK:
col = _get_collection()
if col is None:
handler._send_json(503, {"error": "no palace collection available"})
return True
batch = col.get(limit=limit, offset=offset, include=["metadatas"])
raw_ids = batch.get("ids") or []
metas = batch.get("metadatas") or []
ids = [i for i, m in zip(raw_ids, metas) if not (m or {}).get(REPLICA_ORIGIN_KEY)]
handler._send_json(
200,
{
"ids": ids,
"count": len(ids),
"offset": offset,
"next_offset": offset + len(raw_ids) if raw_ids else None,
},
)
return True
if path == "/snapshot/kg":
table = query.get("table") or ""
try:
rows = _call_kg(lambda kg: kg.dump_rows(table, after_rowid=after, limit=limit))
except ValueError as exc:
handler._send_json(400, {"error": str(exc)})
return True
handler._send_json(200, {"table": table, "rows": rows, "count": len(rows)})
return True
return False
def _http_serve_sync(handler, path: str) -> bool:
"""RFC 004 step 0: pull-based anti-entropy endpoints for peer replicas.
@ -6595,14 +6441,21 @@ def _peer_sync_interval_s() -> float:
def _start_peer_sync_thread() -> None:
"""Background anti-entropy loop when peers.json exists (RFC 004 step 0).
"""Background anti-entropy loop for the logstream (RFC 004 step 0).
Runs in the serving process so a hub with configured peers converges
with zero extra processes. Interval via MEMPALACE_SYNC_INTERVAL
(seconds, default 15; 0 disables). Errors are logged, never fatal —
a dead peer must not take the loop down (R1).
Membership is re-read from peers.json every round, never latched at
startup: joining the mesh is "write peers.json", and requiring a hub
restart for a file the docs describe as picked up "within one sync
cycle" is a silent no-op for anyone following the guide in order (hub
first, peers after). A malformed peers.json logs once per transition
rather than every round, so a typo is visible without flooding the log.
"""
from .logsync import load_peers, sync_all
from .logsync import sync_all
palace_path = getattr(_config, "palace_path", None)
if not palace_path:
@ -6610,14 +6463,9 @@ def _start_peer_sync_thread() -> None:
interval = _peer_sync_interval_s()
if interval <= 0:
return
try:
if not load_peers(palace_path):
return
except ValueError as exc:
logger.error("peers.json is malformed; peer sync disabled: %s", exc)
return
def _loop():
malformed_logged = False
while True:
time.sleep(interval)
try:
@ -6633,6 +6481,11 @@ def _start_peer_sync_thread() -> None:
stats["pulled_events"],
stats["pulled_artifacts"],
)
malformed_logged = False
except ValueError as exc:
if not malformed_logged:
logger.error("peers.json is malformed; skipping peer sync: %s", exc)
malformed_logged = True
except Exception:
logger.warning("peer sync round failed", exc_info=True)

240
mempalace/replica_sync.py
View File

@ -1,240 +0,0 @@
"""
replica_sync.py — Memory read-replica pull engine (RFC 004 step 1)
Pulls the origin palace's *facts* — verbatim drawer documents + metadata and
knowledge-graph rows — over the /snapshot/* endpoints and folds them into
this machine's local palace. The local vector index is derived here, by this
machine's own embedder, exactly per RFC 004 layer 3: sync the facts, derive
the senses. `chroma.sqlite3` is never copied.
Semantics:
- Drawer folds are upserts by drawer id; documents and metadata are stored
verbatim, plus ONE additive provenance key: ``replica_origin`` = the
origin's replica id. That key is what makes delete reconciliation safe —
only drawers this replica *copied from that origin* are ever deleted when
they disappear upstream; locally-authored drawers (a local agent's diary)
are untouchable by reconciliation.
- KG folds are INSERT OR REPLACE by row id: invalidations (valid_to edits)
converge on re-pull; replication never deletes KG rows.
- Every pass is idempotent. Freshness = re-run the pull; the step-2 op-log
replaces polling with precise tails later.
Writes on a step-1 replica remain the origin's job — this engine makes
recall local and durable, not capture. Multi-writer memory is RFC 004
step 3.
"""
import logging
from .logsync import SyncPeerError, _peer_get, load_peers
logger = logging.getLogger("mempalace.replica_sync")
_PAGE = 500
REPLICA_ORIGIN_KEY = "replica_origin"
def _pull_drawers(col, url: str, token: str, origin_id: str, vectors_model: str = None) -> dict:
"""Page and fold remote drawers. With ``vectors_model``, request served
vectors and upsert them explicitly (insert-only fold, RFC 004
distributed derivation); items the peer has no cached vector for fall
back to local embedding. Returns fold stats."""
from .vector_cache import b64_to_vector
upserted = 0
vectors_used = 0
locally_embedded = 0
offset = 0
while True:
params = {"offset": offset, "limit": _PAGE}
if vectors_model:
params["vectors"] = vectors_model
page = _peer_get(url, token, "/snapshot/drawers", params)
items = page.get("items") or []
if items:
with_vec = [item for item in items if item.get("vector_b64")]
without_vec = [item for item in items if not item.get("vector_b64")]
for group, explicit in ((with_vec, True), (without_vec, False)):
if not group:
continue
ids = [item["id"] for item in group]
documents = [item["document"] for item in group]
metadatas = [
{**(item.get("metadata") or {}), REPLICA_ORIGIN_KEY: origin_id}
for item in group
]
if explicit:
embeddings = [b64_to_vector(item["vector_b64"]) for item in group]
col.upsert(
ids=ids, documents=documents, metadatas=metadatas, embeddings=embeddings
)
vectors_used += len(group)
else:
col.upsert(ids=ids, documents=documents, metadatas=metadatas)
locally_embedded += len(group)
upserted += len(items)
# Authored-only serving can return sparse pages: advance by the
# server's raw scan cursor, not by the filtered item count.
next_offset = page.get("next_offset")
if next_offset is None:
break
offset = next_offset
return {
"upserted": upserted,
"vectors_used": vectors_used,
"locally_embedded": locally_embedded,
}
def _pull_remote_ids(url: str, token: str) -> set:
remote_ids = set()
offset = 0
while True:
page = _peer_get(url, token, "/snapshot/ids", {"offset": offset, "limit": 1000})
remote_ids.update(page.get("ids") or [])
next_offset = page.get("next_offset")
if next_offset is None:
break
offset = next_offset
return remote_ids
def _reconcile_deletes(col, remote_ids: set, origin_id: str) -> int:
"""Delete local copies whose upstream original is gone.
Scoped strictly to drawers stamped ``replica_origin == origin_id`` —
reconciliation can never touch locally-authored content.
"""
to_delete = []
offset = 0
while True:
batch = col.get(limit=1000, offset=offset, include=["metadatas"])
ids = batch.get("ids") or []
if not ids:
break
metas = batch.get("metadatas") or []
for drawer_id, meta in zip(ids, metas):
if (meta or {}).get(REPLICA_ORIGIN_KEY) == origin_id and drawer_id not in remote_ids:
to_delete.append(drawer_id)
offset += len(ids)
for start in range(0, len(to_delete), 500):
col.delete(ids=to_delete[start : start + 500])
return len(to_delete)
def _pull_kg(kg, url: str, token: str) -> dict:
counts = {}
for table in ("entities", "triples"):
applied = 0
after = 0
while True:
page = _peer_get(
url, token, "/snapshot/kg", {"table": table, "after": after, "limit": _PAGE}
)
rows = page.get("rows") or []
if not rows:
break
for row in rows:
kg.apply_row(table, row)
after = max(after, row.get("_rowid") or 0)
applied += 1
counts[table] = applied
return counts
def pull_memory(
palace_path: str,
url: str,
token: str = "",
reconcile_deletes: bool = True,
pull_kg: bool = True,
with_vectors: bool = False,
) -> dict:
"""One full read-replica pass against one origin. Returns fold stats.
Opens the LOCAL palace's collection (created on first pull) and KG;
embedding of pulled documents happens here, on this machine's embedder.
"""
from .knowledge_graph import KnowledgeGraph
from .palace import get_collection
manifest = _peer_get(url, token, "/snapshot/manifest")
origin_id = manifest.get("replica_id") or "unknown-origin"
vectors_model = None
if with_vectors:
# Served vectors are only usable under the LOCAL embedder identity —
# a vector from a different model poisons the index silently, so the
# identity gate is loud and mandatory (RFC 001 meets RFC 004).
from .embedding import get_embedder_identity
identity = get_embedder_identity()
vectors_model = identity.model_name
peer_cache = (manifest.get("vector_cache") or {}).get(vectors_model, 0)
if not peer_cache:
raise SyncPeerError(
f"peer has no vector cache for local embedder identity "
f"{vectors_model!r} — run 'mempalace replica embed-cache' on the "
"origin first, or pull without --with-vectors"
)
col = get_collection(palace_path, create=True)
if col is None:
raise SyncPeerError(f"could not open local collection in {palace_path!r}")
fold = _pull_drawers(col, url, token, origin_id, vectors_model=vectors_model)
upserted = fold["upserted"]
deleted = 0
if reconcile_deletes:
deleted = _reconcile_deletes(col, _pull_remote_ids(url, token), origin_id)
import os
# KG rows carry no origin stamp, so a reverse pull could overwrite a
# newer local row with the peer's stale replicated copy (INSERT OR
# REPLACE has no freshness check). Until step 2a gives KG facts real
# op provenance, disable KG on pulls from peers that merely replicate
# your KG back (pull_kg=False / CLI --no-kg).
kg_counts = {"entities": 0, "triples": 0}
if pull_kg:
kg = KnowledgeGraph(
db_path=os.path.join(os.path.expanduser(palace_path), "knowledge_graph.sqlite3")
)
try:
kg_counts = _pull_kg(kg, url, token)
finally:
kg.close()
return {
"origin_url": url,
"origin_replica": origin_id,
"manifest": manifest,
"drawers_upserted": upserted,
"drawers_deleted": deleted,
"vectors_used": fold["vectors_used"],
"locally_embedded": fold["locally_embedded"],
"kg_entities": kg_counts["entities"],
"kg_triples": kg_counts["triples"],
}
def pull_from_peers(palace_path: str, pull_kg: bool = True, with_vectors: bool = False) -> list:
"""Run a pull against every peers.json entry; per-peer errors reported,
never raised (a dead origin must not block the others)."""
results = []
for peer in load_peers(palace_path):
name = peer.get("name") or peer["url"]
try:
stats = pull_memory(
palace_path,
peer["url"],
peer.get("token", ""),
pull_kg=pull_kg,
with_vectors=with_vectors,
)
stats["peer_name"] = name
results.append(stats)
except (SyncPeerError, ValueError) as exc:
results.append({"peer_name": name, "origin_url": peer["url"], "error": str(exc)})
return results

192
mempalace/vector_cache.py
View File

@ -1,192 +0,0 @@
"""
vector_cache.py — Portable embedding cache (RFC 004 distributed derivation)
A vector is a pure function of (content, embedder identity). When a fleet
pins one identity, vectors become portable facts: computed on whichever
machine has the horsepower, shipped with content, folded insert-only on the
receiver. This sidecar cache (``vector_cache.sqlite3`` in the palace dir)
holds vectors keyed ``(drawer_id, embedder)`` as packed float32 blobs.
The cache is strictly derived state about content the palace already holds:
building it READS the content store and writes only the sidecar — safe to
run against a quiescent origin without breaking quiescence guarantees.
Wire format: vectors travel base64(float32 little-endian). ~1.5 KB raw per
384-dim vector; a 156k-drawer fleet merge is a few hundred MB — LAN/tailnet
territory, never WAN-cloud.
"""
import base64
import os
import sqlite3
import struct
import threading
from pathlib import Path
VECTOR_CACHE_FILENAME = "vector_cache.sqlite3"
def pack_vector(vector: list) -> bytes:
return struct.pack(f"<{len(vector)}f", *vector)
def unpack_vector(blob: bytes) -> list:
return list(struct.unpack(f"<{len(blob) // 4}f", blob))
def vector_to_b64(vector: list) -> str:
return base64.b64encode(pack_vector(vector)).decode("ascii")
def b64_to_vector(encoded: str) -> list:
return unpack_vector(base64.b64decode(encoded))
class VectorCache:
"""SQLite-backed vector store, WAL, thread-safe like the other sidecars."""
def __init__(self, palace_path: str):
self.db_path = str(Path(os.path.expanduser(palace_path)) / VECTOR_CACHE_FILENAME)
Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
self._connection = None
self._lock = threading.Lock()
conn = self._conn()
conn.executescript("""
PRAGMA journal_mode=WAL;
CREATE TABLE IF NOT EXISTS vectors (
drawer_id TEXT NOT NULL,
embedder TEXT NOT NULL,
dim INTEGER NOT NULL,
vec BLOB NOT NULL,
PRIMARY KEY (drawer_id, embedder)
);
""")
conn.commit()
def _conn(self):
if self._connection is None:
self._connection = sqlite3.connect(self.db_path, timeout=10, check_same_thread=False)
self._connection.execute("PRAGMA journal_mode=WAL")
self._connection.row_factory = sqlite3.Row
return self._connection
def close(self):
with self._lock:
if self._connection is not None:
self._connection.close()
self._connection = None
def put_many(self, embedder: str, items: list) -> int:
"""items: [(drawer_id, vector: list[float])]. Idempotent by key."""
with self._lock:
conn = self._conn()
with conn:
conn.executemany(
"INSERT OR REPLACE INTO vectors (drawer_id, embedder, dim, vec)"
" VALUES (?, ?, ?, ?)",
[(d, embedder, len(v), pack_vector(v)) for d, v in items],
)
return len(items)
def get_many(self, embedder: str, drawer_ids: list) -> dict:
"""{drawer_id: vector} for the ids present under this embedder."""
if not drawer_ids:
return {}
out = {}
with self._lock:
conn = self._conn()
for start in range(0, len(drawer_ids), 500):
chunk = drawer_ids[start : start + 500]
placeholders = ",".join("?" for _ in chunk)
rows = conn.execute(
f"SELECT drawer_id, vec FROM vectors WHERE embedder = ? "
f"AND drawer_id IN ({placeholders})",
[embedder, *chunk],
).fetchall()
for row in rows:
out[row["drawer_id"]] = unpack_vector(row["vec"])
return out
def missing_ids(self, embedder: str, drawer_ids: list) -> list:
cached = self.get_many(embedder, drawer_ids)
return [d for d in drawer_ids if d not in cached]
def count(self, embedder: str) -> int:
with self._lock:
conn = self._conn()
row = conn.execute(
"SELECT count(*) AS n FROM vectors WHERE embedder = ?", (embedder,)
).fetchone()
return row["n"]
def build_cache(
palace_path: str,
model: str,
batch_size: int = 256,
authored_only: bool = True,
progress=None,
) -> dict:
"""Bulk-embed the local palace's documents into the vector cache.
Reads the content store paged, embeds missing documents with the named
model in ``batch_size`` batches, writes the sidecar. Idempotent and
resumable: already-cached (drawer_id, embedder) pairs are skipped, so a
crash resumes where it left off. ``authored_only`` limits the job to
drawers this replica authors (no ``replica_origin`` stamp) — the set a
peer would pull from us.
"""
from .embedding import get_embedding_function
from .palace import get_collection
from .replica_sync import REPLICA_ORIGIN_KEY
# Resolve the embedding function once, for the requested identity.
ef = get_embedding_function(model=model)
def embed(texts):
vectors = ef(input=texts)
return [list(v) for v in vectors]
col = get_collection(os.path.expanduser(palace_path))
if col is None:
raise ValueError(f"no collection in {palace_path!r}")
cache = VectorCache(palace_path)
try:
scanned = 0
embedded = 0
skipped_cached = 0
offset = 0
while True:
batch = col.get(limit=1000, offset=offset, include=["documents", "metadatas"])
ids = batch.get("ids") or []
if not ids:
break
docs = batch.get("documents") or []
metas = batch.get("metadatas") or []
candidates = [
(i, d)
for i, d, m in zip(ids, docs, metas)
if not (authored_only and (m or {}).get(REPLICA_ORIGIN_KEY))
]
scanned += len(candidates)
todo_ids = cache.missing_ids(model, [i for i, _ in candidates])
skipped_cached += len(candidates) - len(todo_ids)
todo = {i: d for i, d in candidates if i in set(todo_ids)}
pending = list(todo.items())
for start in range(0, len(pending), batch_size):
chunk = pending[start : start + batch_size]
vectors = embed([d for _, d in chunk])
cache.put_many(model, [(i, v) for (i, _), v in zip(chunk, vectors)])
embedded += len(chunk)
if progress:
progress(embedded, scanned)
offset += len(ids)
return {
"model": model,
"scanned": scanned,
"embedded": embedded,
"skipped_cached": skipped_cached,
"cache_total": cache.count(model),
}
finally:
cache.close()

63
tests/test_logstream_sync.py
View File

@ -673,3 +673,66 @@ class TestNodeProfile:
from mempalace import mcp_server as mcp
assert "mempalace_mesh_peers" in mcp._SQLITE_INTEGRITY_ALLOWED_TOOLS
class TestPeerSyncThreadStartup:
"""The loop must not latch membership at startup.
Joining the mesh is "write peers.json", and the guide has users start
the hub before writing it. A thread that returns early when the file
is missing leaves that hub permanently non-syncing with no error —
the failure mode is silence, which is why it needs a test.
"""
def _start(self, tmp_path, monkeypatch, interval="0.05"):
import threading
from mempalace import mcp_server as mcp
monkeypatch.setenv("MEMPALACE_SYNC_INTERVAL", interval)
monkeypatch.setenv("MEMPALACE_PALACE_PATH", str(tmp_path))
# Compare thread objects, not names: earlier tests in this class
# leave their own daemon loop running under the same name.
before = set(threading.enumerate())
mcp._start_peer_sync_thread()
return [
t for t in threading.enumerate() if t.name == "mempalace-logsync" and t not in before
]
def test_thread_starts_without_peers_json(self, tmp_path, monkeypatch):
assert not (tmp_path / "peers.json").exists()
assert self._start(tmp_path, monkeypatch), (
"peer sync thread must start even with no peers.json — otherwise a "
"peers.json written after the hub boots is silently ignored forever"
)
def test_peers_written_after_startup_are_picked_up(self, tmp_path, monkeypatch):
import time as _time
from mempalace import mcp_server as mcp
calls = []
def _fake_sync_all(ls, palace_path, transport=None):
calls.append(palace_path)
return []
monkeypatch.setattr("mempalace.logsync.sync_all", _fake_sync_all)
monkeypatch.setattr(mcp, "_get_logstream", lambda: object())
assert self._start(tmp_path, monkeypatch)
# peers.json appears only now — after the thread is already running.
(tmp_path / "peers.json").write_text(
json.dumps({"peers": [{"name": "a", "url": "http://x", "token": "t"}]}),
encoding="utf-8",
)
deadline = _time.monotonic() + 5
while _time.monotonic() < deadline and not calls:
_time.sleep(0.05)
assert calls, "sync round never ran; membership was latched at startup"
def test_disabled_by_zero_interval(self, tmp_path, monkeypatch):
assert not self._start(tmp_path, monkeypatch, interval="0")

430
tests/test_replica_sync.py
View File

@ -1,430 +0,0 @@
"""
Tests for RFC 004 step 1 — memory read replicas (snapshot pull + local fold).
Origin palace served by the production HTTP server; replica palace folds
via replica_sync. Uses the sqlite_exact backend with a fake embedder (the
same pattern as test_sqlite_exact_backend.py) so no model download happens:
the point is fact replication, and vectors are derived locally by design.
"""
import http.client
import json
import math
import os
import threading
import pytest
from mempalace.knowledge_graph import KnowledgeGraph
from mempalace.replica_sync import REPLICA_ORIGIN_KEY, pull_memory
@pytest.fixture
def fake_embedder(monkeypatch):
import mempalace.backends.embedding_wrapper as embedding_wrapper
def fake_embed(texts):
return [[1.0, 0.0] if "canary" in t else [0.5, math.sqrt(0.75)] for t in texts]
monkeypatch.setenv("MEMPALACE_BACKEND_EXPLICIT", "sqlite_exact")
monkeypatch.setattr(embedding_wrapper, "_embed_texts", fake_embed)
@pytest.fixture
def origin_server(monkeypatch, fake_embedder, config, palace_path):
"""Origin palace with seeded drawers + KG, served over real HTTP."""
from mempalace import mcp_server as mcp
from mempalace.palace import get_collection
monkeypatch.setattr(mcp, "_config", config)
monkeypatch.setattr(mcp, "_logstream_by_path", {})
monkeypatch.setattr(mcp, "_collection_cache", None)
monkeypatch.setattr(mcp, "_client_cache", None)
col = get_collection(palace_path, create=True)
col.upsert(
ids=["drawer_w_r_aaa", "drawer_w_r_bbb", "drawer_w2_r2_ccc"],
documents=[
"the search canary drawer",
"alembic migrations run on postgres",
"sprint planning notes for q3",
],
metadatas=[
{"wing": "w", "room": "r", "source_file": "a.md", "filed_at": "2026-07-01T00:00:00"},
{"wing": "w", "room": "r", "source_file": "b.md", "filed_at": "2026-07-01T00:00:01"},
{"wing": "w2", "room": "r2", "added_by": "mcp", "filed_at": "2026-07-01T00:00:02"},
],
)
kg = KnowledgeGraph(db_path=os.path.join(palace_path, "knowledge_graph.sqlite3"))
kg.add_entity("Alice", entity_type="person")
kg.add_triple("Alice", "works_on", "MemPalace", valid_from="2026-01-01")
kg.close()
monkeypatch.setattr(
mcp,
"_get_kg",
lambda *a, **kw: KnowledgeGraph(
db_path=os.path.join(palace_path, "knowledge_graph.sqlite3")
),
)
httpd = mcp._build_http_server("127.0.0.1", 0)
port = httpd.server_address[1]
thread = threading.Thread(
target=httpd.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True
)
thread.start()
try:
yield f"http://127.0.0.1:{port}", palace_path, mcp
finally:
httpd.shutdown()
httpd.server_close()
thread.join(timeout=5)
for ls in mcp._logstream_by_path.values():
ls.close()
@pytest.fixture
def replica_palace(tmp_dir):
p = os.path.join(tmp_dir, "replica_palace")
os.makedirs(p)
return p
def _get(url, path):
host, port = url.replace("http://", "").split(":")
conn = http.client.HTTPConnection(host, int(port), timeout=5)
try:
conn.request("GET", path)
resp = conn.getresponse()
return resp.status, json.loads(resp.read() or b"{}")
finally:
conn.close()
class _ConfigProxy:
"""Config stand-in pointing the served palace at a different directory."""
def __init__(self, palace_path, base):
self.palace_path = palace_path
self.collection_name = base.collection_name
def __getattr__(self, name):
raise AttributeError(name)
class TestSnapshotEndpoints:
def test_manifest_counts(self, origin_server):
url, _, _ = origin_server
status, manifest = _get(url, "/snapshot/manifest")
assert status == 200
assert manifest["drawers"] == 3
assert manifest["kg"]["triples"] == 1
assert manifest["replica_id"].startswith("rep_")
def test_drawers_paginate(self, origin_server):
url, _, _ = origin_server
status, page1 = _get(url, "/snapshot/drawers?offset=0&limit=2")
status2, page2 = _get(url, "/snapshot/drawers?offset=2&limit=2")
assert status == status2 == 200
ids = [i["id"] for i in page1["items"]] + [i["id"] for i in page2["items"]]
assert sorted(ids) == ["drawer_w2_r2_ccc", "drawer_w_r_aaa", "drawer_w_r_bbb"]
assert all("document" in i and "metadata" in i for i in page1["items"])
def test_ids_endpoint(self, origin_server):
url, _, _ = origin_server
status, page = _get(url, "/snapshot/ids?offset=0&limit=10")
assert status == 200
assert len(page["ids"]) == 3
def test_kg_pages_and_rejects_bad_table(self, origin_server):
url, _, _ = origin_server
status, triples = _get(url, "/snapshot/kg?table=triples&after=0&limit=10")
assert status == 200
assert triples["rows"][0]["subject"] == "alice"
status, bad = _get(url, "/snapshot/kg?table=users&after=0")
assert status == 400
def test_bad_pagination_is_400(self, origin_server):
url, _, _ = origin_server
status, _ = _get(url, "/snapshot/drawers?offset=nope")
assert status == 400
class TestReplicaPull:
def test_full_pull_folds_facts_and_derives_locally(
self, origin_server, replica_palace, fake_embedder
):
url, _, _ = origin_server
stats = pull_memory(replica_palace, url)
assert stats["drawers_upserted"] == 3
assert stats["drawers_deleted"] == 0
assert stats["kg_triples"] == 1
assert stats["kg_entities"] >= 1
# Facts are verbatim and searchable locally.
from mempalace.searcher import search_memories
result = search_memories("search canary", replica_palace, n_results=1)
assert result["results"][0]["text"] == "the search canary drawer"
# Provenance stamp present on the copy.
from mempalace.palace import get_collection
col = get_collection(replica_palace)
copy = col.get(ids=["drawer_w_r_aaa"], include=["metadatas"])
assert copy["metadatas"][0][REPLICA_ORIGIN_KEY].startswith("rep_")
# KG fact replicated with temporal fields intact.
kg = KnowledgeGraph(db_path=os.path.join(replica_palace, "knowledge_graph.sqlite3"))
try:
facts = kg.query_entity("Alice")
assert any(f["object"] == "MemPalace" for f in facts)
finally:
kg.close()
def test_pull_is_idempotent(self, origin_server, replica_palace, fake_embedder):
url, _, _ = origin_server
pull_memory(replica_palace, url)
stats = pull_memory(replica_palace, url)
assert stats["drawers_upserted"] == 3 # upserts, not duplicates
from mempalace.palace import get_collection
assert get_collection(replica_palace).count() == 3
def test_reconcile_deletes_only_origin_copies(
self, origin_server, replica_palace, fake_embedder
):
url, origin_palace, _ = origin_server
pull_memory(replica_palace, url)
# A locally-authored drawer must survive reconciliation forever.
from mempalace.palace import get_collection
replica_col = get_collection(replica_palace)
replica_col.upsert(
ids=["drawer_local_diary"],
documents=["local diary entry, authored on the replica"],
metadatas=[{"wing": "local", "room": "diary"}],
)
# Upstream deletes one drawer; the replica's copy reconciles away.
origin_col = get_collection(origin_palace)
origin_col.delete(ids=["drawer_w_r_bbb"])
stats = pull_memory(replica_palace, url)
assert stats["drawers_deleted"] == 1
remaining = set(replica_col.get(limit=100)["ids"])
assert "drawer_w_r_bbb" not in remaining
assert "drawer_local_diary" in remaining
def test_kg_invalidation_converges_on_repull(
self, origin_server, replica_palace, fake_embedder
):
url, origin_palace, _ = origin_server
pull_memory(replica_palace, url)
origin_kg = KnowledgeGraph(db_path=os.path.join(origin_palace, "knowledge_graph.sqlite3"))
origin_kg.invalidate("Alice", "works_on", "MemPalace", ended="2026-07-01")
origin_kg.close()
pull_memory(replica_palace, url)
kg = KnowledgeGraph(db_path=os.path.join(replica_palace, "knowledge_graph.sqlite3"))
try:
live_now = kg.query_entity("Alice", as_of="2026-07-02")
assert not any(f["object"] == "MemPalace" for f in live_now)
finally:
kg.close()
def test_bidirectional_two_origin_pull_has_no_echo(
self, origin_server, replica_palace, fake_embedder, monkeypatch
):
"""Option C from the fleet: each machine authors its own history and
pulls the other's. Snapshots serve authored-only content, so a pull
pair must never echo a replica's own drawers back re-stamped."""
url, origin_palace, mcp = origin_server
pull_memory(replica_palace, url) # replica now holds origin's 3 drawers
# Replica authors net-new local content (the Windows mining case).
from mempalace.palace import get_collection
replica_col = get_collection(replica_palace)
replica_col.upsert(
ids=["drawer_win_convo_001"],
documents=["windows conversation canary from the second origin"],
metadatas=[{"wing": "claude_conversations_windows", "room": "technical"}],
)
# Reverse direction: serve the REPLICA palace, pull into the origin.
monkeypatch.setattr(mcp, "_config", _ConfigProxy(replica_palace, mcp._config))
monkeypatch.setattr(mcp, "_collection_cache", None)
monkeypatch.setattr(mcp, "_client_cache", None)
monkeypatch.setattr(mcp, "_logstream_by_path", {})
import threading
httpd = mcp._build_http_server("127.0.0.1", 0)
port = httpd.server_address[1]
thread = threading.Thread(
target=httpd.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True
)
thread.start()
try:
stats = pull_memory(origin_palace, f"http://127.0.0.1:{port}", pull_kg=False)
# Only the replica-AUTHORED drawer crosses; the origin's own 3
# drawers are not echoed back.
assert stats["drawers_upserted"] == 1
origin_col = get_collection(origin_palace)
copy = origin_col.get(ids=["drawer_win_convo_001"], include=["documents", "metadatas"])
assert copy["documents"][0] == "windows conversation canary from the second origin"
assert copy["metadatas"][0][REPLICA_ORIGIN_KEY].startswith("rep_")
# Origin's own drawers keep clean provenance (no stamp).
own = origin_col.get(ids=["drawer_w_r_aaa"], include=["metadatas"])
assert REPLICA_ORIGIN_KEY not in (own["metadatas"][0] or {})
finally:
httpd.shutdown()
httpd.server_close()
thread.join(timeout=5)
def test_cli_replica_pull(self, origin_server, replica_palace, fake_embedder, capsys):
url, _, _ = origin_server
from types import SimpleNamespace
from mempalace.cli import cmd_replica
cmd_replica(
SimpleNamespace(
palace=replica_palace,
replica_action="pull",
peer=url,
token=None,
no_reconcile=False,
json=True,
)
)
results = json.loads(capsys.readouterr().out)
assert results[0]["drawers_upserted"] == 3
class TestDistributedVectors:
"""RFC 004 distributed derivation: vectors computed at the origin under
the puller's embedder identity, folded insert-only."""
@pytest.fixture
def fake_identity(self, monkeypatch):
import types
import mempalace.embedding as embedding
def fake_ef_factory(device=None, model=None):
def ef(input):
return [[1.0, 0.0] if "canary" in t else [0.0, 1.0] for t in input]
return ef
monkeypatch.setattr(embedding, "get_embedding_function", fake_ef_factory)
monkeypatch.setattr(
embedding,
"get_embedder_identity",
lambda device=None, model=None: types.SimpleNamespace(model_name="minilm", dimension=2),
)
def test_cache_roundtrip_and_wire_format(self, replica_palace):
from mempalace.vector_cache import VectorCache, b64_to_vector, vector_to_b64
cache = VectorCache(replica_palace)
try:
cache.put_many("minilm", [("d1", [0.25, -1.5]), ("d2", [3.0, 4.0])])
got = cache.get_many("minilm", ["d1", "d2", "d3"])
assert got["d1"] == [0.25, -1.5]
assert "d3" not in got
assert cache.missing_ids("minilm", ["d1", "d3"]) == ["d3"]
assert cache.count("minilm") == 2
assert cache.count("other-model") == 0
assert b64_to_vector(vector_to_b64([0.5, 0.5])) == [0.5, 0.5]
finally:
cache.close()
def test_build_cache_authored_only_and_resumable(
self, origin_server, fake_identity, fake_embedder
):
from mempalace.vector_cache import build_cache
url, origin_palace, _ = origin_server
stats = build_cache(origin_palace, "minilm", batch_size=2)
assert stats["embedded"] == 3 # the 3 seeded (authored) drawers
assert stats["cache_total"] == 3
again = build_cache(origin_palace, "minilm", batch_size=2)
assert again["embedded"] == 0
assert again["skipped_cached"] == 3
def test_endpoint_ships_vectors_and_manifest_advertises(
self, origin_server, fake_identity, fake_embedder
):
from mempalace.vector_cache import b64_to_vector, build_cache
url, origin_palace, _ = origin_server
build_cache(origin_palace, "minilm")
status, manifest = _get(url, "/snapshot/manifest")
assert manifest["vector_cache"] == {"minilm": 3}
status, page = _get(url, "/snapshot/drawers?offset=0&limit=10&vectors=minilm")
assert status == 200
assert page["vectors_included"] == 3
assert page["vectors_model"] == "minilm"
canary = next(i for i in page["items"] if "canary" in i["document"])
assert b64_to_vector(canary["vector_b64"]) == [1.0, 0.0]
def test_pull_with_vectors_is_insert_only(
self, origin_server, replica_palace, fake_identity, fake_embedder, monkeypatch
):
from mempalace.vector_cache import build_cache
url, origin_palace, _ = origin_server
build_cache(origin_palace, "minilm")
# Any local embedding during the fold is a failure of the design.
import mempalace.backends.embedding_wrapper as embedding_wrapper
def forbid(texts):
raise AssertionError(f"local embedding invoked for {len(texts)} docs")
monkeypatch.setattr(embedding_wrapper, "_embed_texts", forbid)
stats = pull_memory(replica_palace, url, with_vectors=True, pull_kg=False)
assert stats["vectors_used"] == 3
assert stats["locally_embedded"] == 0
# Folded vectors are live in the local index: vector search works.
from mempalace.palace import get_collection
col = get_collection(replica_palace)
result = col.query(query_embeddings=[[1.0, 0.0]], n_results=1)
assert "canary" in result["documents"][0][0]
def test_pull_with_vectors_refuses_when_peer_has_no_cache(
self, origin_server, replica_palace, fake_identity, fake_embedder
):
from mempalace.logsync import SyncPeerError
url, _, _ = origin_server
with pytest.raises(SyncPeerError, match="no vector cache"):
pull_memory(replica_palace, url, with_vectors=True, pull_kg=False)
def test_partial_cache_falls_back_locally(
self, origin_server, replica_palace, fake_identity, fake_embedder
):
from mempalace.vector_cache import VectorCache, build_cache
url, origin_palace, _ = origin_server
build_cache(origin_palace, "minilm")
# Simulate one drawer missing from the origin's cache.
cache = VectorCache(origin_palace)
try:
with cache._lock:
with cache._conn() as conn:
conn.execute("DELETE FROM vectors WHERE drawer_id = 'drawer_w_r_bbb'")
finally:
cache.close()
stats = pull_memory(replica_palace, url, with_vectors=True, pull_kg=False)
assert stats["vectors_used"] == 2
assert stats["locally_embedded"] == 1

1
website/.vitepress/config.mts
View File

@ -78,7 +78,6 @@ export default withMermaid(
{ text: 'Knowledge Graph', link: '/concepts/knowledge-graph' },
{ text: 'Specialist Agents', link: '/concepts/agents' },
{ text: 'Agent Logstream', link: '/concepts/agent-logstream' },
{ text: 'Replicated Palace', link: '/concepts/replicated-palace' },
{ text: 'Contradiction Detection', link: '/concepts/contradiction-detection' },
],
},

192
website/concepts/replicated-palace.md
View File

@ -1,192 +0,0 @@
# The Replicated Palace
Memory is identity — and you don't park your identity on a single machine.
The replicated palace turns MemPalace from *one palace on one computer* into
**one logical palace, fully replicated across every machine you own**. Agents
always talk to the MemPalace service on `127.0.0.1`; the services converge
with each other in the background. If your desktop sleeps, your laptop still
remembers everything. When it wakes, the machines reconcile on their own.
This is the design from RFC 004, and it is running today as the project's own
production infrastructure — the maintainers' agent fleet coordinates and
remembers through it.
## The availability invariant
One rule governs everything here: **recall reads and capture writes never
block on the network — only convergence may wait.** A memory system that adds
a network round-trip to remembering has stopped being local-first, so every
mesh feature is judged against offline operation as the default posture, not
an edge case.
## Three layers
```
Machine A Machine B Machine C
┌─────────────────────┐ ┌─────────────────────┐ ┌──────────────────┐
│ agents → 127.0.0.1 │ ops │ agents → 127.0.0.1 │ ops │ agents → local │
│ ┌─────────────────┐ │ ◀─────▶ │ ┌─────────────────┐ │◀────▶│ ┌──────────────┐ │
│ │ mempalace hub │ │ │ │ mempalace hub │ │ │ │ mempalace hub│ │
│ │ event log │ │ │ │ event log │ │ │ │ event log │ │
│ │ derived index │ │ │ │ derived index │ │ │ │ derived index│ │
│ └─────────────────┘ │ │ └─────────────────┘ │ │ └──────────────┘ │
└─────────────────────┘ └─────────────────────┘ └──────────────────┘
```
1. **Transport** — how replicas reach and trust each other. Today: any
mutually reachable network (a Tailscale-style tailnet works well) with
per-hub bearer tokens. The transport lives behind a seam
(`MEMPALACE_TRANSPORT`), so a decentralized mesh-identity transport can
replace tokens without touching anything above it.
2. **Sync** — append-only logs of *ops*, merged by union. Each op is
immutable and provenance-stamped, and travels between replicas. Today
coordination events and artifacts move this way; memory content moves by
one-way replica pull until the memory op-log lands (see
[What syncs today](#what-syncs-today)).
3. **Derived state** — vector indexes and caches are rebuilt or folded
locally, never copied. **Sync the facts, derive the senses**: ops are
kilobytes; vector indexes are gigabytes. Every machine remembers
everything; each machine senses with its own hardware.
## Ops, clocks, and version vectors
Each replica has a stable identity (`replica.json`, e.g.
`rep_0123456789abcdef0123456789abcdef`) and stamps everything it authors with:
- `origin_replica` — which machine authored it (provenance, forever)
- `origin_seq` — the author's own counter
- `hlc` — a hybrid logical clock: physical time + logical counter + replica
tiebreak, rendered as a sortable string. Total order across machines
without trusting anyone's wall clock.
A replica's knowledge is summarized by a **version vector** —
`{origin → highest sequence applied}`. Peers exchange vectors, compute
exactly which op ranges each is missing, and pull them. The engine is
idempotent end to end: re-delivering an op is a no-op, a crash mid-round
means the next round re-pulls the tail, and **every replica carries every
origin's ops** — so two machines that have never exchanged credentials still
converge through a common peer. Gossip, in the practical sense.
## What syncs today
| Layer | Mechanism | Status |
|---|---|---|
| Coordination events + artifacts (the [agent logstream](/concepts/agent-logstream)) | op sync, multi-master | shipping |
| Memory content (drawers) | snapshot pull + local fold (**one-way**) | shipping |
| Knowledge graph | snapshot pull + local fold (**one-way**) | shipping |
| Vectors | never synced — derived locally, or folded from a peer's [vector cache](#distributed-embedding) | shipping |
| Memory content (drawers), bidirectional | memory op-log + anti-entropy | next |
| Organization (wings/rooms/tunnels as ops) | op vocabulary reserved | next |
Read the split carefully, because it is the difference between what works
today and what the rest of this page describes. **Coordination is already
multi-master**: any agent on any machine appends events, and the logstream
converges in both directions. **Memory is not yet.** Drawers and graph facts
move via `mempalace replica pull` — a one-way, insert-only fold from an
origin you name. Two machines that each capture their own conversations do
not merge; each pulls what it wants from the other.
The **memory op-log** — provenance-stamped ops for every drawer and graph
write, anti-entropy sync, and a fold that resolves cross-replica edits by
last-writer-wins — is the mechanism that closes that gap. It is designed
(RFC 004 step 2a) and staged for a later release, along with the
content-pure id recipe it depends on. Until it lands, treat each replica's
own captures as authoritative locally.
## Bootstrapping a new machine
A new replica doesn't replay months of history — it bootstraps from a
snapshot, then tails ops:
```bash
# on the new machine, hub not yet running
mempalace replica pull --with-vectors
```
`peers.json` in the palace directory names the peers:
```json
{
"peers": [
{ "name": "desktop", "url": "https://desktop.example.com", "token": "..." },
{ "name": "laptop", "url": "https://laptop.example.com", "token": "..." }
]
}
```
Each origin serves only the content it *authored* (never copies it received
from someone else), so bidirectional pulls converge without echo. Pulled
drawers carry a `replica_origin` stamp naming where they came from — ask
"who is X?" on any machine and the answer arrives with its provenance.
Once the hub starts, the background sync loop (every
`MEMPALACE_SYNC_INTERVAL` seconds, default 15) keeps the logstream
converging on its own — no cron jobs, no manual steps. Memory pulls are
still an explicit `mempalace replica pull` until the memory op-log lands.
## Distributed embedding
Vectors are a pure function of *(content, embedder identity)* — so they can
be computed wherever the best hardware lives:
```bash
# on the machine with the GPU: precompute vectors into a portable cache
mempalace replica embed-cache
# on any other machine: fold content WITH those vectors — zero local embedding
mempalace replica pull --with-vectors
```
The cache is a sidecar keyed by `(drawer_id, embedder identity)`; the fold is
identity-gated, so vectors are only ever reused under the exact same model.
Measured on real palace content (MiniLM, 384-d): a desktop CPU manages
~100 docs/s, a commodity CUDA GPU ~1,300 docs/s, and a current-generation
Windows GPU via DirectML ~2,550 docs/s — a one-line
`pip install onnxruntime-directml`, no vendor toolchain. A six-figure drawer
count that would take half an hour of CPU embedding folds in about a minute
of GPU time computed once, anywhere in your mesh.
## Watching the mesh
Every hub answers `GET /sync/peers` (bearer-authenticated) with its view of
the estate: its own identity and version vector, each configured peer's
reachability, last sync outcome and remote vector, and — the interesting
part — `unnamed_origins`: replicas it knows about *only through gossip*.
Drift between any two machines is one vector comparison. This endpoint is
what mesh dashboards draw from; tokens are never included in the payload.
The same payload is exposed as the `mempalace_mesh_peers` MCP tool, so
desktop apps consume the estate through the bridge they already have.
Every node also advertises a **profile** — roles (`replica` / `agents` /
`compute`), resolved accelerator and embedder, live drawer count, hardware
string — derived entirely from what the daemon can observe about itself,
never from configuration. Profiles ride the sync surfaces, so a carrier
relays them for replicas it only knows transitively: dashboards render
what each machine *reported about itself*, not what a UI guessed.
## Trust, today and next
Today, authorization is a bearer token per hub, exchanged out-of-band by the
human — workable, but it costs one manual credential relay per edge and has
no real revocation story. The planned transport binds the mesh identity
itself: each device's cryptographic key becomes its replica id (provenance
and authentication as one fact), mesh membership becomes the only ACL,
admission becomes a one-time join ceremony, and revoking a lost laptop is a
single command that propagates. The transport seam exists so that swap
touches none of the sync machinery above it.
Two things replication is **not**:
- **Not a cloud.** No third party ever holds queryable palace content. The
mesh is your machines converging with your machines.
- **Not a backup.** Deletions propagate faithfully — a mass delete
replicates like anything else. Keep snapshots separately.
## See also
- [Shared Brain guide](/guide/shared-brain) — the operational setup, hub and
agents included
- [Agent Logstream](/concepts/agent-logstream) — the coordination layer that
pioneered the op-sync machinery
- [CLI reference](/reference/cli) — `mempalace replica`

67
website/guide/shared-brain.md
View File

@ -345,20 +345,18 @@ The same non-negotiables that govern memory govern coordination:
`patch_submit` — are hidden and refused. Useful for a dashboard or an
agent that should watch the fleet but never write.
## From hub to mesh
## Coordinating across machines
Everything above uses one hub as the fleet's shared memory — which also
makes that machine a single point of failure: when it sleeps, every other
machine loses recall, capture, and coordination at once. The next stage
removes that dependency: **every machine runs its own hub over a full local
replica, and the hubs converge with each other** ([The Replicated
Palace](/concepts/replicated-palace) explains the architecture).
Everything above uses one hub as the fleet's shared memory. Agents on other
machines can join that hub's coordination stream without giving up their own
local palace: **each machine runs its own hub, and the hubs sync their
logstreams with each other.** An agent's inbox then survives any single
machine sleeping.
Joining the mesh is three steps per machine:
Two steps per machine:
1. **Run a hub locally** (same `mempalace serve` as above, LaunchAgent /
systemd unit recommended) — agents on that machine now point at
`127.0.0.1` instead of a remote hub.
systemd unit recommended) — agents on that machine point at `127.0.0.1`.
2. **Name the peers** in `peers.json` in the palace directory — each entry
is a `name`, the peer hub's `url`, and its bearer `token` (exchange
tokens out-of-band; never through the coordination stream):
@ -372,43 +370,32 @@ Joining the mesh is three steps per machine:
```
The hub's background loop picks up `peers.json` changes within one sync
cycle — coordination events and artifacts converge every
`MEMPALACE_SYNC_INTERVAL` seconds (default 15) with no further action.
3. **Pull the memory**, with the local hub stopped:
`mempalace replica pull --with-vectors` folds every peer's authored
content — and their precomputed vectors — into the local palace. See the
[CLI reference](/reference/cli#mempalace-replica).
cycle — events and artifacts converge every `MEMPALACE_SYNC_INTERVAL`
seconds (default 15) with no further action. Sync is multi-master and
idempotent: every replica carries every origin's events, so two machines
that have never exchanged credentials still converge through a common
peer, and a machine that was offline for a week just re-pulls the tail.
After that, delegation works exactly as described in this guide — but an
agent's inbox survives any single machine sleeping.
`GET /sync/peers` on any hub shows the estate: which peers were reachable
last round, their version vectors, and any replicas known only through
gossip. The same payload is the `mempalace_mesh_peers` MCP tool.
::: warning Coordination converges on its own; memory does not yet
Step 2 is continuous and bidirectional. Step 3 is a **one-way pull you
re-run**: it folds what the peers have authored into this machine, and it is
insert-only and resumable, so re-running it heals any gap. What it does not
do is merge — a drawer edited on two machines will not reconcile itself.
Bidirectional memory convergence is the memory op-log (RFC 004 step 2a),
staged for a later release. Until then, put `mempalace replica pull` on a
schedule if you want each machine to stay current.
::: `GET /sync/peers` on any hub shows the estate: which peers were
reachable last round, their version vectors, and any replicas known only
through gossip.
::: warning This syncs coordination, not memory
Peer sync covers the **logstream** — events and artifacts. Each machine's
drawers and knowledge graph stay local to that machine. Agents on two
synced machines share an inbox and can hand patches back and forth, but
they do not yet share recall: ask one of them what it remembers and you get
that machine's palace.
::: tip Already have a palace on that machine?
Joining the mesh is **additive**. An existing palace — even one built
long before replication existed — keeps every drawer it has: nothing is
re-mined, nothing is lost, and the bootstrap pull folds *around* your
existing content. Your machine's history becomes part of the shared brain
in the other direction too: run `mempalace replica embed-cache` once and
every peer can pull your palace's full past, vectors included, in minutes.
Capture-now is forward-compatible by contract — memory filed years before
the mesh is a first-class citizen of it.
Replicating memory itself is [RFC 004](https://github.com/MemPalace/mempalace/blob/develop/docs/rfcs/004-replicated-palace.md),
staged for a later release. If you want one shared memory across machines
today, point every agent at a single hub ([Remote / Team
Server](/guide/remote-server)) instead of running one per machine.
:::
## See also
- [The Replicated Palace](/concepts/replicated-palace) — one palace, every machine: ops, clocks, folds
- [Agent Logstream](/concepts/agent-logstream) — the event/artifact model in depth
- [Remote / Team Server](/guide/remote-server) — full hub deployment: tokens, TLS, backends, Docker/systemd
- [MCP Integration](/guide/mcp-integration) — the memory tools every connected agent gets
- [CLI Reference](/reference/cli#mempalace-logstream) — `mempalace logstream`, `mempalace artifact`, `mempalace replica`
- [CLI Reference](/reference/cli#mempalace-logstream) — `mempalace logstream`, `mempalace artifact`

41
website/reference/cli.md
View File

@ -214,44 +214,3 @@ mempalace artifact get art_... --out /tmp/handoff.patch
|------------|-------------|
| `put` | Store content (`--kind patch\|file\|log\|json\|note`, `--created-by` required; `--content`, `--file`, or stdin) |
| `get` | Print exact content to stdout, or `--out FILE`; `--json` for metadata |
## `mempalace replica`
Memory replication across your machines (RFC 004; see
[The Replicated Palace](/concepts/replicated-palace)). Bootstraps a new
replica from its peers and moves precomputed vectors between machines.
```bash
# Bootstrap: fold every peer's authored content into this palace.
# STOP the local hub first — this writes the palace directly.
mempalace replica pull --with-vectors
# One specific origin instead of peers.json:
mempalace replica pull --peer https://desktop.example.com --token "$TOKEN"
# Precompute vectors into the portable cache (safe alongside a live hub):
mempalace replica embed-cache --batch 512 --json
```
| Subcommand | Description |
|------------|-------------|
| `pull` | Fold drawers + knowledge graph from origins (`--peer`/`--token` or `peers.json`; `--with-vectors` uses origin-precomputed vectors, `--no-kg`, `--no-reconcile`) |
| `embed-cache` | Bulk-embed local content into `vector_cache.sqlite3` so peers can pull `--with-vectors` (`--model`, `--batch`, `--all`) |
`pull` requires the local hub to be stopped (single-writer rule) and the
origins to be quiescent (no active mines). Pulls are insert-only and
resumable — re-running heals any gap. Raise
`MEMPALACE_SYNC_HTTP_TIMEOUT` (seconds, default 30) for large bootstraps.
A running hub syncs the logstream automatically every
`MEMPALACE_SYNC_INTERVAL` seconds (default 15). The CLI verbs exist for
bootstraps, offline machines, and inspection.
::: tip Full convergence is staged
`mempalace replica pull` gives you **read replicas**: a one-way fold of
drawers and graph facts from an origin. Bidirectional convergence — where
every replica is an equal writer and edits merge automatically — is RFC 004
step 2a (the memory op-log), staged for a later release. Until then, treat
each replica's own captures as authoritative locally and pull from the
origins that hold what you want mirrored.
:::

2
website/reference/mcp-tools.md
View File

@ -612,7 +612,7 @@ Convenience: store a patch artifact and append its `patch.ready` event in one ca
### `mempalace_mesh_peers`
Mesh estate snapshot (see [The Replicated Palace](/concepts/replicated-palace)):
Mesh estate snapshot — this hub's view of its logstream peers (see [Shared Brain](/guide/shared-brain#coordinating-across-machines)):
this replica's identity, version vector and self-derived node profile; each
configured peer's reachability, last sync outcome, remote version vector and
advertised profile; origins known only transitively; and `origin_profiles`