diff --git a/mempalace/cli.py b/mempalace/cli.py index c2881cc..a150a03 100644 --- a/mempalace/cli.py +++ b/mempalace/cli.py @@ -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() diff --git a/mempalace/mcp_server.py b/mempalace/mcp_server.py index 85cd537..45543f7 100644 --- a/mempalace/mcp_server.py +++ b/mempalace/mcp_server.py @@ -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) diff --git a/mempalace/replica_sync.py b/mempalace/replica_sync.py deleted file mode 100644 index fd98ce8..0000000 --- a/mempalace/replica_sync.py +++ /dev/null @@ -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 diff --git a/mempalace/vector_cache.py b/mempalace/vector_cache.py deleted file mode 100644 index 1cd9f1f..0000000 --- a/mempalace/vector_cache.py +++ /dev/null @@ -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() diff --git a/tests/test_logstream_sync.py b/tests/test_logstream_sync.py index c3fdf81..a6a5412 100644 --- a/tests/test_logstream_sync.py +++ b/tests/test_logstream_sync.py @@ -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") diff --git a/tests/test_replica_sync.py b/tests/test_replica_sync.py deleted file mode 100644 index c832f8a..0000000 --- a/tests/test_replica_sync.py +++ /dev/null @@ -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 diff --git a/website/.vitepress/config.mts b/website/.vitepress/config.mts index 0641792..680296d 100644 --- a/website/.vitepress/config.mts +++ b/website/.vitepress/config.mts @@ -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' }, ], }, diff --git a/website/concepts/replicated-palace.md b/website/concepts/replicated-palace.md deleted file mode 100644 index e07020d..0000000 --- a/website/concepts/replicated-palace.md +++ /dev/null @@ -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` diff --git a/website/guide/shared-brain.md b/website/guide/shared-brain.md index 9e01531..dd73879 100644 --- a/website/guide/shared-brain.md +++ b/website/guide/shared-brain.md @@ -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` diff --git a/website/reference/cli.md b/website/reference/cli.md index e801d27..9af9134 100644 --- a/website/reference/cli.md +++ b/website/reference/cli.md @@ -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. -::: diff --git a/website/reference/mcp-tools.md b/website/reference/mcp-tools.md index d8bef28..06d21d4 100644 --- a/website/reference/mcp-tools.md +++ b/website/reference/mcp-tools.md @@ -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`