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/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/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/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 ff4f28fd7..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", @@ -420,6 +426,7 @@ def _parse_args(): "mempalace_delete_drawer", "mempalace_checkpoint", "mempalace_delete_by_source", + "mempalace_supersede_drawer", "mempalace_mine", "mempalace_sync", "mempalace_update_drawer", @@ -808,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. @@ -2084,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() @@ -2373,6 +2583,8 @@ def tool_search( max_distance: float = 1.5, min_similarity: float = None, context: str = None, + verify_authority: bool = False, + include_superseded: bool = False, ): limit = max(1, min(limit, _MAX_RESULTS)) try: @@ -2406,6 +2618,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 +2639,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 @@ -2443,6 +2657,14 @@ def tool_search( } if context: result["context_received"] = True + 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 @@ -2915,7 +3137,15 @@ 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, + decision_key: str = None, ): """File verbatim content into a wing/room. Checks for duplicates first. @@ -2937,6 +3167,10 @@ 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 "" + decision_key = strip_lone_surrogates(decision_key or "").strip() except ValueError as e: return {"success": False, "error": str(e)} @@ -2967,6 +3201,15 @@ 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 + 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). @@ -3052,6 +3295,79 @@ 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 = _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", + { + "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 @@ -3492,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 @@ -3505,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 @@ -4419,15 +4757,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) + 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: @@ -4988,6 +5345,14 @@ 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.", + }, + "include_superseded": { + "type": "boolean", + "description": "Include explicitly superseded decision drawers (default false)", + }, }, "required": ["query"], }, @@ -5024,6 +5389,22 @@ 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)", + }, + "decision_key": { + "type": "string", + "description": "Stable logical key for a versioned decision (optional)", + }, }, "required": ["wing", "room", "content"], }, @@ -5049,6 +5430,11 @@ def tool_patch_submit( "type": "string", "description": "Verbatim content to store", }, + "authority_uri": {"type": "string"}, + "authority_version": {"type": "string"}, + "memory_kind": {"type": "string"}, + "decision_key": {"type": "string"}, + "supersedes_id": {"type": "string"}, }, "required": ["wing", "room", "content"], }, @@ -5079,6 +5465,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": { @@ -5095,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", @@ -5163,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": { @@ -5174,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": { @@ -5606,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. @@ -6237,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 {} @@ -6266,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 @@ -6422,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/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/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/mempalace/service.py b/mempalace/service.py index a54824730..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,8 @@ 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", "mempalace_kg_add", @@ -321,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") @@ -337,9 +369,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, "ingress": ingress, "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..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 @@ -1379,6 +1521,45 @@ 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() + make_minimal_chroma_sqlite(palace) + project = tmp_path / "project" + project.mkdir() + captured = {} + + def fake_sync(**kwargs): + captured.update(kwargs) + 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( + { + "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_mcp_server.py b/tests/test_mcp_server.py index 7bf3c5e50..ef6b34f04 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -2172,6 +2172,149 @@ 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_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 @@ -7774,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/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 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 diff --git a/website/reference/mcp-tools.md b/website/reference/mcp-tools.md index f6234ab96..7967c392a 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. @@ -167,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` | + +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` -**Returns:** `{ scanned, kept, gitignored, missing, no_source, out_of_scope, removed_drawers, removed_closets, dry_run, by_source }` +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. ---