From c740ba4611852f0dfc1bbbee68120d243aceae9b Mon Sep 17 00:00:00 2001 From: Oleksii Pylypchuk Date: Mon, 13 Jul 2026 21:00:03 +0300 Subject: [PATCH 1/4] feat(search): add authority-backed memory freshness Expose optional authority provenance and verification on retrieved memories so agents can distinguish current, stale, and unverified evidence. Cache local-file verification during each search and deduplicate ranked hits without changing legacy retrieval defaults. --- mempalace/mcp_server.py | 50 ++++++++++++++++- mempalace/searcher.py | 114 ++++++++++++++++++++++++++++++++++++++- tests/test_mcp_server.py | 33 ++++++++++++ tests/test_searcher.py | 66 +++++++++++++++++++++++ 4 files changed, 259 insertions(+), 4 deletions(-) diff --git a/mempalace/mcp_server.py b/mempalace/mcp_server.py index ff4f28fd7..5ba5e7d8a 100644 --- a/mempalace/mcp_server.py +++ b/mempalace/mcp_server.py @@ -2373,6 +2373,7 @@ def tool_search( max_distance: float = 1.5, min_similarity: float = None, context: str = None, + verify_authority: bool = False, ): limit = max(1, min(limit, _MAX_RESULTS)) try: @@ -2406,6 +2407,7 @@ def tool_search( max_distance=dist, vector_disabled=_vector_disabled, collection_name=_config.collection_name, + verify_authority=bool(verify_authority), ) if _is_transient_index_error(result): # Post-bulk-write HNSW flush window (#1315): drop caches, give @@ -2426,6 +2428,7 @@ def tool_search( max_distance=dist, vector_disabled=_vector_disabled, collection_name=_config.collection_name, + verify_authority=bool(verify_authority), ) if not _is_transient_index_error(result): result["index_recovered"] = True @@ -2915,7 +2918,14 @@ def _build_chunk_rows(drawer_id: str, content: str, meta: dict, chunk_size: int) def tool_add_drawer( - wing: str, room: str, content: str, source_file: str = None, added_by: str = "mcp" + wing: str, + room: str, + content: str, + source_file: str = None, + added_by: str = "mcp", + authority_uri: str = None, + authority_version: str = None, + memory_kind: str = None, ): """File verbatim content into a wing/room. Checks for duplicates first. @@ -2937,6 +2947,9 @@ def tool_add_drawer( if source_file: source_file = strip_lone_surrogates(source_file) added_by = strip_lone_surrogates(added_by) + authority_uri = strip_lone_surrogates(authority_uri or "") + authority_version = strip_lone_surrogates(authority_version or "") + memory_kind = sanitize_name(memory_kind, "memory_kind") if memory_kind else "" except ValueError as e: return {"success": False, "error": str(e)} @@ -2967,6 +2980,12 @@ def tool_add_drawer( "filed_at": datetime.now().isoformat(), "id_recipe": ID_RECIPE, } + if authority_uri: + base_meta["authority_uri"] = authority_uri + if authority_version: + base_meta["authority_version"] = authority_version + if memory_kind: + base_meta["memory_kind"] = memory_kind # Idempotency. Three cases to detect a prior committed write: # (a) Single-doc path: drawer_id row exists (the only id used). @@ -4427,7 +4446,15 @@ def tool_checkpoint(items, diary=None, dedup_threshold=0.9, added_by=None): # string by the guard above) we still file rather than drop the # memory: verbatim recall is the priority and add_drawer's own # idempotency blocks exact duplicates. - res = tool_add_drawer(wing=wing, room=room, content=content, added_by=resolved_added_by) + res = tool_add_drawer( + wing=wing, + room=room, + content=content, + added_by=resolved_added_by, + authority_uri=item.get("authority_uri"), + authority_version=item.get("authority_version"), + memory_kind=item.get("memory_kind"), + ) if res.get("success"): out["added"].append(res) else: @@ -4988,6 +5015,10 @@ def tool_patch_submit( "type": "string", "description": "Background context for the search (optional). NOT used for embedding — only for future re-ranking.", }, + "verify_authority": { + "type": "boolean", + "description": "Verify file:// or absolute-path authority version tokens. Adds current/stale/unverified status to each result.", + }, }, "required": ["query"], }, @@ -5024,6 +5055,18 @@ def tool_patch_submit( }, "source_file": {"type": "string", "description": "Where this came from (optional)"}, "added_by": {"type": "string", "description": "Who is filing this (default: mcp)"}, + "authority_uri": { + "type": "string", + "description": "Canonical local file path or file:// URI (optional)", + }, + "authority_version": { + "type": "string", + "description": "sha256: or mtime_ns: token (optional)", + }, + "memory_kind": { + "type": "string", + "description": "Memory class such as decision, finding, or preference (optional)", + }, }, "required": ["wing", "room", "content"], }, @@ -5049,6 +5092,9 @@ def tool_patch_submit( "type": "string", "description": "Verbatim content to store", }, + "authority_uri": {"type": "string"}, + "authority_version": {"type": "string"}, + "memory_kind": {"type": "string"}, }, "required": ["wing", "room", "content"], }, diff --git a/mempalace/searcher.py b/mempalace/searcher.py index 174a97379..764a704fd 100644 --- a/mempalace/searcher.py +++ b/mempalace/searcher.py @@ -13,11 +13,14 @@ import logging import math import os +import hashlib import re import sqlite3 from datetime import timedelta from pathlib import Path from typing import Optional +from urllib.parse import urlsplit +from urllib.request import url2pathname from .backends import ( BackendError, @@ -805,6 +808,7 @@ def _bm25_only_via_sqlite( stop_words: frozenset = frozenset(), since_dt=None, before_dt=None, + verify_authority: bool = False, ) -> dict: """BM25-only search reading drawers directly from chroma.sqlite3. @@ -1007,6 +1011,7 @@ def _metadata_filter_sql(row_id_expr: str) -> tuple[str, list[str]]: # Apply wing/room filters in Python (FTS5 candidates may include # entries from other wings). candidates = [] + authority_cache: dict[tuple[str, str, int, int], dict] = {} for d in drawers.values(): meta = d["metadata"] if wing and meta.get("wing") != wing: @@ -1039,6 +1044,9 @@ def _metadata_filter_sql(row_id_expr: str) -> tuple[str, list[str]]: # multiple chunks. Stripped before this helper returns. "_source_file_full": full_source, "_chunk_index": meta.get("chunk_index"), + "authority": resolve_authority_status( + meta, verify=verify_authority, cache=authority_cache + ), } ) @@ -1207,6 +1215,93 @@ def _candidate_pool_size(n_results: int, date_window_active: bool) -> int: } +def _local_authority_path(uri: str) -> Optional[Path]: + """Resolve an absolute path or local file URI without losing escaped characters.""" + if uri.startswith("file:"): + parsed = urlsplit(uri) + if parsed.scheme != "file" or parsed.netloc not in {"", "localhost"}: + return None + return Path(url2pathname(parsed.path)).expanduser() + path = Path(uri).expanduser() + return path if path.is_absolute() else None + + +def resolve_authority_status( + metadata: dict, + *, + verify: bool = False, + cache: Optional[dict[tuple[str, str, int, int], dict]] = None, +) -> dict: + """Return a stable authority envelope for a search hit. + + The PoC deliberately supports only local file authorities. Unknown schemes + remain ``unverified`` so retrieval never upgrades an unsupported authority + to current by assumption. + """ + uri = str((metadata or {}).get("authority_uri") or "") + version = str((metadata or {}).get("authority_version") or "") + declared = str((metadata or {}).get("authority_status") or "") + result = {"uri": uri, "version": version, "status": declared or "unverified"} + if not uri or not version: + result["reason"] = "missing_authority" if not uri else "missing_version" + return result + if declared in {"stale", "superseded", "historical"}: + result["reason"] = f"declared_{declared}" + return result + if not verify: + result["status"] = declared or "unverified" + result["reason"] = "verification_not_requested" + return result + + path = _local_authority_path(uri) + if path is None: + result["status"] = "unverified" + result["reason"] = "unsupported_authority_uri" + return result + if not path.is_file(): + result["status"] = "stale" + result["reason"] = "authority_missing" + return result + stat = path.stat() + cache_key = (uri, version, stat.st_mtime_ns, stat.st_size) + if cache is not None and cache_key in cache: + return dict(cache[cache_key]) + if version.startswith("mtime_ns:"): + actual = f"mtime_ns:{stat.st_mtime_ns}" + elif version.startswith("sha256:"): + actual = f"sha256:{hashlib.sha256(path.read_bytes()).hexdigest()}" + else: + result["status"] = "unverified" + result["reason"] = "unsupported_authority_version" + return result + result["actual_version"] = actual + result["status"] = "current" if actual == version else "stale" + result["reason"] = "authority_matches" if actual == version else "authority_changed" + if cache is not None: + cache[cache_key] = dict(result) + return result + + +def _deduplicate_ranked_hits(hits: list[dict]) -> list[dict]: + """Collapse repeated logical hits while preserving distinct source sections.""" + deduplicated = [] + seen = set() + for hit in hits: + source = hit.get("_source_file_full") or hit.get("source_path") or hit.get("source_file") + parent = hit.get("_parent_drawer_id") or "" + if hit.get("drawer_index") is not None: + key = ("hydrated", source, parent, hit.get("drawer_index")) + elif source and hit.get("_chunk_index") is not None: + key = ("chunk", source, parent, hit.get("_chunk_index")) + else: + key = ("text", source, hit.get("text")) + if key in seen: + continue + seen.add(key) + deduplicated.append(hit) + return deduplicated + + def _validate_candidate_strategy(strategy: str) -> None: """Raise ``ValueError`` for unknown strategies. @@ -1292,9 +1387,10 @@ def _finalize_candidate_hits( ), ) - hits = _hybrid_rank( + ranked = _hybrid_rank( hits, query, metric=_metric_for_collection(drawers_col), stop_words=stop_words - )[:n_results] + ) + hits = _deduplicate_ranked_hits(ranked)[:n_results] for h in hits: h.pop("_sort_key", None) h.pop("_source_file_full", None) @@ -1381,6 +1477,7 @@ def _window_and_fallback_gate( collection_name, source_file, stop_words: frozenset = frozenset(), + verify_authority: bool = False, ): """Front gate for ``search_memories``: parse the window, route the fallback. @@ -1414,6 +1511,7 @@ def _window_and_fallback_gate( since_dt=since_dt, before_dt=before_dt, stop_words=stop_words, + verify_authority=verify_authority, ), ) return since_dt, before_dt, active, None @@ -1449,6 +1547,7 @@ def _vector_disabled_with_window( since_dt, before_dt, stop_words: frozenset = frozenset(), + verify_authority: bool = False, ) -> dict: """Run the BM25-only route and echo the raw window strings. @@ -1467,6 +1566,7 @@ def _vector_disabled_with_window( since_dt=since_dt, before_dt=before_dt, stop_words=stop_words, + verify_authority=verify_authority, ) if "filters" in result: result["filters"]["since"] = since @@ -1486,6 +1586,7 @@ def _vector_disabled_search( stop_words: frozenset = frozenset(), since_dt=None, before_dt=None, + verify_authority: bool = False, ) -> dict: try: backend_name = resolve_backend_name(palace_path) @@ -1511,6 +1612,7 @@ def _vector_disabled_search( stop_words=stop_words, since_dt=since_dt, before_dt=before_dt, + verify_authority=verify_authority, ) @@ -1612,6 +1714,7 @@ def search_memories( candidate_strategy: str = "vector", collection_name: str = None, lang: Optional[str] = None, + verify_authority: bool = False, ) -> dict: """Programmatic search — returns a dict instead of printing. @@ -1644,6 +1747,8 @@ def search_memories( detects a divergence that would segfault chromadb on segment load. candidate_strategy: How candidates for the hybrid re-rank are gathered. + verify_authority: Resolve local-file authority version tokens and mark + results current or stale. Disabled by default to keep search cheap. * ``"vector"`` (default) — preserves historical behavior: top ``n_results * 3`` rows from the vector index are the rerank pool. @@ -1688,6 +1793,7 @@ def search_memories( collection_name=collection_name, source_file=source_file, stop_words=stop_words, + verify_authority=verify_authority, ) if short_circuit is not None: return short_circuit @@ -1755,6 +1861,7 @@ def search_memories( CLOSET_DISTANCE_CAP = 1.5 # cosine dist > 1.5 = too weak to use as signal scored: list = [] + authority_cache: dict[tuple[str, str, int, int], dict] = {} drawer_docs = _first_or_empty(drawer_results, "documents") stored_drawer_ids = _aligned_query_ids(drawer_results, len(drawer_docs)) for stored_drawer_id, doc, meta, dist in zip( @@ -1810,6 +1917,9 @@ def search_memories( "_source_file_full": source, "_chunk_index": meta.get("chunk_index"), "_parent_drawer_id": meta.get("parent_drawer_id"), + "authority": resolve_authority_status( + meta, verify=verify_authority, cache=authority_cache + ), } if closet_preview: entry["closet_preview"] = closet_preview diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 7bf3c5e50..dcd55f6c5 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -2172,6 +2172,39 @@ def test_checkpoint_files_items_and_writes_diary(self, monkeypatch, config, pala assert all(a["success"] for a in result["added"]) assert result["diary"]["success"] is True + def test_checkpoint_forwards_authority_metadata(self, monkeypatch, config, kg): + from mempalace import mcp_server + + monkeypatch.setattr( + mcp_server, "tool_check_duplicate", lambda *_a, **_k: {"is_duplicate": False} + ) + filed = {} + + def _add(**kwargs): + filed.update(kwargs) + return {"success": True, "drawer_id": "d1"} + + monkeypatch.setattr(mcp_server, "tool_add_drawer", _add) + result = mcp_server.tool_checkpoint( + items=[ + { + "wing": "w", + "room": "decisions", + "content": "Use the planner as authority.", + "authority_uri": "file:///repo/planner/task.md", + "authority_version": "sha256:abc", + "memory_kind": "decision", + } + ], + added_by="planner-agent", + ) + + assert result["errors"] == [] + assert filed["added_by"] == "planner-agent" + assert filed["authority_uri"] == "file:///repo/planner/task.md" + assert filed["authority_version"] == "sha256:abc" + assert filed["memory_kind"] == "decision" + def test_checkpoint_skips_semantic_duplicates(self, monkeypatch, config, kg): from mempalace import mcp_server diff --git a/tests/test_searcher.py b/tests/test_searcher.py index 30db70d91..3748107a0 100644 --- a/tests/test_searcher.py +++ b/tests/test_searcher.py @@ -7,6 +7,8 @@ import sqlite3 from datetime import datetime +import hashlib +from pathlib import Path from unittest.mock import MagicMock, patch import pytest @@ -1533,3 +1535,67 @@ def test_cli_search_inverted_window_raises(self, palace_path, seeded_collection) with pytest.raises(SearchError, match="must be earlier than"): search("anything", palace_path, since="2026-01-04", before="2026-01-01") + + +def test_resolve_authority_status_detects_current_and_stale_file(tmp_path): + from mempalace.searcher import resolve_authority_status + + authority = tmp_path / "plan.md" + authority.write_text("v1", encoding="utf-8") + version = f"sha256:{hashlib.sha256(authority.read_bytes()).hexdigest()}" + metadata = {"authority_uri": str(authority), "authority_version": version} + + assert resolve_authority_status(metadata, verify=True)["status"] == "current" + authority.write_text("v2", encoding="utf-8") + stale = resolve_authority_status(metadata, verify=True) + assert stale["status"] == "stale" + assert stale["reason"] == "authority_changed" + + +def test_resolve_authority_status_is_unverified_by_default(tmp_path): + from mempalace.searcher import resolve_authority_status + + result = resolve_authority_status( + {"authority_uri": str(tmp_path / "plan.md"), "authority_version": "sha256:abc"} + ) + assert result["status"] == "unverified" + assert result["reason"] == "verification_not_requested" + + +def test_resolve_authority_status_decodes_local_file_uri(tmp_path): + from mempalace.searcher import resolve_authority_status + + authority = tmp_path / "plan with spaces.md" + authority.write_text("v1", encoding="utf-8") + version = f"sha256:{hashlib.sha256(authority.read_bytes()).hexdigest()}" + + result = resolve_authority_status( + {"authority_uri": authority.as_uri(), "authority_version": version}, verify=True + ) + + assert result["status"] == "current" + assert result["reason"] == "authority_matches" + + +def test_resolve_authority_status_reuses_per_search_cache(tmp_path, monkeypatch): + from mempalace.searcher import resolve_authority_status + + authority = tmp_path / "plan.md" + authority.write_text("v1", encoding="utf-8") + version = f"sha256:{hashlib.sha256(authority.read_bytes()).hexdigest()}" + metadata = {"authority_uri": authority.as_uri(), "authority_version": version} + original = Path.read_bytes + calls = 0 + + def counted(path): + nonlocal calls + calls += 1 + return original(path) + + monkeypatch.setattr(Path, "read_bytes", counted) + cache = {} + first = resolve_authority_status(metadata, verify=True, cache=cache) + second = resolve_authority_status(metadata, verify=True, cache=cache) + + assert first == second + assert calls == 1 From fec6d54eb7f08dd4164505ec2e9470022b80084c Mon Sep 17 00:00:00 2001 From: Oleksii Pylypchuk Date: Mon, 13 Jul 2026 21:03:19 +0300 Subject: [PATCH 2/4] feat(ingest): add changed-set project updates Accept backend-owned changed/deleted manifests so project updates avoid a full filesystem scan. Keep path containment, tombstones, gitignore cleanup, and wing-scoped reindexing explicit to preserve the existing corpus safely. --- docs/changed-set-ingest.md | 29 ++++++++ mempalace/changed_set.py | 139 ++++++++++++++++++++++++++++++++++ mempalace/cli.py | 46 ++++++++++++ mempalace/miner.py | 50 +++++++++---- mempalace/palace.py | 13 +++- mempalace/service.py | 15 ++++ tests/test_changed_set.py | 148 +++++++++++++++++++++++++++++++++++++ tests/test_closets.py | 1 + tests/test_daemon.py | 32 ++++++++ tests/test_miner.py | 89 +++++++++++++++++++++- 10 files changed, 545 insertions(+), 17 deletions(-) create mode 100644 docs/changed-set-ingest.md create mode 100644 mempalace/changed_set.py create mode 100644 tests/test_changed_set.py diff --git a/docs/changed-set-ingest.md b/docs/changed-set-ingest.md new file mode 100644 index 000000000..0b09092e4 --- /dev/null +++ b/docs/changed-set-ingest.md @@ -0,0 +1,29 @@ +# Changed-set ingest PoC + +`mempalace sync` can accept a producer-supplied manifest and reindex only the +listed project-relative files. Git is not required by core; an IDE, watcher, or +build system may create the same JSON shape. + +```json +{ + "changed": ["src/app.py", "README.md"], + "deleted": ["src/old.py"] +} +``` + +Preview and apply: + +```bash +mempalace sync /path/to/project --manifest changed.json --wing project +mempalace sync /path/to/project --manifest changed.json --wing project --apply --daemon +``` + +Paths must remain within the project root. Apply holds the palace writer lock, +purges old drawers and closets for every affected source, and invokes the normal +project miner only for `changed`. `deleted` sources are never opened. The daemon +payload contains the parsed manifest, avoiding a manifest-file time-of-check / +time-of-use race between client and writer. + +This is intentionally a PoC contract. A production version should add a job +idempotency key and committed palace generation before making changed-set sync a +default hook path. diff --git a/mempalace/changed_set.py b/mempalace/changed_set.py new file mode 100644 index 000000000..d3be38445 --- /dev/null +++ b/mempalace/changed_set.py @@ -0,0 +1,139 @@ +"""Changed-set project ingest without a full filesystem walk. + +The producer may be Git, an IDE, or a file watcher. Core accepts only project- +relative paths and keeps all mutation semantics inside MemPalace. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import TypedDict + +from .miner import ( + is_gitignored, + load_config, + load_gitignore_matcher, + process_file, +) +from .palace import ( + get_closets_collection, + get_collection, + mine_palace_lock, +) + + +class ChangedSetReport(TypedDict): + changed: int + deleted: int + ignored: int + reindexed: int + drawers_added: int + dry_run: bool + + +def _resolve_source(root: Path, value: str) -> Path: + if not isinstance(value, str) or not value.strip(): + raise ValueError("changed-set paths must be non-empty strings") + candidate = (root / value).resolve(strict=False) + try: + candidate.relative_to(root) + except ValueError as exc: + raise ValueError(f"changed-set path escapes project root: {value}") from exc + return candidate + + +def normalize_changed_set( + project_root: str | Path, changed: list[str], deleted: list[str] +) -> tuple[list[Path], list[Path]]: + """Validate, resolve, sort, and deduplicate an external changed manifest.""" + if not isinstance(changed, list) or not all(isinstance(value, str) for value in changed): + raise ValueError("changed must be an array of strings") + if not isinstance(deleted, list) or not all(isinstance(value, str) for value in deleted): + raise ValueError("deleted must be an array of strings") + root = Path(project_root).expanduser().resolve() + if not root.is_dir(): + raise ValueError(f"project root does not exist: {root}") + changed_paths = sorted({_resolve_source(root, value) for value in changed}, key=str) + deleted_paths = sorted({_resolve_source(root, value) for value in deleted}, key=str) + overlap = set(changed_paths) & set(deleted_paths) + if overlap: + raise ValueError(f"paths cannot be both changed and deleted: {sorted(map(str, overlap))}") + missing_changed = [str(path) for path in changed_paths if not path.is_file()] + if missing_changed: + raise ValueError(f"changed paths must exist as files: {missing_changed}") + return changed_paths, deleted_paths + + +def _is_gitignored_source(root: Path, path: Path) -> bool: + """Apply root and nested gitignore rules to one explicit changed path.""" + matchers = [] + cache: dict[Path, object] = {} + current = root + directories = [root] + for part in path.relative_to(root).parts[:-1]: + current /= part + directories.append(current) + for directory in directories: + matcher = load_gitignore_matcher(directory, cache) + if matcher is not None: + matchers.append(matcher) + return bool(matchers and is_gitignored(path, matchers, is_dir=False)) + + +def sync_changed_sources( + *, + palace_path: str, + project_root: str | Path, + changed: list[str], + deleted: list[str], + wing: str | None = None, + agent: str = "mempalace", + dry_run: bool = True, +) -> ChangedSetReport: + """Serialize replacement of changed sources and removal of deleted sources.""" + root = Path(project_root).expanduser().resolve() + changed_paths, deleted_paths = normalize_changed_set(root, changed, deleted) + ignored_paths = [path for path in changed_paths if _is_gitignored_source(root, path)] + ignored_set = set(ignored_paths) + indexable_paths = [path for path in changed_paths if path not in ignored_set] + report: ChangedSetReport = { + "changed": len(changed_paths), + "deleted": len(deleted_paths), + "ignored": len(ignored_paths), + "reindexed": 0, + "drawers_added": 0, + "dry_run": dry_run, + } + if dry_run: + return report + + project_config = load_config(str(root)) + resolved_wing = wing or project_config["wing"] + rooms = project_config.get("rooms", [{"name": "general", "description": "All files"}]) + affected = [*ignored_paths, *deleted_paths] + with mine_palace_lock(palace_path): + drawers = get_collection(palace_path, create=False) + closets = get_closets_collection(palace_path, create=True) + for path in affected: + source = str(path) + drawers.delete(where={"$and": [{"source_file": source}, {"wing": resolved_wing}]}) + closets.delete(where={"$and": [{"source_file": source}, {"wing": resolved_wing}]}) + for path in indexable_paths: + added, _room, skip_reason = process_file( + path, + root, + drawers, + resolved_wing, + rooms, + agent, + False, + closets_col=closets, + force_reindex=True, + ) + if skip_reason is None and added > 0: + report["reindexed"] += 1 + report["drawers_added"] += added + return report + + +__all__ = ["ChangedSetReport", "normalize_changed_set", "sync_changed_sources"] diff --git a/mempalace/cli.py b/mempalace/cli.py index b28e3b56a..8d0bbd5aa 100644 --- a/mempalace/cli.py +++ b/mempalace/cli.py @@ -33,6 +33,7 @@ import argparse import contextlib +import json import os import shlex import sys @@ -1058,6 +1059,27 @@ def cmd_sync(args): """Prune drawers whose source files are gitignored, deleted, or moved (#1252).""" palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path + changed_set = None + manifest_path = getattr(args, "manifest", None) + if manifest_path: + if not args.dir: + print("mempalace: sync --manifest requires a project root argument", file=sys.stderr) + sys.exit(2) + try: + with open(os.path.expanduser(manifest_path), encoding="utf-8") as handle: + changed_set = json.load(handle) + if not isinstance(changed_set, dict): + raise ValueError("manifest must be a JSON object") + changed_set = { + "changed": changed_set.get("changed") or [], + "deleted": changed_set.get("deleted") or [], + } + if not all(isinstance(value, list) for value in changed_set.values()): + raise ValueError("manifest changed/deleted values must be arrays") + except (OSError, ValueError, json.JSONDecodeError) as exc: + print(f"mempalace: invalid changed-set manifest: {exc}", file=sys.stderr) + sys.exit(2) + if getattr(args, "background", False) and not getattr(args, "daemon", False): print("mempalace: --background requires --daemon", file=sys.stderr) sys.exit(2) @@ -1069,6 +1091,9 @@ def cmd_sync(args): "wing": args.wing, "dry_run": args.dry_run, } + if changed_set is not None: + payload["changed_set"] = changed_set + payload["agent"] = getattr(args, "agent", "mempalace") _submit_daemon_cli_job("sync", payload, args, background=getattr(args, "background", False)) return @@ -1077,6 +1102,7 @@ def cmd_sync(args): from .backends import detect_backend_for_path from .palace import _backend_artifact_label, resolve_backend_name from .sync import sync_palace + from .changed_set import sync_changed_sources if not os.path.isdir(palace_path): print(f"\n No palace found at {palace_path}") @@ -1116,6 +1142,21 @@ def cmd_sync(args): print(f"{'-' * 55}\n") try: + if changed_set is not None: + report = sync_changed_sources( + palace_path=palace_path, + project_root=project_dirs[0], + changed=changed_set["changed"], + deleted=changed_set["deleted"], + wing=args.wing, + agent=getattr(args, "agent", "mempalace"), + dry_run=args.dry_run, + ) + print( + f" Changed-set: changed={report['changed']} deleted={report['deleted']} " + f"reindexed={report['reindexed']} drawers_added={report['drawers_added']}" + ) + return report = sync_palace( palace_path=palace_path, project_dirs=project_dirs, @@ -2646,6 +2687,11 @@ def main(): help="Project root to sync (optional; auto-detects from drawer metadata)", ) p_sync.add_argument("--wing", default=None, help="Limit to one wing") + p_sync.add_argument( + "--manifest", + help="JSON changed-set with project-relative changed/deleted arrays; requires dir", + ) + p_sync.add_argument("--agent", default="mempalace", help="Agent recorded on reindexed drawers") p_sync.add_argument( "--root", action="append", diff --git a/mempalace/miner.py b/mempalace/miner.py index a12f00d2c..37255417c 100644 --- a/mempalace/miner.py +++ b/mempalace/miner.py @@ -59,11 +59,13 @@ def _path_within_root(path: Path, root: Path) -> bool: def _read_text_no_follow(filepath: Path, root: Path) -> Optional[tuple[str, float]]: - """Read ``filepath`` and return ``(content, mtime)`` from the SAME - ``fstat()`` call that validated the file, so callers never need a - separate, later ``os.path.getmtime()`` that could observe a file - modified in between (see #22: a stale re-stat lets appended content - be silently and permanently skipped).""" + """Read ``filepath`` and return ``(content, mtime)``. + + ``mtime`` comes from the SAME ``fstat()`` call that validated the file, + so callers never need a separate, later ``os.path.getmtime()`` that + could observe a file modified in between (see #22: a stale re-stat + lets appended content be silently and permanently skipped). + """ if not _path_within_root(filepath, root): return None # O_NONBLOCK is what makes the S_ISREG check below reachable. Opening a @@ -94,9 +96,11 @@ def _read_text_no_follow(filepath: Path, root: Path) -> Optional[tuple[str, floa if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_FILE_SIZE: return None mtime = st.st_mtime - with os.fdopen(fd, "r", encoding="utf-8", errors="replace") as f: + with os.fdopen(fd, "rb") as f: fd = -1 - return f.read(), mtime + raw_content = f.read() + content = raw_content.decode("utf-8", errors="replace") + return content, mtime except OSError: return None finally: @@ -1400,6 +1404,8 @@ def _build_drawer_metadata( line_end: Optional[int] = None, content_date: Optional[str] = None, chunk_total: Optional[int] = None, + authority_uri: Optional[str] = None, + authority_version: Optional[str] = None, ) -> dict: """Build the metadata dict for one drawer without upserting. @@ -1444,6 +1450,10 @@ def _build_drawer_metadata( metadata["content_date"] = content_date if chunk_total is not None: metadata["chunk_total"] = chunk_total + if authority_uri: + metadata["authority_uri"] = authority_uri + if authority_version: + metadata["authority_version"] = authority_version metadata["hall"] = detect_hall(content) entities = _extract_entities_for_metadata(content) if entities: @@ -1494,6 +1504,7 @@ def process_file( chunk_overlap: int = None, min_chunk_size: int = None, max_chunks_per_file: Optional[int] = None, + force_reindex: bool = False, ) -> tuple: """Read, chunk, route, and file one file. @@ -1508,14 +1519,21 @@ def process_file( # Skip if already filed source_file = str(filepath) - if not dry_run and file_already_mined(collection, source_file, check_mtime=True): + if ( + not dry_run + and not force_reindex + and file_already_mined(collection, source_file, check_mtime=True, wing=wing) + ): return 0, "general", None read_result = _read_text_no_follow(filepath, project_path) if read_result is None: return 0, "general", None - content, read_mtime = read_result - + if len(read_result) == 3: + content, read_mtime, authority_version = read_result + else: + content, read_mtime = read_result + authority_version = f"sha256:{hashlib.sha256(content.encode('utf-8')).hexdigest()}" content = content.strip() if len(content) < effective_min: return 0, "general", None @@ -1553,7 +1571,9 @@ def process_file( # both delete, and both insert — creating duplicates or losing data. with mine_lock(source_file): # Re-check after acquiring lock — another agent may have just finished - if file_already_mined(collection, source_file, check_mtime=True): + if not force_reindex and file_already_mined( + collection, source_file, check_mtime=True, wing=wing + ): return 0, room, None # Purge stale drawers for this file before re-inserting the fresh chunks. @@ -1570,7 +1590,7 @@ def process_file( # leaves the old drawers' stored mtime untouched, so the next mine # still sees a mismatch against the current on-disk mtime and retries. try: - collection.delete(where={"source_file": source_file}) + collection.delete(where={"$and": [{"source_file": source_file}, {"wing": wing}]}) except Exception as exc: print( f" ! [skip] {filepath.name[:50]:50} stale-drawer purge failed " @@ -1578,7 +1598,6 @@ def process_file( f"on the next mine", file=sys.stderr, ) - logger.debug("Stale-drawer purge failed for %s", source_file, exc_info=True) return 0, room, None # source_mtime is the mtime paired with the content actually read @@ -1595,6 +1614,7 @@ def process_file( # mtime hierarchy. Returns None when nothing usable found → caller # falls back to filed_at downstream. file_content_date = _extract_content_date(source_file, content) + authority_uri = filepath.expanduser().resolve().as_uri() drawers_added = 0 # Accumulate drawer metadata across batches so the closet emitter @@ -1627,6 +1647,8 @@ def process_file( line_end=chunk.get("line_end"), content_date=file_content_date, chunk_total=len(chunks), + authority_uri=authority_uri, + authority_version=authority_version, ) ) assert_no_collisions(list(zip(batch_ids, batch_metas)), collection) @@ -1692,6 +1714,8 @@ def process_file( "drawer_count": drawers_added, "filed_at": datetime.now().isoformat(), "normalize_version": NORMALIZE_VERSION, + "authority_uri": authority_uri, + "authority_version": authority_version, } if entities: closet_meta["entities"] = entities diff --git a/mempalace/palace.py b/mempalace/palace.py index 255d7518b..72e004e13 100644 --- a/mempalace/palace.py +++ b/mempalace/palace.py @@ -769,15 +769,18 @@ def _build_date_line_segment(drawer_metas): return f"{date_part}:L{line_start}-L{line_end}" -def purge_file_closets(closets_col, source_file: str) -> None: +def purge_file_closets(closets_col, source_file: str, wing: Optional[str] = None) -> None: """Delete every closet associated with ``source_file``. Call this before ``upsert_closet_lines`` on a re-mine so stale topics from a prior schema/version don't survive in the closet collection. Mirrors the drawer-purge step in process_file(). """ + where: dict[str, object] = {"source_file": source_file} + if wing: + where = {"$and": [{"source_file": source_file}, {"wing": wing}]} try: - closets_col.delete(where={"source_file": source_file}) + closets_col.delete(where=where) except Exception: logger.debug("Closet purge failed for %s", source_file, exc_info=True) @@ -1432,6 +1435,7 @@ def file_already_mined( source_file: str, check_mtime: bool = False, extract_mode: Optional[str] = None, + wing: Optional[str] = None, ) -> bool: """Check if a file has already been filed in the palace. @@ -1484,8 +1488,11 @@ def file_already_mined( # been seen so far toward that group's own chunk_total (#21). group_counts: dict = {} while True: + where: dict[str, object] = {"source_file": source_file} + if wing: + where = {"$and": [{"source_file": source_file}, {"wing": wing}]} results = collection.get( - where={"source_file": source_file}, + where=where, limit=1000, offset=offset, include=["metadatas"], diff --git a/mempalace/service.py b/mempalace/service.py index a54824730..f324c1732 100644 --- a/mempalace/service.py +++ b/mempalace/service.py @@ -337,9 +337,24 @@ def run_sync(payload: dict[str, Any]) -> dict[str, Any]: print(f"{'-' * 55}\n") try: + from .changed_set import sync_changed_sources from .sync import sync_palace from .wal import _wal_log + changed_set = payload.get("changed_set") + if changed_set is not None: + if not project_dirs: + raise ValueError("changed-set sync requires a project root") + report = sync_changed_sources( + palace_path=palace_path, + project_root=project_dirs[0], + changed=changed_set.get("changed") or [], + deleted=changed_set.get("deleted") or [], + wing=payload.get("wing"), + agent=payload.get("agent") or "mempalace", + dry_run=dry_run, + ) + return {"success": True, "report": report, "exit_code": 0} report = sync_palace( palace_path=palace_path, project_dirs=project_dirs, diff --git a/tests/test_changed_set.py b/tests/test_changed_set.py new file mode 100644 index 000000000..d7cff0c85 --- /dev/null +++ b/tests/test_changed_set.py @@ -0,0 +1,148 @@ +from pathlib import Path + +import pytest + +from mempalace.changed_set import normalize_changed_set, sync_changed_sources + + +def test_normalize_changed_set_rejects_escape_and_overlap(tmp_path): + (tmp_path / "a.py").write_text("x", encoding="utf-8") + with pytest.raises(ValueError, match="escapes project root"): + normalize_changed_set(tmp_path, ["../outside.py"], []) + with pytest.raises(ValueError, match="both changed and deleted"): + normalize_changed_set(tmp_path, ["a.py"], ["a.py"]) + + +def test_dry_run_does_not_open_palace(tmp_path): + (tmp_path / "a.py").write_text("x", encoding="utf-8") + report = sync_changed_sources( + palace_path="/not/opened", + project_root=tmp_path, + changed=["a.py"], + deleted=["old.py"], + ) + assert report == { + "changed": 1, + "deleted": 1, + "ignored": 0, + "reindexed": 0, + "drawers_added": 0, + "dry_run": True, + } + + +def test_apply_purges_affected_sources_and_indexes_only_changed(monkeypatch, tmp_path): + from mempalace import changed_set + + changed = tmp_path / "a.py" + changed.write_text("print('new')", encoding="utf-8") + calls = {"deleted": [], "closets": [], "processed": []} + + class FakeCollection: + def delete(self, *, where): + calls["deleted"].append(where) + + class Lock: + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + monkeypatch.setattr(changed_set, "mine_palace_lock", lambda _path: Lock()) + monkeypatch.setattr(changed_set, "get_collection", lambda *_a, **_k: FakeCollection()) + monkeypatch.setattr(changed_set, "get_closets_collection", lambda *_a, **_k: object()) + + class FakeClosets: + def delete(self, *, where): + calls["closets"].append(where) + + monkeypatch.setattr(changed_set, "get_closets_collection", lambda *_a, **_k: FakeClosets()) + monkeypatch.setattr( + changed_set, + "load_config", + lambda _root: {"wing": "project", "rooms": [{"name": "general"}]}, + ) + + def fake_process(path: Path, *_args, **_kwargs): + calls["processed"].append(path) + return 2, "general", None + + monkeypatch.setattr(changed_set, "process_file", fake_process) + report = sync_changed_sources( + palace_path="/palace", + project_root=tmp_path, + changed=["a.py"], + deleted=["old.py"], + dry_run=False, + ) + + assert calls["processed"] == [changed] + assert len(calls["deleted"]) == 1 + assert len(calls["closets"]) == 1 + assert report["drawers_added"] == 2 + assert report["reindexed"] == 1 + + +def test_explicit_changed_set_purges_gitignored_sources_without_reindex(monkeypatch, tmp_path): + from mempalace import changed_set + + generated = tmp_path / "generated" / "bundle.md" + generated.parent.mkdir() + generated.write_text("duplicated aggregate", encoding="utf-8") + (tmp_path / ".gitignore").write_text("generated/\n", encoding="utf-8") + calls = {"drawers": [], "closets": [], "processed": []} + + class FakeCollection: + def __init__(self, key): + self.key = key + + def delete(self, *, where): + calls[self.key].append(where) + + class Lock: + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + monkeypatch.setattr(changed_set, "mine_palace_lock", lambda _path: Lock()) + monkeypatch.setattr(changed_set, "get_collection", lambda *_a, **_k: FakeCollection("drawers")) + monkeypatch.setattr( + changed_set, + "get_closets_collection", + lambda *_a, **_k: FakeCollection("closets"), + ) + monkeypatch.setattr( + changed_set, + "load_config", + lambda _root: {"wing": "project", "rooms": [{"name": "general"}]}, + ) + monkeypatch.setattr( + changed_set, + "process_file", + lambda path, *_a, **_k: calls["processed"].append(path), + ) + + report = sync_changed_sources( + palace_path="/palace", + project_root=tmp_path, + changed=["generated/bundle.md"], + deleted=[], + dry_run=False, + ) + + assert report["ignored"] == 1 + assert report["reindexed"] == 0 + assert calls["processed"] == [] + assert len(calls["drawers"]) == 1 + assert len(calls["closets"]) == 1 + + +@pytest.mark.parametrize("field", ["changed", "deleted"]) +def test_normalize_changed_set_rejects_non_array_manifests(tmp_path, field): + values = {"changed": [], "deleted": []} + values[field] = "a.py" + with pytest.raises(ValueError, match=f"{field} must be an array of strings"): + normalize_changed_set(tmp_path, values["changed"], values["deleted"]) diff --git a/tests/test_closets.py b/tests/test_closets.py index 6f894629c..191256911 100644 --- a/tests/test_closets.py +++ b/tests/test_closets.py @@ -1519,6 +1519,7 @@ def test_hybrid_search_enrichment_populates_drawer_index_and_total(self, palace_ # The hybrid path promotes the closet-agreeing source to drawer+closet. boosted = [h for h in result["results"] if h["matched_via"] == "drawer+closet"] assert boosted, "hybrid search should mark the closet-agreeing source" + assert len(boosted) == 1, "identical hydrated windows must collapse to one result" top = boosted[0] assert top["total_drawers"] == 5 assert isinstance(top["drawer_index"], int) diff --git a/tests/test_daemon.py b/tests/test_daemon.py index bbf80cfea..aa19ae949 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -1379,6 +1379,38 @@ def _boom(*args, **kwargs): assert runtime.store.get(job.id).state == "running" +def test_run_sync_dispatches_changed_set_without_full_scan(tmp_path, monkeypatch): + import mempalace.changed_set as changed_set_module + from mempalace import service + + palace = tmp_path / "palace" + palace.mkdir() + (palace / "chroma.sqlite3").touch() + project = tmp_path / "project" + project.mkdir() + captured = {} + + def fake_sync(**kwargs): + captured.update(kwargs) + return {"changed": 1, "deleted": 1, "reindexed": 1, "drawers_added": 2, "dry_run": False} + + monkeypatch.setattr(changed_set_module, "sync_changed_sources", fake_sync) + result = service.run_sync( + { + "palace_path": str(palace), + "dir": str(project), + "wing": "demo", + "dry_run": False, + "changed_set": {"changed": ["a.py"], "deleted": ["old.py"]}, + } + ) + + assert result["success"] is True + assert captured["project_root"] == str(project) + assert captured["changed"] == ["a.py"] + assert captured["deleted"] == ["old.py"] + + # --- post-merge review follow-ups (Copilot review on #1826) --- diff --git a/tests/test_miner.py b/tests/test_miner.py index 037cd63d2..72d4253ea 100644 --- a/tests/test_miner.py +++ b/tests/test_miner.py @@ -1,3 +1,4 @@ +import hashlib import os import shlex import shutil @@ -1208,6 +1209,7 @@ def test_process_file_uses_bounded_upsert_batches(tmp_path, monkeypatch): class FakeCol: def __init__(self): self.batch_sizes = [] + self.metadatas = [] def get(self, *args, **kwargs): return {"ids": []} @@ -1217,6 +1219,7 @@ def delete(self, *args, **kwargs): def upsert(self, documents, ids, metadatas): self.batch_sizes.append(len(documents)) + self.metadatas.extend(metadatas) source = tmp_path / "src.py" source.write_text("print('hello')\n" * 20, encoding="utf-8") @@ -1241,6 +1244,83 @@ def upsert(self, documents, ids, metadatas): assert room == "general" assert skip_reason is None assert col.batch_sizes == [2, 2, 1] + expected_version = f"sha256:{hashlib.sha256(source.read_bytes()).hexdigest()}" + assert {meta["authority_uri"] for meta in col.metadatas} == {source.resolve().as_uri()} + assert {meta["authority_version"] for meta in col.metadatas} == {expected_version} + + +@pytest.mark.skipif(sys.platform == "win32", reason="symlink semantics require POSIX O_NOFOLLOW") +def test_process_file_force_reindex_does_not_follow_symlink(tmp_path): + """A changed-set reindex must retain the normal no-follow source boundary.""" + from mempalace import miner + + project = tmp_path / "project" + project.mkdir() + outside = tmp_path / "outside.md" + outside.write_text("secret content " * 20, encoding="utf-8") + linked_source = project / "linked.md" + linked_source.symlink_to(outside) + + added, room, reason = miner.process_file( + linked_source, + project, + object(), + "wing", + [{"name": "general", "description": "General"}], + "agent", + False, + force_reindex=True, + ) + + assert (added, room, reason) == (0, "general", None) + + +def test_process_file_reindex_is_scoped_to_target_wing(tmp_path, collection, monkeypatch): + from mempalace import miner + + source = tmp_path / "shared.md" + source.write_text("new canonical content " * 20, encoding="utf-8") + collection.upsert( + ids=["wing-a-old", "wing-b-preserved"], + documents=["old A", "stable B"], + metadatas=[ + { + "wing": "wing-a", + "room": "general", + "source_file": str(source), + "chunk_index": 0, + "normalize_version": NORMALIZE_VERSION, + "source_mtime": 0.0, + }, + { + "wing": "wing-b", + "room": "general", + "source_file": str(source), + "chunk_index": 0, + "normalize_version": NORMALIZE_VERSION, + "source_mtime": source.stat().st_mtime, + }, + ], + ) + monkeypatch.setattr(miner, "detect_hall", lambda _content: "documentation") + monkeypatch.setattr(miner, "_extract_entities_for_metadata", lambda _content: "") + + added, _room, reason = miner.process_file( + source, + tmp_path, + collection, + "wing-a", + [{"name": "general", "description": "General"}], + "agent", + False, + force_reindex=True, + ) + + assert reason is None + assert added > 0 + preserved = collection.get(ids=["wing-b-preserved"], include=["documents", "metadatas"]) + assert preserved["documents"] == ["stable B"] + assert preserved["metadatas"][0]["wing"] == "wing-b" def test_process_file_stamps_chunk_total_for_completion_check(tmp_path, monkeypatch): @@ -1540,7 +1620,14 @@ def get(self, where=None, limit=None, offset=0, include=None): } def delete(self, where=None): - source_file = where.get("source_file") if where else None + source_file = None + if where: + if "source_file" in where: + source_file = where["source_file"] + elif "$and" in where: + for clause in where["$and"]: + if "source_file" in clause: + source_file = clause["source_file"] self.deleted_sources.append(source_file) self.records = [ record From ed24af32284419b89c31477d2731626bdbd2ef4a Mon Sep 17 00:00:00 2001 From: Oleksii Pylypchuk Date: Mon, 13 Jul 2026 21:06:56 +0300 Subject: [PATCH 3/4] feat(memory): add explicit decision supersession Let callers replace a decision through an exact decision key and predecessor ID while retaining the original verbatim drawer as queryable history. Similarity never performs implicit replacement, and normal search excludes superseded entries by default. --- docs/decision-memory.md | 28 ++++++ mempalace/mcp_server.py | 151 ++++++++++++++++++++++++++++++--- mempalace/service.py | 1 + tests/test_mcp_server.py | 110 ++++++++++++++++++++++++ website/reference/mcp-tools.md | 35 +++++++- 5 files changed, 310 insertions(+), 15 deletions(-) create mode 100644 docs/decision-memory.md diff --git a/docs/decision-memory.md b/docs/decision-memory.md new file mode 100644 index 000000000..0c07acb1b --- /dev/null +++ b/docs/decision-memory.md @@ -0,0 +1,28 @@ +# Authority-aware decision memory PoC + +Decision drawers may carry an optional structured envelope while their content +remains verbatim: + +- `decision_key`: stable logical identity across versions; +- `authority_uri`: canonical local file path or `file://` URI; +- `authority_version`: `sha256:` or `mtime_ns:`; +- `memory_kind`: for example `decision`, `finding`, or `preference`; +- `authority_status`: `current`, `stale`, `unverified`, or `superseded`. + +`mempalace_search(verify_authority=true)` compares supported local authority +tokens and includes an authority envelope on every result. Verification is +opt-in because hashing large files has a real read-path cost. Unsupported and +legacy authorities remain `unverified`; they are never assumed current. + +`mempalace_supersede_drawer` requires both the predecessor ID and the exact same +non-empty `decision_key`. It files a new verbatim drawer, then marks the old +drawer `superseded` with `superseded_by=`. Default MCP search hides that +history; `include_superseded=true` exposes it. No semantic-similarity threshold +can supersede a decision implicitly. + +Checkpoint items accept the same fields plus `supersedes_id`, so an agent can +save a reviewed decision transition in one call. + +This PoC resolves only local files. Production authority adapters could support +Git blobs, GitHub issues, planners, or document systems without changing the +drawer lifecycle contract. diff --git a/mempalace/mcp_server.py b/mempalace/mcp_server.py index 5ba5e7d8a..402689fb4 100644 --- a/mempalace/mcp_server.py +++ b/mempalace/mcp_server.py @@ -420,6 +420,7 @@ def _parse_args(): "mempalace_delete_drawer", "mempalace_checkpoint", "mempalace_delete_by_source", + "mempalace_supersede_drawer", "mempalace_mine", "mempalace_sync", "mempalace_update_drawer", @@ -2374,6 +2375,7 @@ def tool_search( min_similarity: float = None, context: str = None, verify_authority: bool = False, + include_superseded: bool = False, ): limit = max(1, min(limit, _MAX_RESULTS)) try: @@ -2446,6 +2448,13 @@ def tool_search( } if context: result["context_received"] = True + if not include_superseded and isinstance(result.get("results"), list): + result["results"] = [ + hit + for hit in result["results"] + if (hit.get("authority") or {}).get("status") != "superseded" + ] + result["count"] = len(result["results"]) return result @@ -2926,6 +2935,7 @@ def tool_add_drawer( authority_uri: str = None, authority_version: str = None, memory_kind: str = None, + decision_key: str = None, ): """File verbatim content into a wing/room. Checks for duplicates first. @@ -2950,6 +2960,7 @@ def tool_add_drawer( authority_uri = strip_lone_surrogates(authority_uri or "") authority_version = strip_lone_surrogates(authority_version or "") memory_kind = sanitize_name(memory_kind, "memory_kind") if memory_kind else "" + decision_key = strip_lone_surrogates(decision_key or "").strip() except ValueError as e: return {"success": False, "error": str(e)} @@ -2986,6 +2997,9 @@ def tool_add_drawer( base_meta["authority_version"] = authority_version if memory_kind: base_meta["memory_kind"] = memory_kind + if decision_key: + base_meta["decision_key"] = decision_key + base_meta["authority_status"] = "current" # Idempotency. Three cases to detect a prior committed write: # (a) Single-doc path: drawer_id row exists (the only id used). @@ -3071,6 +3085,76 @@ def tool_add_drawer( return {"success": False, "error": str(e)} +def tool_supersede_drawer( + supersedes_id: str, + decision_key: str, + wing: str, + room: str, + content: str, + source_file: str = None, + added_by: str = "mcp", + authority_uri: str = None, + authority_version: str = None, + memory_kind: str = "decision", +): + """Add a new decision drawer and explicitly mark its predecessor superseded.""" + global _metadata_cache + col = _get_collection() + if not col: + return _collection_error_or_no_palace() + old = _logical_drawer_record(col, supersedes_id) + if old is None: + return {"success": False, "error": f"Drawer not found: {supersedes_id}"} + old_key = str((old.get("metadata") or {}).get("decision_key") or "") + if not decision_key or old_key != decision_key: + return { + "success": False, + "error": "supersession requires the same non-empty decision_key as the predecessor", + } + + added = tool_add_drawer( + wing=wing, + room=room, + content=content, + source_file=source_file, + added_by=added_by, + authority_uri=authority_uri, + authority_version=authority_version, + memory_kind=memory_kind, + decision_key=decision_key, + ) + if not added.get("success"): + return added + replacement_id = added["drawer_id"] + if replacement_id == supersedes_id: + return {"success": False, "error": "replacement content resolves to predecessor drawer id"} + + updated_metas = [] + for metadata in old["metadatas"]: + updated_metas.append( + { + **_safe_meta(metadata), + "authority_status": "superseded", + "superseded_by": replacement_id, + } + ) + col.upsert(ids=old["ids"], documents=old["documents"], metadatas=updated_metas) + _wal_log( + "supersede_drawer", + { + "drawer_id": supersedes_id, + "superseded_by": replacement_id, + "decision_key": decision_key, + }, + ) + _metadata_cache = None + return { + **added, + "supersedes_id": supersedes_id, + "decision_key": decision_key, + } + + def tool_delete_drawer(drawer_id: str): """Delete a single logical drawer by ID.""" global _metadata_cache @@ -4438,23 +4522,34 @@ def tool_checkpoint(items, diary=None, dedup_threshold=0.9, added_by=None): {"item": item, "error": "wing, room, content must be non-empty strings"} ) continue - dup = tool_check_duplicate(content, threshold=dedup_threshold) - if dup.get("is_duplicate"): - out["duplicates"].append({"room": room, "matches": dup.get("matches", [])}) - continue + # Explicit version transitions must not be blocked by semantic dedup: + # replacement decisions are often intentionally similar to predecessors. + if not item.get("supersedes_id"): + dup = tool_check_duplicate(content, threshold=dedup_threshold) + if dup.get("is_duplicate"): + out["duplicates"].append({"room": room, "matches": dup.get("matches", [])}) + continue # On a dedup error (genuine index failure — content is guaranteed a # string by the guard above) we still file rather than drop the # memory: verbatim recall is the priority and add_drawer's own # idempotency blocks exact duplicates. - res = tool_add_drawer( - wing=wing, - room=room, - content=content, - added_by=resolved_added_by, - authority_uri=item.get("authority_uri"), - authority_version=item.get("authority_version"), - memory_kind=item.get("memory_kind"), - ) + common = { + "wing": wing, + "room": room, + "content": content, + "added_by": resolved_added_by, + "authority_uri": item.get("authority_uri"), + "authority_version": item.get("authority_version"), + "memory_kind": item.get("memory_kind"), + "decision_key": item.get("decision_key"), + } + if item.get("supersedes_id"): + res = tool_supersede_drawer( + supersedes_id=item["supersedes_id"], + **common, + ) + else: + res = tool_add_drawer(**common) if res.get("success"): out["added"].append(res) else: @@ -5019,6 +5114,10 @@ def tool_patch_submit( "type": "boolean", "description": "Verify file:// or absolute-path authority version tokens. Adds current/stale/unverified status to each result.", }, + "include_superseded": { + "type": "boolean", + "description": "Include explicitly superseded decision drawers (default false)", + }, }, "required": ["query"], }, @@ -5067,6 +5166,10 @@ def tool_patch_submit( "type": "string", "description": "Memory class such as decision, finding, or preference (optional)", }, + "decision_key": { + "type": "string", + "description": "Stable logical key for a versioned decision (optional)", + }, }, "required": ["wing", "room", "content"], }, @@ -5095,6 +5198,8 @@ def tool_patch_submit( "authority_uri": {"type": "string"}, "authority_version": {"type": "string"}, "memory_kind": {"type": "string"}, + "decision_key": {"type": "string"}, + "supersedes_id": {"type": "string"}, }, "required": ["wing", "room", "content"], }, @@ -5125,6 +5230,26 @@ def tool_patch_submit( }, "handler": tool_checkpoint, }, + "mempalace_supersede_drawer": { + "description": "Create a replacement decision and explicitly preserve its predecessor as superseded history.", + "input_schema": { + "type": "object", + "properties": { + "supersedes_id": {"type": "string"}, + "decision_key": {"type": "string"}, + "wing": {"type": "string"}, + "room": {"type": "string"}, + "content": {"type": "string"}, + "source_file": {"type": "string"}, + "added_by": {"type": "string"}, + "authority_uri": {"type": "string"}, + "authority_version": {"type": "string"}, + "memory_kind": {"type": "string"}, + }, + "required": ["supersedes_id", "decision_key", "wing", "room", "content"], + }, + "handler": tool_supersede_drawer, + }, "mempalace_delete_drawer": { "description": "Delete a drawer by ID. Irreversible.", "input_schema": { diff --git a/mempalace/service.py b/mempalace/service.py index f324c1732..d7cb6ef6b 100644 --- a/mempalace/service.py +++ b/mempalace/service.py @@ -88,6 +88,7 @@ def _wrapped(*args, **kwargs): "mempalace_checkpoint", "mempalace_delete_by_source", "mempalace_delete_drawer", + "mempalace_supersede_drawer", "mempalace_update_drawer", "mempalace_diary_write", "mempalace_kg_add", diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index dcd55f6c5..808fad438 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -2205,6 +2205,116 @@ def _add(**kwargs): assert filed["authority_version"] == "sha256:abc" assert filed["memory_kind"] == "decision" + def test_supersede_drawer_marks_predecessor_without_deleting_it(self, monkeypatch): + from mempalace import mcp_server + + writes = [] + + class FakeCollection: + def upsert(self, **kwargs): + writes.append(kwargs) + + old = { + "ids": ["old"], + "documents": ["Use SQLite."], + "metadatas": [{"decision_key": "db/backend", "authority_status": "current"}], + "metadata": {"decision_key": "db/backend", "authority_status": "current"}, + } + monkeypatch.setattr(mcp_server, "_get_collection", lambda: FakeCollection()) + monkeypatch.setattr(mcp_server, "_logical_drawer_record", lambda _col, _id: old) + monkeypatch.setattr( + mcp_server, + "tool_add_drawer", + lambda **_kwargs: {"success": True, "drawer_id": "new", "chunks": 1}, + ) + monkeypatch.setattr(mcp_server, "_wal_log", lambda *_a, **_k: None) + + result = mcp_server.tool_supersede_drawer( + supersedes_id="old", + decision_key="db/backend", + wing="project", + room="decisions", + content="Use PostgreSQL.", + ) + + assert result["success"] is True + assert result["supersedes_id"] == "old" + assert writes[0]["ids"] == ["old"] + assert writes[0]["documents"] == ["Use SQLite."] + assert writes[0]["metadatas"][0]["authority_status"] == "superseded" + assert writes[0]["metadatas"][0]["superseded_by"] == "new" + + def test_supersede_drawer_rejects_decision_key_mismatch(self, monkeypatch): + from mempalace import mcp_server + + monkeypatch.setattr(mcp_server, "_get_collection", lambda: object()) + monkeypatch.setattr( + mcp_server, + "_logical_drawer_record", + lambda *_a: {"metadata": {"decision_key": "other"}}, + ) + result = mcp_server.tool_supersede_drawer( + supersedes_id="old", + decision_key="db/backend", + wing="project", + room="decisions", + content="Use PostgreSQL.", + ) + assert result["success"] is False + assert "same non-empty decision_key" in result["error"] + + def test_search_hides_superseded_by_default(self, monkeypatch): + from mempalace import mcp_server + + monkeypatch.setattr(mcp_server, "_refresh_vector_disabled_flag", lambda: None) + monkeypatch.setattr( + mcp_server, + "search_memories", + lambda *_a, **_k: { + "results": [ + {"text": "old", "authority": {"status": "superseded"}}, + {"text": "new", "authority": {"status": "current"}}, + ], + "count": 2, + }, + ) + result = mcp_server.tool_search("database") + history = mcp_server.tool_search("database", include_superseded=True) + assert [hit["text"] for hit in result["results"]] == ["new"] + assert result["count"] == 1 + assert history["count"] == 2 + + def test_checkpoint_supersession_bypasses_semantic_dedup(self, monkeypatch): + from mempalace import mcp_server + + monkeypatch.setattr( + mcp_server, + "tool_check_duplicate", + lambda *_a, **_k: (_ for _ in ()).throw(AssertionError("dedup must be bypassed")), + ) + captured = {} + + def fake_supersede(**kwargs): + captured.update(kwargs) + return {"success": True, "drawer_id": "new"} + + monkeypatch.setattr(mcp_server, "tool_supersede_drawer", fake_supersede) + result = mcp_server.tool_checkpoint( + items=[ + { + "wing": "w", + "room": "decisions", + "content": "Use PostgreSQL instead of SQLite.", + "decision_key": "db/backend", + "supersedes_id": "old", + } + ], + added_by="planner-agent", + ) + assert result["errors"] == [] + assert captured["supersedes_id"] == "old" + assert captured["added_by"] == "planner-agent" + def test_checkpoint_skips_semantic_duplicates(self, monkeypatch, config, kg): from mempalace import mcp_server diff --git a/website/reference/mcp-tools.md b/website/reference/mcp-tools.md index f6234ab96..7306f7867 100644 --- a/website/reference/mcp-tools.md +++ b/website/reference/mcp-tools.md @@ -58,8 +58,11 @@ Semantic search. Returns verbatim drawer content with similarity scores. | `limit` | integer | No | Max results (default: 5) | | `wing` | string | No | Filter by wing | | `room` | string | No | Filter by room | +| `source_file` | string | No | Exact full source path filter | +| `verify_authority` | boolean | No | Resolve supported local authority version and return current/stale/unverified status | +| `include_superseded` | boolean | No | Include explicitly superseded decision history (default false) | -**Returns:** `{ query, filters, results: [{ text, wing, room, source_file, similarity }] }` +**Returns:** `{ query, filters, results: [{ text, wing, room, source_file, similarity, authority }] }` --- @@ -101,6 +104,10 @@ File verbatim content into the palace. Identical content (same deterministic dra | `content` | string | **Yes** | Verbatim content to store | | `source_file` | string | No | Where this came from | | `added_by` | string | No | Who is filing (default: "mcp") | +| `authority_uri` | string | No | Canonical absolute local path or `file://` URI | +| `authority_version` | string | No | `sha256:` or `mtime_ns:` token | +| `memory_kind` | string | No | Memory class such as `decision` or `finding` | +| `decision_key` | string | No | Stable logical key for a versioned decision | **Returns:** `{ success, drawer_id, wing, room }` @@ -112,7 +119,7 @@ Save a whole session in one call. Semantic-dedups each item, files the non-dupli | Parameter | Type | Required | Description | |-----------|------|----------|-------------| -| `items` | array | **Yes** | Verbatim items to file. Each is `{ wing, room, content }` | +| `items` | array | **Yes** | Verbatim items to file. Supports authority fields plus `decision_key` and explicit `supersedes_id` | | `diary` | object | No | Diary entry written after filing: `{ agent_name, entry, topic?, wing? }` (`entry` is AAAK-format) | | `dedup_threshold` | number | No | Similarity threshold 0–1 for the per-item dedup check (default 0.9) | | `added_by` | string | No | Who is filing these drawers. An explicit value takes precedence; otherwise the diary `agent_name`, else `checkpoint` | @@ -121,6 +128,30 @@ Save a whole session in one call. Semantic-dedups each item, files the non-dupli --- +### `mempalace_supersede_drawer` + +Create a replacement decision while preserving its predecessor as explicit +history. The predecessor must carry the same non-empty `decision_key`; semantic +similarity alone never triggers supersession. + +| Parameter | Type | Required | Description | +|-----------|------|----------|-------------| +| `supersedes_id` | string | **Yes** | Logical drawer ID of the predecessor | +| `decision_key` | string | **Yes** | Stable decision identity; must match predecessor | +| `wing` | string | **Yes** | Wing for the replacement | +| `room` | string | **Yes** | Room for the replacement | +| `content` | string | **Yes** | New verbatim decision content | +| `authority_uri` | string | No | Canonical local authority | +| `authority_version` | string | No | Authority version token | + +The predecessor receives `authority_status=superseded` and +`superseded_by=`. Default MCP search hides it; +`include_superseded=true` returns the historical record. + +**Returns:** `{ success, drawer_id, supersedes_id, decision_key, ... }` + +--- + ### `mempalace_delete_drawer` Delete a drawer by ID. Irreversible. From fb83ba32869268020cec10df438ab5df8b6f338d Mon Sep 17 00:00:00 2001 From: Oleksii Pylypchuk Date: Mon, 27 Jul 2026 12:09:26 +0300 Subject: [PATCH 4/4] feat(daemon): serialize MCP writes and expose job diagnostics Route mutating MCP operations through one durable daemon writer instead of process-lifetime peer locks. Add leased, coalesced jobs with bounded wait responses and safe status diagnostics so concurrent clients can submit writes without exposing queued payloads. --- mempalace/daemon.py | 284 ++++++++++++++++-- mempalace/mcp_server.py | 524 ++++++++++++++++++++++++++++++--- mempalace/service.py | 33 ++- tests/test_daemon.py | 157 +++++++++- tests/test_mcp_server.py | 267 +++++++++++++++++ website/reference/mcp-tools.md | 24 +- 6 files changed, 1216 insertions(+), 73 deletions(-) diff --git a/mempalace/daemon.py b/mempalace/daemon.py index 66d104e90..f65c5b0a5 100644 --- a/mempalace/daemon.py +++ b/mempalace/daemon.py @@ -79,6 +79,10 @@ def _lock_defer_backoff_seconds() -> float: # Override via env for operators; tests patch the module attribute directly. LOCK_DEFER_BACKOFF_SECONDS = _lock_defer_backoff_seconds() + + +JOB_LEASE_SECONDS = 30.0 +JOB_HEARTBEAT_SECONDS = 5.0 MAX_BODY_BYTES = 1 << 20 # 1 MiB cap on request bodies (auth-gated DoS guard) SHUTDOWN_DRAIN_SECONDS = 10.0 # Terminal jobs are kept for diagnostics then pruned so the queue DB (which @@ -113,6 +117,10 @@ def _now() -> str: return datetime.now(timezone.utc).isoformat() +def _lease_deadline() -> str: + return (datetime.now(timezone.utc) + timedelta(seconds=JOB_LEASE_SECONDS)).isoformat() + + def canonical_palace_path(path: str | None = None) -> str: value = path or MempalaceConfig().palace_path return os.path.abspath(os.path.realpath(os.path.expanduser(value))) @@ -256,6 +264,7 @@ class Job: result: dict[str, Any] | None error: dict[str, Any] | None attempts: int + lease_token: str | None class QueueStore: @@ -309,6 +318,17 @@ def _init_db(self) -> None: ) conn.execute("CREATE INDEX IF NOT EXISTS idx_jobs_state ON jobs(state, priority)") conn.execute("CREATE INDEX IF NOT EXISTS idx_jobs_dedupe ON jobs(dedupe_key, state)") + columns = {row[1] for row in conn.execute("PRAGMA table_info(jobs)").fetchall()} + for name, ddl in ( + ("coalesce_key", "TEXT"), + ("source_version", "INTEGER"), + ("tombstone", "INTEGER NOT NULL DEFAULT 0"), + ("heartbeat_at", "TEXT"), + ("lease_expires_at", "TEXT"), + ("lease_token", "TEXT"), + ): + if name not in columns: + conn.execute(f"ALTER TABLE jobs ADD COLUMN {name} {ddl}") # Unique partial index: at most one queued/running job per dedupe_key. # Enforces the dedupe invariant across processes (TOCTOU-safe); finished # jobs drop out of the index so a later identical enqueue is allowed. @@ -316,6 +336,13 @@ def _init_db(self) -> None: "CREATE UNIQUE INDEX IF NOT EXISTS idx_jobs_dedupe_active " "ON jobs(dedupe_key) WHERE state IN ('queued', 'running')" ) + # A running file event may already have been read by the worker, so + # only queued successors coalesce. This permits exactly one newer + # generation to wait behind an in-flight generation. + conn.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_jobs_coalesce_queued " + "ON jobs(coalesce_key) WHERE state = 'queued' AND coalesce_key IS NOT NULL" + ) # The queue DB holds verbatim payloads (diary text, source paths) — lock it # down to owner-only regardless of the invoking user's umask. The WAL/SHM # sidecars carry the same un-checkpointed payloads, so harden them too when @@ -381,13 +408,58 @@ def recover_running(self) -> int: cur = conn.execute( """ UPDATE jobs - SET state = 'queued', started_at = NULL + SET state = 'queued', started_at = NULL, heartbeat_at = NULL, + lease_expires_at = NULL, lease_token = NULL WHERE state = 'running' AND attempts < ? """, (MAX_ATTEMPTS,), ) return int(cur.rowcount or 0) + def recover_expired_leases(self) -> int: + """Requeue retryable jobs whose worker stopped renewing its lease.""" + + now = _now() + with self._lock, self._connect() as conn: + conn.execute( + """ + UPDATE jobs + SET state = 'failed', finished_at = ?, lease_expires_at = NULL, + error_json = COALESCE(error_json, ?) + WHERE state = 'running' AND lease_expires_at < ? AND attempts >= ? + """, + ( + now, + json.dumps( + {"error_class": "LeaseExpired", "message": "job lease expired"}, + ensure_ascii=False, + ), + now, + MAX_ATTEMPTS, + ), + ) + cur = conn.execute( + """ + UPDATE jobs + SET state = 'queued', started_at = NULL, heartbeat_at = NULL, + lease_expires_at = NULL, lease_token = NULL + WHERE state = 'running' AND lease_expires_at < ? AND attempts < ? + """, + (now, MAX_ATTEMPTS), + ) + return int(cur.rowcount or 0) + + def heartbeat(self, job_id: str, lease_token: str) -> bool: + with self._lock, self._connect() as conn: + cur = conn.execute( + """ + UPDATE jobs SET heartbeat_at = ?, lease_expires_at = ? + WHERE id = ? AND state = 'running' AND lease_token = ? + """, + (_now(), _lease_deadline(), job_id, lease_token), + ) + return cur.rowcount == 1 + def enqueue( self, kind: str, @@ -395,7 +467,19 @@ def enqueue( *, dedupe_key: str | None = None, priority: int = 0, + coalesce_key: str | None = None, + source_version: int | None = None, + tombstone: bool = False, ) -> Job: + if coalesce_key: + return self.enqueue_coalesced( + kind, + payload, + coalesce_key=coalesce_key, + source_version=int(source_version or 0), + tombstone=tombstone, + priority=priority, + ) payload_json = json.dumps(payload, ensure_ascii=False, sort_keys=True) with self._lock, self._connect() as conn: if dedupe_key: @@ -451,7 +535,80 @@ def enqueue( row = conn.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone() return self._row_to_job(row) + def enqueue_coalesced( + self, + kind: str, + payload: dict[str, Any], + *, + coalesce_key: str, + source_version: int, + tombstone: bool, + priority: int = 0, + ) -> Job: + """Atomically retain the newest queued event for one canonical file. + + A delete wins ties so a stale changed event cannot resurrect a file. + A newer version may replace a tombstone, covering delete-then-recreate. + Running work is immutable; a single queued successor is coalesced behind it. + """ + + payload_json = json.dumps(payload, ensure_ascii=False, sort_keys=True) + with self._lock, self._connect() as conn: + row = conn.execute( + "SELECT * FROM jobs WHERE coalesce_key = ? AND state = 'queued' LIMIT 1", + (coalesce_key,), + ).fetchone() + if row is not None: + old_version = int(row["source_version"] or 0) + old_tombstone = bool(row["tombstone"]) + replace = source_version > old_version or ( + source_version == old_version and tombstone and not old_tombstone + ) + if replace: + conn.execute( + """ + UPDATE jobs + SET kind = ?, payload_json = ?, priority = ?, source_version = ?, + tombstone = ?, created_at = ? + WHERE id = ? AND state = 'queued' + """, + ( + kind, + payload_json, + int(priority), + int(source_version), + int(tombstone), + _now(), + row["id"], + ), + ) + current = conn.execute("SELECT * FROM jobs WHERE id = ?", (row["id"],)).fetchone() + return self._row_to_job(current) + + job_id = uuid.uuid4().hex + conn.execute( + """ + INSERT INTO jobs ( + id, kind, payload_json, state, priority, created_at, attempts, + coalesce_key, source_version, tombstone + ) VALUES (?, ?, ?, 'queued', ?, ?, 0, ?, ?, ?) + """, + ( + job_id, + kind, + payload_json, + int(priority), + _now(), + coalesce_key, + int(source_version), + int(tombstone), + ), + ) + row = conn.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone() + return self._row_to_job(row) + def claim_next(self, *, exclude: set[str] | None = None) -> Job | None: + # Atomic across processes: the UPDATE only fires if the row is still # 'queued'. If two daemon processes SELECT the same row, the first to # UPDATE it flips state to 'running' (rowcount=1); the second's UPDATE @@ -499,14 +656,16 @@ def claim_next(self, *, exclude: set[str] | None = None) -> Job | None: job_id = next((r["id"] for r in candidates if r["id"] not in exclude), None) if job_id is None: return None + lease_token = uuid.uuid4().hex cur = conn.execute( """ UPDATE jobs - SET state = 'running', started_at = ?, attempts = attempts + 1, + SET state = 'running', started_at = ?, heartbeat_at = ?, + lease_expires_at = ?, lease_token = ?, attempts = attempts + 1, error_json = NULL WHERE id = ? AND state = 'queued' """, - (_now(), job_id), + (_now(), _now(), _lease_deadline(), lease_token, job_id), ) if cur.rowcount != 1: # Lost the race to another process — nothing to run this iteration. @@ -522,6 +681,7 @@ def finish( result: dict[str, Any] | None = None, error: dict[str, Any] | None = None, only_if_running: bool = False, + lease_token: str | None = None, ) -> Job: # ``only_if_running`` guards the worker's finish against a lost race with # shutdown's cancel: if the active job was already flipped to 'cancelled' @@ -529,21 +689,27 @@ def finish( # 'succeeded'/'failed' (which would un-cancel a job recover_running must # not re-run). The conditional UPDATE makes the worker's finish a no-op in # that window instead of relying on process-exit timing. - where = "WHERE id = ?" + (" AND state = 'running'" if only_if_running else "") + where = "WHERE id = ?" + if only_if_running: + where += " AND state = 'running' AND lease_token = ?" with self._lock, self._connect() as conn: + params = [ + state, + _now(), + json.dumps(result or {}, ensure_ascii=False), + json.dumps(error or {}, ensure_ascii=False) if error else None, + job_id, + ] + if only_if_running: + params.append(lease_token) conn.execute( f""" UPDATE jobs - SET state = ?, finished_at = ?, result_json = ?, error_json = ? + SET state = ?, finished_at = ?, result_json = ?, error_json = ?, + heartbeat_at = NULL, lease_expires_at = NULL, lease_token = NULL {where} """, - ( - state, - _now(), - json.dumps(result or {}, ensure_ascii=False), - json.dumps(error or {}, ensure_ascii=False) if error else None, - job_id, - ), + params, ) row = conn.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone() return self._row_to_job(row) @@ -641,6 +807,7 @@ def _loads(value): result=_loads(row["result_json"]), error=_loads(row["error_json"]), attempts=int(row["attempts"]), + lease_token=row["lease_token"], ) @@ -689,6 +856,7 @@ def __init__(self, palace_path: str, backend: str | None = None): self.shutdown_event = threading.Event() self.worker_wake = threading.Event() self.active_job_id: str | None = None + self.active_lease_token: str | None = None self.worker_thread: threading.Thread | None = None # job_id -> monotonic deadline before which a lock-refused job is not # re-claimed (#2014). Worker-owned; only _worker_loop touches it. @@ -713,11 +881,26 @@ def start_worker(self) -> threading.Thread: def worker_alive(self) -> bool: return self.worker_thread is not None and self.worker_thread.is_alive() - def _safe_finish(self, job_id: str, *, state: str, result: dict, error: dict | None) -> None: + def _safe_finish( + self, + job_id: str, + lease_token: str, + *, + state: str, + result: dict, + error: dict | None, + ) -> None: try: # only_if_running: if shutdown already cancelled this job, don't # resurrect it. A finish failure must not kill the worker regardless. - self.store.finish(job_id, state=state, result=result, error=error, only_if_running=True) + self.store.finish( + job_id, + state=state, + result=result, + error=error, + only_if_running=True, + lease_token=lease_token, + ) except Exception: # noqa: BLE001 - a finish failure must not kill the worker pass @@ -750,6 +933,7 @@ def _worker_loop(self) -> None: from .service import execute_job while not self.shutdown_event.is_set(): + self.store.recover_expired_leases() try: job = self.store.claim_next(exclude=self._cooling_job_ids()) except Exception: # noqa: BLE001 - sqlite/disk errors must not kill the worker @@ -760,6 +944,20 @@ def _worker_loop(self) -> None: self.worker_wake.clear() continue self.active_job_id = job.id + self.active_lease_token = job.lease_token + heartbeat_stop = threading.Event() + + def _heartbeat() -> None: + while not heartbeat_stop.wait(JOB_HEARTBEAT_SECONDS): + if not self.store.heartbeat(job.id, job.lease_token or ""): + return + + heartbeat_thread = threading.Thread( + target=_heartbeat, + name=f"mempalace-job-heartbeat-{job.id[:8]}", + daemon=True, + ) + heartbeat_thread.start() try: payload = dict(job.payload) # Override, never trust the client: an authenticated request for @@ -797,7 +995,9 @@ def _worker_loop(self) -> None: continue state = "succeeded" if ok else "failed" error = None if ok else {"message": result.get("error", "job failed")} - self._safe_finish(job.id, state=state, result=result, error=error) + self._safe_finish( + job.id, job.lease_token or "", state=state, result=result, error=error + ) except (Exception, SystemExit) as exc: # SystemExit is BaseException, not Exception — catching it here is # deliberate. Without it, a sys.exit() in a dependency would slip @@ -807,23 +1007,41 @@ def _worker_loop(self) -> None: # same BaseException-slip-past semantics, documented in comments.) self._safe_finish( job.id, + job.lease_token or "", state="failed", result={"success": False, "exit_code": 1}, error={"error_class": type(exc).__name__, "message": str(exc)}, ) finally: + heartbeat_stop.set() + heartbeat_thread.join(timeout=1.0) self.active_job_id = None + self.active_lease_token = None + +def _json_response(handler: BaseHTTPRequestHandler, status: int, payload: dict[str, Any]) -> bool: + """Write one JSON response, treating a disconnected client as normal. + + Poll clients use short deadlines and can disappear after the server has + prepared a response. Propagating EPIPE/connection-reset into ``do_GET`` + makes its generic error handler attempt a second response on the same dead + socket, producing a second traceback. Consume only disconnect errors here; + all real response-generation failures still propagate. + """ -def _json_response(handler: BaseHTTPRequestHandler, status: int, payload: dict[str, Any]) -> None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") - handler.send_response(status) - handler.send_header("Content-Type", "application/json; charset=utf-8") - handler.send_header("Content-Length", str(len(body))) - handler.send_header("Connection", "close") - handler.end_headers() - handler.wfile.write(body) - handler.close_connection = True + try: + handler.send_response(status) + handler.send_header("Content-Type", "application/json; charset=utf-8") + handler.send_header("Content-Length", str(len(body))) + handler.send_header("Connection", "close") + handler.end_headers() + handler.wfile.write(body) + return True + except (BrokenPipeError, ConnectionResetError): + return False + finally: + handler.close_connection = True def _restore_server_process_state(previous_env: dict[str, str | None], previous_umask: int) -> None: @@ -1006,6 +1224,9 @@ def do_POST(self): body.get("payload") or {}, dedupe_key=body.get("dedupe_key"), priority=int(body.get("priority") or 0), + coalesce_key=body.get("coalesce_key"), + source_version=body.get("source_version"), + tombstone=bool(body.get("tombstone")), ) runtime.worker_wake.set() except Exception as exc: # noqa: BLE001 - client gets structured failure @@ -1082,9 +1303,11 @@ def _drain_and_cleanup( if worker is not None: worker.join(timeout=SHUTDOWN_DRAIN_SECONDS) active = runtime.active_job_id - if active: + active_lease_token = runtime.active_lease_token + if active and active_lease_token: runtime._safe_finish( active, + active_lease_token, state="cancelled", result={"success": False, "exit_code": 1}, error={"message": "cancelled by daemon shutdown"}, @@ -1182,11 +1405,22 @@ def submit( *, dedupe_key: str | None = None, priority: int = 0, + coalesce_key: str | None = None, + source_version: int | None = None, + tombstone: bool = False, ) -> dict[str, Any]: return self.request( "POST", "/jobs", - {"kind": kind, "payload": payload, "dedupe_key": dedupe_key, "priority": priority}, + { + "kind": kind, + "payload": payload, + "dedupe_key": dedupe_key, + "priority": priority, + "coalesce_key": coalesce_key, + "source_version": source_version, + "tombstone": tombstone, + }, )["job"] def get_job(self, job_id: str) -> dict[str, Any]: diff --git a/mempalace/mcp_server.py b/mempalace/mcp_server.py index 402689fb4..8111c9830 100644 --- a/mempalace/mcp_server.py +++ b/mempalace/mcp_server.py @@ -345,10 +345,10 @@ def _parse_args(): # MCP startup/open SQLite integrity gate (#1818). # -# The peer-writer guard prevents new concurrent writers, but an MCP server can -# still start against a palace that was already left corrupt by a prior writer -# crash/kill. Run the existing read-only SQLite quick_check once on startup/open -# and fail loudly instead of silently serving a malformed FTS5/HNSW index. +# The daemon queue serializes current writes, but a server can still open a +# palace left corrupt by an earlier writer crash. Run the existing read-only +# SQLite quick_check once on startup/open and fail loudly instead of silently +# serving a malformed FTS5/HNSW index. _sqlite_integrity_checked = False _sqlite_integrity_errors: list[str] = [] _sqlite_integrity_check_error = "" @@ -408,6 +408,12 @@ def _parse_args(): _MCP_WRITER_ATEXIT_REGISTERED = False _MCP_ALLOW_PEER_WRITER_ENV = "MEMPALACE_MCP_ALLOW_PEER_WRITER" + +_MCP_WRITE_WAIT_SECONDS = 1.0 + + +# Write tools share the daemon queue and are the only tools disabled by the +# explicit server read-only mode. _MUTATING_TOOLS = frozenset( { "mempalace_kg_add", @@ -809,6 +815,210 @@ def _mcp_peer_writer_refusal(req_id, tool_name: str): } +def _daemon_job_payload(tool_name: str, tool_args: dict) -> tuple[str, dict, bool]: + """Translate an MCP mutation into one daemon-queue job. + + The bool marks maintenance work that must return immediately. Ordinary + writes get a very short compatibility wait, then degrade to the same job + handle rather than holding the MCP request open behind a busy writer. + """ + + if tool_name == "mempalace_mine": + return "mine", dict(tool_args), True + if tool_name == "mempalace_sync": + payload = { + "dir": tool_args.get("project_dir"), + "wing": tool_args.get("wing"), + "agent": tool_args.get("agent") or "mempalace", + "dry_run": not bool(tool_args.get("apply")), + } + if "changed" in tool_args or "deleted" in tool_args: + payload["changed_set"] = { + "changed": tool_args.get("changed") or [], + "deleted": tool_args.get("deleted") or [], + } + return "sync", payload, True + return "mcp_tool", {"name": tool_name, "arguments": dict(tool_args)}, False + + +def _accepted_daemon_job(job: dict) -> dict: + return { + "success": True, + "accepted": True, + "job_id": job.get("id"), + "state": job.get("state", "queued"), + } + + +def _file_ingress_event(project_dir: str, path: str, *, deleted: bool) -> tuple[str, int, dict]: + """Build a stable, versioned event for one repository-relative file.""" + + import hashlib + + project = os.path.realpath(os.path.abspath(os.path.expanduser(project_dir))) + candidate = os.path.realpath(os.path.join(project, path)) + if os.path.commonpath((project, candidate)) != project: + raise ValueError(f"changed-set path escapes project_dir: {path!r}") + rel = os.path.relpath(candidate, project).replace(os.sep, "/") + project_id = hashlib.sha256(os.path.normcase(project).encode("utf-8")).hexdigest()[:24] + key = f"file:{project_id}:{os.path.normcase(rel)}" + if deleted: + return ( + key, + time.time_ns(), + { + "canonical_path": candidate, + "relative_path": rel, + "deleted": True, + "content_hash": None, + }, + ) + stat = os.stat(candidate) + return ( + key, + int(stat.st_mtime_ns), + { + "canonical_path": candidate, + "relative_path": rel, + "deleted": False, + "source_version": int(stat.st_mtime_ns), + "size": int(stat.st_size), + "content_hash": None, + "hash_state": "deferred_to_worker", + }, + ) + + +def _submit_file_sync_jobs(client, tool_args: dict) -> dict: + project_dir = tool_args.get("project_dir") + if not project_dir: + raise ValueError("changed-set sync requires project_dir") + jobs = [] + events = [*((path, False) for path in tool_args.get("changed") or [])] + events.extend((path, True) for path in tool_args.get("deleted") or []) + for path, deleted in events: + key, version, ingress = _file_ingress_event(project_dir, path, deleted=deleted) + rel = ingress["relative_path"] + payload = { + "dir": project_dir, + "wing": tool_args.get("wing"), + "agent": tool_args.get("agent") or "mempalace", + "dry_run": not bool(tool_args.get("apply")), + "changed_set": { + "changed": [] if deleted else [rel], + "deleted": [rel] if deleted else [], + }, + "ingress": ingress, + } + job = client.submit( + "sync", + payload, + coalesce_key=key, + source_version=version, + tombstone=deleted, + ) + jobs.append({"id": job.get("id"), "state": job.get("state"), "path": rel}) + return { + "success": True, + "accepted": True, + "job_id": jobs[0]["id"] if len(jobs) == 1 else None, + "job_ids": [job["id"] for job in jobs], + "jobs": jobs, + } + + +def _submit_mcp_mutation(tool_name: str, tool_args: dict) -> dict: + """Submit every MCP mutation to the palace's single daemon writer. + + Multiple MCP sessions remain independent readers while one per-palace + daemon serializes writes. Mine/sync return an accepted job handle and + callers continue without routine polling. + Small writes preserve the historical immediate-result shape when they + finish inside a one-second budget; queue contention never blocks longer. + """ + + from .daemon import DaemonError, ensure_client + + kind, payload, return_immediately = _daemon_job_payload(tool_name, tool_args) + backend = os.environ.get("MEMPALACE_BACKEND_EXPLICIT") + try: + client = ensure_client(_config.palace_path, backend=backend, auto_start=True) + if tool_name == "mempalace_sync" and ("changed" in tool_args or "deleted" in tool_args): + return _submit_file_sync_jobs(client, tool_args) + job = client.submit(kind, payload) + if return_immediately: + return _accepted_daemon_job(job) + try: + finished = client.wait(job["id"], timeout=_MCP_WRITE_WAIT_SECONDS) + except DaemonError as exc: + if "timed out waiting" in str(exc).lower(): + return _accepted_daemon_job(job) + raise + except (DaemonError, OSError, ValueError) as exc: + return { + "success": False, + "error": f"daemon writer unavailable: {exc}", + "error_class": type(exc).__name__, + } + + result = finished.get("result") or {} + if not isinstance(result, dict): + result = {"success": finished.get("state") == "succeeded", "value": result} + result = dict(result) + result["daemon_job"] = {"id": finished.get("id"), "state": finished.get("state")} + if finished.get("state") != "succeeded": + result.setdefault("success", False) + error = finished.get("error") or {} + result.setdefault("error", error.get("message", "daemon write failed")) + return result + + +def _sanitized_job(job: dict) -> dict: + """Return daemon job metadata without exposing its queued payload.""" + + job = dict(job) + job.pop("payload", None) + state = job.get("state") + terminal = state in {"succeeded", "failed", "cancelled"} + job["success"] = True + job["status"] = state if terminal else "pending" + job["terminal"] = terminal + return job + + +def tool_job_status(job_id: str): + """Return one sanitized daemon job state for targeted diagnosis.""" + + from .daemon import DaemonError, get_client_if_running + + try: + client = get_client_if_running(_config.palace_path, health_timeout=0.5) + if client is None: + return {"success": False, "error": "daemon writer is not running"} + return _sanitized_job(client.get_job(job_id)) + except DaemonError as exc: + return {"success": False, "error": str(exc), "error_class": type(exc).__name__} + + +def tool_get_jobs(): + """Return active daemon jobs for debugging, without queued payloads.""" + + from .daemon import DaemonError, get_client_if_running + + try: + client = get_client_if_running(_config.palace_path, health_timeout=0.5) + if client is None: + return {"success": True, "jobs": [], "count": 0} + jobs = [ + _sanitized_job(job) + for job in client.list_jobs(limit=100) + if job.get("state") in {"queued", "running"} + ] + return {"success": True, "jobs": jobs, "count": len(jobs)} + except DaemonError as exc: + return {"success": False, "error": str(exc), "error_class": type(exc).__name__} + + def _startup_integrity_size_limit_bytes() -> int: """Byte size above which the startup SQLite quick_check is skipped. @@ -2085,7 +2295,6 @@ def tool_status(): # is detected so status stays reachable. db_exists = _backend_db_exists() _refresh_vector_disabled_flag() - if _vector_disabled: return _tool_status_via_sqlite() @@ -2448,12 +2657,13 @@ def tool_search( } if context: result["context_received"] = True - if not include_superseded and isinstance(result.get("results"), list): - result["results"] = [ - hit - for hit in result["results"] - if (hit.get("authority") or {}).get("status") != "superseded" - ] + if isinstance(result.get("results"), list): + if not include_superseded: + result["results"] = [ + hit + for hit in result["results"] + if (hit.get("authority") or {}).get("status") != "superseded" + ] result["count"] = len(result["results"]) return result @@ -3138,6 +3348,9 @@ def tool_supersede_drawer( "superseded_by": replacement_id, } ) + col = _get_collection() + if not col: + return _collection_error_or_no_palace() col.upsert(ids=old["ids"], documents=old["documents"], metadatas=updated_metas) _wal_log( "supersede_drawer", @@ -3595,8 +3808,15 @@ def tool_delete_by_source(source_file: str, dry_run: bool = True): return {"success": False, "error": str(e)} -def tool_sync(project_dir: str = None, wing: str = None, apply: bool = False): - """Prune drawers whose source files are gitignored, missing, or moved (#1252).""" +def tool_sync( + project_dir: str = None, + wing: str = None, + apply: bool = False, + changed: Optional[list[str]] = None, + deleted: Optional[list[str]] = None, + agent: str = "mempalace", +): + """Sync a full project scope or an explicit repository changed set.""" global _metadata_cache from .daemon import LOCK_REFUSAL_ERROR_CLASS from .palace import MineAlreadyRunning @@ -3608,13 +3828,28 @@ def tool_sync(project_dir: str = None, wing: str = None, apply: bool = False): project_dirs = [project_dir] if project_dir else None try: try: - report = sync_palace( - palace_path=_config.palace_path, - project_dirs=project_dirs, - wing=wing, - dry_run=not apply, - wal_log=_wal_log, - ) + if changed is not None or deleted is not None: + if not project_dir: + raise ValueError("changed-set sync requires project_dir") + from .changed_set import sync_changed_sources + + report = sync_changed_sources( + palace_path=_config.palace_path, + project_root=project_dir, + changed=changed or [], + deleted=deleted or [], + wing=wing, + agent=agent, + dry_run=not apply, + ) + else: + report = sync_palace( + palace_path=_config.palace_path, + project_dirs=project_dirs, + wing=wing, + dry_run=not apply, + wal_log=_wal_log, + ) return {"success": True, **report} # Order matters: typed handlers must precede the bare Exception # below, otherwise MineAlreadyRunning and ValueError fall into the @@ -5266,10 +5501,9 @@ def tool_patch_submit( "Mine a directory into the palace — the MCP equivalent of `mempalace mine`. " "mode='projects' (default) ingests code/docs; mode='convos' ingests chat " "transcripts; mode='extract' ingests office documents (PDF/DOCX/RTF, requires " - "the mempalace[extract] extra). Runs synchronously and returns the miner's " - "summary as `output`. The palace write lock is automatic; a concurrent mine " - "returns a structured already-running error. Orphan cleanup is separate — use " - "mempalace_sync." + "the mempalace[extract] extra). Queues work on the single palace writer and " + "returns immediately with an accepted job id. Orphan cleanup " + "is separate — use mempalace_sync." ), "input_schema": { "type": "object", @@ -5334,7 +5568,7 @@ def tool_patch_submit( "handler": tool_delete_by_source, }, "mempalace_sync": { - "description": "Prune drawers whose source files are gitignored, deleted, or moved. Returns dry-run report by default; pass apply=true to commit deletions.", + "description": "Queue an explicit changed-set sync or stale-drawer prune on the single palace writer. Accepted work is fire-and-forget; inspect jobs only after an error or suspected stall.", "input_schema": { "type": "object", "properties": { @@ -5345,12 +5579,45 @@ def tool_patch_submit( "wing": {"type": "string", "description": "Limit to one wing (optional)"}, "apply": { "type": "boolean", - "description": "Actually delete drawers; default is dry-run preview", + "description": "Apply reindex/deletion writes; default is dry-run preview", + }, + "changed": { + "type": "array", + "items": {"type": "string"}, + "description": "Repository-relative files to reindex; requires project_dir. Gitignored files are purged instead of reindexed.", + }, + "deleted": { + "type": "array", + "items": {"type": "string"}, + "description": "Repository-relative files to remove; requires project_dir", + }, + "agent": { + "type": "string", + "description": "Agent recorded on reindexed drawers (default: mempalace)", }, }, }, "handler": tool_sync, }, + "mempalace_job_status": { + "description": "Inspect one background job after an enqueue error or suspected stall. Non-terminal jobs report status=pending. Never returns the queued verbatim request payload.", + "input_schema": { + "type": "object", + "properties": { + "job_id": { + "type": "string", + "description": "Job id returned by a mutating MCP tool", + }, + }, + "required": ["job_id"], + }, + "handler": tool_job_status, + }, + "mempalace_get_jobs": { + "description": "Debug active queued/running MemPalace jobs. Call only after an enqueue error or suspected stall; accepted writes are otherwise fire-and-forget.", + "input_schema": {"type": "object", "properties": {}}, + "handler": tool_get_jobs, + }, "mempalace_get_drawer": { "description": "Fetch a single drawer by ID — returns full content and metadata.", "input_schema": { @@ -5777,7 +6044,7 @@ def _mcp_read_only_refusal(req_id, tool_name: str): """Refuse state-changing tools when the server runs in read-only mode (#1877). Read-only is an operator-set server mode (``--read-only`` / - ``MEMPALACE_MCP_READ_ONLY``), distinct from the dynamic peer-writer lock: + ``MEMPALACE_MCP_READ_ONLY``), distinct from daemon write serialization: it is an unconditional gate so a shared team server can expose recall without write access. Enforced at dispatch, not merely hidden from tools/list, so a client that calls a mutating tool by name is still refused. @@ -6408,14 +6675,194 @@ def _decorate_mcp_tool_result(tool_name: str, result): return result +def _missing_required_tool_args(tool_name: str, tool_args: dict) -> list[str]: + missing = [ + name + for name in TOOLS[tool_name]["input_schema"].get("required", []) + if tool_args.get(name) is None + ] + if tool_name == "mempalace_diary_write" and tool_args.get("entry") is None: + missing.append("entry") + return missing + + +def _dispatch_mcp_tool(req_id, tool_name: str, tool_args: dict): + from .service import classify_tool + + preflight_error = _mcp_tool_preflight_refusal(req_id, tool_name) + if preflight_error is not None: + return None, preflight_error + classification = classify_tool(tool_name) + if classification == "write" or tool_name in {"mempalace_mine", "mempalace_sync"}: + return _submit_mcp_mutation(tool_name, tool_args), None + result = _decorate_mcp_tool_result(tool_name, TOOLS[tool_name]["handler"](**tool_args)) + return result, None + + +def _mcp_error_response(req_id, code: int, message: str) -> dict: + """Build a JSON-RPC error response for dispatcher validation failures.""" + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": code, "message": message}, + } + + +def _mcp_tool_success_response(req_id, result) -> dict: + """Wrap a successful tool result in the MCP JSON-RPC envelope.""" + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": json.dumps(result, indent=2, ensure_ascii=False)}] + }, + } + + +def _prepare_mcp_tool_call(req_id, params): + """Validate, normalize, and authorize one ``tools/call`` request.""" + if not isinstance(params, dict) or "name" not in params: + return ( + None, + None, + _mcp_error_response( + req_id, -32602, "Invalid params: 'name' is required for tools/call" + ), + ) + + tool_name = params.get("name") + tool_args = params.get("arguments") or {} + if tool_name not in TOOLS: + return None, None, _mcp_error_response(req_id, -32601, f"Unknown tool: {tool_name}") + + # Enforce operator read-only mode before argument validation or daemon + # submission. A hidden write tool must consistently report its access + # denial rather than leaking a secondary schema error. + read_only_error = _mcp_read_only_refusal(req_id, tool_name) + if read_only_error is not None: + return None, None, read_only_error + + # Whitelist arguments to declared schema properties only. Prevents callers + # from spoofing internal params like added_by/source_file. Skip filtering if + # handler explicitly accepts **kwargs; default to filtering on inspect + # failure (safe fallback). + import inspect + + schema_props = TOOLS[tool_name]["input_schema"].get("properties", {}) + try: + handler = TOOLS[tool_name]["handler"] + sig = inspect.signature(handler) + accepts_var_keyword = any( + param.kind == inspect.Parameter.VAR_KEYWORD for param in sig.parameters.values() + ) + except (ValueError, TypeError): + accepts_var_keyword = False + if not accepts_var_keyword: + # An unknown kwarg here is almost always a wrong parameter *name* + # (e.g. text= instead of content=). Silently dropping it makes the + # cause surface only indirectly as a later "Missing required 'X'", + # so name it explicitly — symmetric with the missing-required path + # below. wait_for_previous is an internal transport kwarg in no tool + # schema; it is popped before dispatch further down, so it must not be + # reported as unknown here. + unknown = [ + key for key in tool_args if key not in schema_props and key != "wait_for_previous" + ] + if unknown: + quoted = ", ".join(f"'{key}'" for key in unknown) + word = "parameter" if len(unknown) == 1 else "parameters" + logger.debug("Tool %s: unknown %s %s", tool_name, word, quoted) + return ( + None, + None, + _mcp_error_response( + req_id, -32602, f"Unknown {word} {quoted} for tool {tool_name}" + ), + ) + tool_args = {key: value for key, value in tool_args.items() if key in schema_props} + + # MCP JSON transport may deliver integers as floats or strings; ChromaDB + # and Python slicing require native int. + for key, value in list(tool_args.items()): + prop_schema = schema_props.get(key, {}) + declared_type = prop_schema.get("type") + try: + if declared_type == "integer" and not isinstance(value, int): + tool_args[key] = int(value) + elif declared_type == "number" and not isinstance(value, (int, float)): + tool_args[key] = float(value) + except (ValueError, TypeError): + return ( + None, + None, + _mcp_error_response(req_id, -32602, f"Invalid value for parameter '{key}'"), + ) + tool_args.pop("wait_for_previous", None) + + # 'content' is an accepted alias for diary_write's 'entry' (callers often + # reuse add_drawer's 'content' name). Map it before dispatch so a + # content-only call still satisfies the required 'entry' parameter while + # the signature-based missing-parameter diagnostic (-32602) keeps working. + # 'entry' wins if both are supplied. + if tool_name == "mempalace_diary_write" and "content" in tool_args: + content_val = tool_args.pop("content") + # Only fill from the alias when the caller did not supply 'entry' at + # all (or passed it as null). An explicit entry — even "" — wins. + if "entry" not in tool_args or tool_args["entry"] is None: + tool_args["entry"] = content_val + missing = _missing_required_tool_args(tool_name, tool_args) + if missing: + quoted = ", ".join(f"'{name}'" for name in missing) + word = "parameter" if len(missing) == 1 else "parameters" + return ( + None, + None, + _mcp_error_response( + req_id, -32602, f"Missing required {word} {quoted} for tool {tool_name}" + ), + ) + return tool_name, tool_args, None + + +def _run_mcp_tool_call(req_id, tool_name: str, tool_args: dict): + """Dispatch a prepared tool call and translate handler failures to JSON-RPC.""" + try: + result, preflight_error = _dispatch_mcp_tool(req_id, tool_name, tool_args) + if preflight_error is not None: + return preflight_error + return _mcp_tool_success_response(req_id, result) + except TypeError as exc: + # Qualname match prevents leaking internal helper/param names raised + # inside the handler body — see test_handler_internal_signature_shape_stays_generic. + message = str(exc) + handler = TOOLS[tool_name]["handler"] + handler_qn = getattr(handler, "__qualname__", None) or getattr(handler, "__name__", "") + # Qualname can include "" for nested defs and "" for + # lambdas — accept Python's TypeError emit verbatim. + missing_match = re.match( + r"^([\w\.<>]+)\(\) missing \d+ required " + r"(?:positional |keyword-only )?arguments?: (.+)$", + message, + ) + if missing_match and missing_match.group(1) == handler_qn: + names = re.findall(r"'(\w+)'", missing_match.group(2)) + if names: + quoted = ", ".join(f"'{name}'" for name in names) + word = "parameter" if len(names) == 1 else "parameters" + logger.debug("Tool %s: missing required %s %s", tool_name, word, quoted) + return _mcp_error_response( + req_id, -32602, f"Missing required {word} {quoted} for tool {tool_name}" + ) + return _internal_tool_error(req_id, tool_name, exc) + except Exception as exc: + return _internal_tool_error(req_id, tool_name, exc) + + def handle_request(request): + """Route one JSON-RPC request to MCP initialization, discovery, or a tool.""" global _last_request_time if not isinstance(request, dict): - return { - "jsonrpc": "2.0", - "id": None, - "error": {"code": -32600, "message": "Invalid Request"}, - } + return _mcp_error_response(None, -32600, "Invalid Request") _last_request_time = time.monotonic() method = request.get("method") or "" params = request.get("params") or {} @@ -6437,10 +6884,10 @@ def handle_request(request): "serverInfo": {"name": "mempalace", "version": __version__}, }, } - elif method == "ping": + if method == "ping": return {"jsonrpc": "2.0", "id": req_id, "result": {}} - elif method.startswith("notifications/"): - # Notifications (no id) never get a response per JSON-RPC spec + if method.startswith("notifications/"): + # Notifications (no id) never get a response per JSON-RPC spec. return None elif method == "tools/list": # In read-only mode, hide the refused tools so clients don't advertise @@ -6593,12 +7040,9 @@ def handle_request(request): # Notifications (missing id) must never get a response if req_id is None: + # Notifications (missing id) must never get a response. return None - return { - "jsonrpc": "2.0", - "id": req_id, - "error": {"code": -32601, "message": f"Unknown method: {method}"}, - } + return _mcp_error_response(req_id, -32601, f"Unknown method: {method}") def _restore_stdout(): diff --git a/mempalace/service.py b/mempalace/service.py index d7cb6ef6b..942af0464 100644 --- a/mempalace/service.py +++ b/mempalace/service.py @@ -76,6 +76,8 @@ def _wrapped(*args, **kwargs): "mempalace_list_drawers", "mempalace_diary_read", "mempalace_memories_filed_away", + "mempalace_job_status", + "mempalace_get_jobs", "mempalace_kg_query", "mempalace_kg_stats", "mempalace_kg_timeline", @@ -88,6 +90,7 @@ def _wrapped(*args, **kwargs): "mempalace_checkpoint", "mempalace_delete_by_source", "mempalace_delete_drawer", + "mempalace_delete_by_source", "mempalace_supersede_drawer", "mempalace_update_drawer", "mempalace_diary_write", @@ -322,6 +325,34 @@ def run_sync(payload: dict[str, Any]) -> dict[str, Any]: project_dirs.extend(os.path.expanduser(str(root)) for root in payload.get("root") or []) project_dirs = project_dirs or None dry_run = bool(payload.get("dry_run", True)) + ingress = payload.get("ingress") or {} + if ingress and not ingress.get("deleted"): + import hashlib + + source = ingress.get("canonical_path") + try: + stat = os.stat(source) + if int(stat.st_mtime_ns) != int(ingress.get("source_version") or 0): + return { + "success": False, + "error": "source changed after enqueue; submit its newer generation", + "error_class": "StaleSourceVersion", + "exit_code": 1, + } + digest = hashlib.sha256() + with open(source, "rb") as file_handle: + for chunk in iter(lambda: file_handle.read(1024 * 1024), b""): + digest.update(chunk) + ingress = dict(ingress) + ingress["content_hash"] = digest.hexdigest() + ingress["hash_state"] = "computed_by_worker" + except OSError as exc: + return { + "success": False, + "error": f"could not read queued source: {exc}", + "error_class": type(exc).__name__, + "exit_code": 1, + } print(f"\n{'=' * 55}") print(" MemPalace Sync -- Gitignore-aware drawer prune") @@ -355,7 +386,7 @@ def run_sync(payload: dict[str, Any]) -> dict[str, Any]: agent=payload.get("agent") or "mempalace", dry_run=dry_run, ) - return {"success": True, "report": report, "exit_code": 0} + return {"success": True, "report": report, "ingress": ingress, "exit_code": 0} report = sync_palace( palace_path=palace_path, project_dirs=project_dirs, diff --git a/tests/test_daemon.py b/tests/test_daemon.py index aa19ae949..6218532fd 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -9,8 +9,43 @@ from _chroma_palace_helper import make_minimal_chroma_sqlite -from mempalace import daemon -from mempalace import service +from mempalace import daemon, service + + +@pytest.mark.parametrize("disconnect", [BrokenPipeError, ConnectionResetError]) +def test_json_response_swallows_client_disconnect_without_second_write(disconnect): + class _DeadSocket: + def __init__(self): + self.writes = 0 + + def write(self, _body): + self.writes += 1 + raise disconnect("client disconnected") + + class _Handler: + def __init__(self): + self.wfile = _DeadSocket() + self.close_connection = False + self.statuses = [] + + def send_response(self, status): + self.statuses.append(status) + + def send_header(self, *_args): + pass + + def end_headers(self): + pass + + handler = _Handler() + + written = daemon._json_response(handler, 200, {"ok": True}) + + assert written is False + assert handler.wfile.writes == 1 + assert handler.statuses == [200] + assert handler.close_connection is True + _LOCK_CONTENDER = """ from mempalace.palace import MineAlreadyRunning, mine_palace_lock @@ -169,6 +204,27 @@ def test_daemon_holds_local_backend_writer_lease_for_lifetime(tmp_path, monkeypa assert released.returncode == 0 +def test_two_independent_clients_enqueue_through_one_writer(tmp_path, monkeypatch): + calls = [] + + def fake_execute(kind, payload): + calls.append((kind, payload["name"])) + return {"success": True, "exit_code": 0} + + first, thread, palace, holders = _start_server(tmp_path, monkeypatch, fake_execute) + second = daemon.DaemonClient(str(palace)) + try: + job_a = first.submit("mcp_tool", {"name": "agent-a"}) + job_b = second.submit("mcp_tool", {"name": "agent-b"}) + + assert job_a["id"] != job_b["id"] + assert first.wait(job_a["id"], timeout=5)["state"] == "succeeded" + assert second.wait(job_b["id"], timeout=5)["state"] == "succeeded" + assert calls == [("mcp_tool", "agent-a"), ("mcp_tool", "agent-b")] + finally: + _stop_server(first, thread, holders) + + def test_submit_job_uses_client_and_waits(monkeypatch, tmp_path): palace = tmp_path / "palace" palace.mkdir() @@ -908,6 +964,92 @@ def fake_execute(kind, payload): _stop_server(client, thread, holders) +def test_file_ingress_coalesces_newest_queued_version_and_delete_wins_tie(tmp_path): + store = daemon.QueueStore(tmp_path / "queue.sqlite3") + first = store.enqueue( + "sync", + {"changed_set": {"changed": ["src/a.py"], "deleted": []}, "hash": "old"}, + coalesce_key="project:src/a.py", + source_version=10, + ) + newer = store.enqueue( + "sync", + {"changed_set": {"changed": ["src/a.py"], "deleted": []}, "hash": "new"}, + coalesce_key="project:src/a.py", + source_version=11, + ) + deleted = store.enqueue( + "sync", + {"changed_set": {"changed": [], "deleted": ["src/a.py"]}}, + coalesce_key="project:src/a.py", + source_version=11, + tombstone=True, + ) + + assert first.id == newer.id == deleted.id + queued = store.get(first.id) + assert queued.payload["changed_set"]["deleted"] == ["src/a.py"] + + +def test_file_ingress_keeps_one_successor_behind_running_generation(tmp_path): + store = daemon.QueueStore(tmp_path / "queue.sqlite3") + first = store.enqueue( + "sync", + {"generation": 1}, + coalesce_key="project:src/a.py", + source_version=1, + ) + assert store.claim_next().id == first.id + + successor = store.enqueue( + "sync", + {"generation": 2}, + coalesce_key="project:src/a.py", + source_version=2, + ) + latest = store.enqueue( + "sync", + {"generation": 3}, + coalesce_key="project:src/a.py", + source_version=3, + ) + + assert successor.id == latest.id + assert successor.id != first.id + assert store.get(successor.id).payload["generation"] == 3 + + +def test_expired_job_lease_is_requeued_and_heartbeat_renews_it(tmp_path, monkeypatch): + store = daemon.QueueStore(tmp_path / "queue.sqlite3") + job = store.enqueue("sync", {"generation": 1}) + claimed = store.claim_next() + assert claimed.id == job.id + assert store.heartbeat(job.id, claimed.lease_token) is True + + with store._connect() as conn: + conn.execute( + "UPDATE jobs SET lease_expires_at = ? WHERE id = ?", + ("2000-01-01T00:00:00+00:00", job.id), + ) + + assert store.recover_expired_leases() == 1 + assert store.get(job.id).state == "queued" + + reclaimed = store.claim_next() + assert reclaimed.lease_token != claimed.lease_token + assert store.heartbeat(job.id, claimed.lease_token) is False + store.finish( + job.id, + state="succeeded", + result={"stale": True}, + only_if_running=True, + lease_token=claimed.lease_token, + ) + still_running = store.get(job.id) + assert still_running.state == "running" + assert still_running.result is None + + def test_claim_next_does_not_reclaim_running_job(tmp_path, monkeypatch): """The conditional UPDATE (WHERE state='queued') means a job already flipped to 'running' cannot be claimed again — the cross-process @@ -1385,14 +1527,21 @@ def test_run_sync_dispatches_changed_set_without_full_scan(tmp_path, monkeypatch palace = tmp_path / "palace" palace.mkdir() - (palace / "chroma.sqlite3").touch() + make_minimal_chroma_sqlite(palace) project = tmp_path / "project" project.mkdir() captured = {} def fake_sync(**kwargs): captured.update(kwargs) - return {"changed": 1, "deleted": 1, "reindexed": 1, "drawers_added": 2, "dry_run": False} + return { + "changed": 1, + "deleted": 1, + "ignored": 0, + "reindexed": 1, + "drawers_added": 2, + "dry_run": False, + } monkeypatch.setattr(changed_set_module, "sync_changed_sources", fake_sync) result = service.run_sync( diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 808fad438..ef6b34f04 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -7917,3 +7917,270 @@ def test_search_schema_declares_window_properties(self): assert "before" in schema["properties"] assert schema["properties"]["since"]["type"] == "string" assert schema["properties"]["before"]["type"] == "string" + + +def test_tool_sync_routes_changed_set_through_mcp_writer(monkeypatch, tmp_path): + """Changed paths are applied inside the MCP process that owns the writer lease.""" + from mempalace import mcp_server + + palace = tmp_path / "palace" + project = tmp_path / "project" + project.mkdir() + cfg = MagicMock() + cfg.palace_path = str(palace) + monkeypatch.setattr(mcp_server, "_config", cfg) + + captured = {} + + def _sync_changed_sources(**kwargs): + captured.update(kwargs) + return { + "changed": 1, + "deleted": 1, + "ignored": 0, + "reindexed": 1, + "drawers_added": 2, + "dry_run": False, + } + + import mempalace.changed_set as changed_set_mod + + monkeypatch.setattr( + changed_set_mod, + "sync_changed_sources", + _sync_changed_sources, + ) + + result = mcp_server.tool_sync( + project_dir=str(project), + wing="contextunity", + apply=True, + changed=["src/new.py"], + deleted=["src/old.py"], + agent="contextunity", + ) + + assert result["success"] is True + assert result["changed"] == 1 + assert captured == { + "palace_path": str(palace), + "project_root": str(project), + "changed": ["src/new.py"], + "deleted": ["src/old.py"], + "wing": "contextunity", + "agent": "contextunity", + "dry_run": False, + } + + +def test_tool_sync_changed_set_requires_project_dir(monkeypatch, tmp_path): + """Reject changed-set mode without an explicit project security boundary.""" + from mempalace import mcp_server + + cfg = MagicMock() + cfg.palace_path = str(tmp_path / "palace") + monkeypatch.setattr(mcp_server, "_config", cfg) + + result = mcp_server.tool_sync(changed=["src/new.py"]) + + assert result["success"] is False + assert "project_dir" in result["error"] + + +def test_tool_sync_schema_exposes_changed_set_arguments(): + """Clients can discover the changed-set contract during MCP initialization.""" + from mempalace import mcp_server + + properties = mcp_server.TOOLS["mempalace_sync"]["input_schema"]["properties"] + + assert properties["changed"]["items"] == {"type": "string"} + assert properties["deleted"]["items"] == {"type": "string"} + assert properties["agent"]["type"] == "string" + + +def test_mcp_mine_is_submitted_without_waiting(monkeypatch, tmp_path): + from mempalace import daemon, mcp_server + + submitted = {} + + class _Client: + def submit(self, kind, payload, **kwargs): + submitted.update(kind=kind, payload=payload, kwargs=kwargs) + return {"id": "mine-1", "state": "queued"} + + def wait(self, *args, **kwargs): + raise AssertionError("long-running maintenance must not block MCP") + + monkeypatch.setattr(daemon, "ensure_client", lambda *args, **kwargs: _Client()) + monkeypatch.setattr( + mcp_server, + "_config", + type("Config", (), {"palace_path": str(tmp_path), "backend": "chroma"})(), + ) + + result = mcp_server._submit_mcp_mutation( + "mempalace_mine", + {"source": "/project", "mode": "projects", "agent": "codex"}, + ) + + assert result["accepted"] is True + assert result["job_id"] == "mine-1" + assert "poll_tool" not in result + assert submitted["kind"] == "mine" + assert submitted["payload"]["source"] == "/project" + + +def test_quick_write_returns_result_or_job_handle_with_bounded_wait(monkeypatch, tmp_path): + from mempalace import daemon, mcp_server + + class _Client: + def submit(self, kind, payload, **kwargs): + assert kind == "mcp_tool" + return {"id": "write-1", "state": "queued"} + + def wait(self, job_id, *, timeout): + assert timeout == mcp_server._MCP_WRITE_WAIT_SECONDS + raise daemon.DaemonError("timed out waiting for job write-1") + + monkeypatch.setattr(daemon, "ensure_client", lambda *args, **kwargs: _Client()) + monkeypatch.setattr( + mcp_server, + "_config", + type("Config", (), {"palace_path": str(tmp_path), "backend": "chroma"})(), + ) + + result = mcp_server._submit_mcp_mutation( + "mempalace_add_drawer", + {"wing": "w", "room": "r", "content": "verbatim"}, + ) + + assert result == { + "success": True, + "accepted": True, + "job_id": "write-1", + "state": "queued", + } + + +def test_job_status_never_returns_queued_write_payload(monkeypatch, tmp_path): + from mempalace import daemon, mcp_server + + class _Client: + def get_job(self, job_id): + return { + "id": job_id, + "state": "succeeded", + "payload": {"arguments": {"content": "secret verbatim text"}}, + "result": {"success": True, "drawer_id": "drawer-1"}, + } + + monkeypatch.setattr(daemon, "get_client_if_running", lambda *args, **kwargs: _Client()) + monkeypatch.setattr( + mcp_server, + "_config", + type("Config", (), {"palace_path": str(tmp_path)})(), + ) + + result = mcp_server.tool_job_status("write-1") + + assert "payload" not in result + assert result["success"] is True + assert result["terminal"] is True + assert result["status"] == "succeeded" + assert result["result"]["drawer_id"] == "drawer-1" + + +def test_job_status_reports_pending_without_false_failure(monkeypatch, tmp_path): + from mempalace import daemon, mcp_server + + class _Client: + def get_job(self, job_id): + return {"id": job_id, "state": "running", "payload": {"secret": "value"}} + + monkeypatch.setattr(daemon, "get_client_if_running", lambda *args, **kwargs: _Client()) + monkeypatch.setattr( + mcp_server, + "_config", + type("Config", (), {"palace_path": str(tmp_path)})(), + ) + + result = mcp_server.tool_job_status("write-1") + + assert result["status"] == "pending" + assert result["terminal"] is False + assert "succeeded" not in result + assert "payload" not in result + + +def test_get_jobs_returns_only_active_sanitized_jobs(monkeypatch, tmp_path): + from mempalace import daemon, mcp_server + + class _Client: + def list_jobs(self, limit): + assert limit == 100 + return [ + {"id": "job-1", "state": "succeeded", "payload": {"secret": "one"}}, + {"id": "job-2", "state": "queued", "payload": {"secret": "two"}}, + {"id": "job-3", "state": "running", "payload": {"secret": "three"}}, + ] + + monkeypatch.setattr(daemon, "get_client_if_running", lambda *args, **kwargs: _Client()) + monkeypatch.setattr( + mcp_server, + "_config", + type("Config", (), {"palace_path": str(tmp_path)})(), + ) + + result = mcp_server.tool_get_jobs() + + assert result["count"] == 2 + assert [job["id"] for job in result["jobs"]] == ["job-2", "job-3"] + assert all(job["status"] == "pending" for job in result["jobs"]) + assert all("payload" not in job for job in result["jobs"]) + + +def test_get_jobs_schema_requires_no_arguments(): + from mempalace import mcp_server + + schema = mcp_server.TOOLS["mempalace_get_jobs"]["input_schema"] + + assert schema == {"type": "object", "properties": {}} + + +def test_changed_set_enqueues_versioned_file_jobs_with_tombstone_precedence(monkeypatch, tmp_path): + from mempalace import daemon, mcp_server + + project = tmp_path / "project" + project.mkdir() + (project / "a.py").write_text("print('new')\n", encoding="utf-8") + calls = [] + + class _Client: + def submit(self, kind, payload, **kwargs): + calls.append((kind, payload, kwargs)) + return {"id": f"job-{len(calls)}", "state": "queued"} + + monkeypatch.setattr(daemon, "ensure_client", lambda *args, **kwargs: _Client()) + monkeypatch.setattr( + mcp_server, + "_config", + type("Config", (), {"palace_path": str(tmp_path / "palace")})(), + ) + + result = mcp_server._submit_mcp_mutation( + "mempalace_sync", + { + "project_dir": str(project), + "changed": ["a.py"], + "deleted": ["gone.py"], + "apply": True, + }, + ) + + assert result["job_ids"] == ["job-1", "job-2"] + assert calls[0][2]["coalesce_key"].endswith(":a.py") + assert calls[0][1]["ingress"]["content_hash"] is None + assert calls[0][1]["ingress"]["hash_state"] == "deferred_to_worker" + assert calls[0][2]["tombstone"] is False + assert calls[1][2]["coalesce_key"].endswith(":gone.py") + assert calls[1][2]["tombstone"] is True diff --git a/website/reference/mcp-tools.md b/website/reference/mcp-tools.md index 7306f7867..7967c392a 100644 --- a/website/reference/mcp-tools.md +++ b/website/reference/mcp-tools.md @@ -198,15 +198,33 @@ Bulk-delete every drawer mined from one `source_file` (exact match). Use this to ### `mempalace_sync` -Prune drawers whose source files are gitignored, deleted, or moved. Returns a dry-run report by default; pass `apply=true` to commit deletions. +Sync explicit repository changes inside the active MCP writer, or prune stale drawers across a project scope. Returns a dry-run report by default; pass `apply=true` to write. | Parameter | Type | Required | Description | |-----------|------|----------|-------------| | `project_dir` | string | No | Project root to scope the sync (auto-detected from drawer metadata if omitted) | | `wing` | string | No | Limit to one wing | -| `apply` | boolean | No | Actually delete drawers; default is dry-run preview | +| `apply` | boolean | No | Apply reindex/deletion writes; default is dry-run preview | +| `changed` | string[] | No | Repository-relative files to reindex; requires `project_dir` | +| `deleted` | string[] | No | Repository-relative files to remove; requires `project_dir` | +| `agent` | string | No | Agent recorded on reindexed drawers; default `mempalace` | -**Returns:** `{ scanned, kept, gitignored, missing, no_source, out_of_scope, removed_drawers, removed_closets, dry_run, by_source }` +With `changed` or `deleted`, the tool uses changed-set mode and returns `{ changed, deleted, ignored, reindexed, drawers_added, dry_run }`. Gitignored entries in `changed` are purge-only: existing drawers/closets are removed and the source is not reindexed. Otherwise it returns `{ scanned, kept, gitignored, missing, no_source, out_of_scope, removed_drawers, removed_closets, dry_run, by_source }`. + +--- + +### `mempalace_job_status` + +Inspect one background job by `job_id` after an enqueue error or suspected +stall. Queued and running jobs report `status: "pending"`; terminal jobs report +`status: "succeeded"`, `"failed"`, or `"cancelled"`. The queued verbatim request +payload is never returned. + +### `mempalace_get_jobs` + +Debug the current queued/running jobs without arguments. It returns sanitized +`jobs` and `count`, or an empty list when no daemon is active. Accepted writes +are fire-and-forget; callers should not invoke this tool after every enqueue. ---