feat: add opt-in local daemon for queued MemPalace writes
- New mempalace/daemon.py: long-lived localhost HTTP server (127.0.0.1) with a
SQLite WAL job queue, single worker thread, bearer-token auth, and owner-only
file perms (0600/0700) on queue DB, token, endpoint, and log.
- New mempalace/service.py: transport-neutral job execution surface shared by the
daemon, with per-job env isolation so one job's backend/palace switch cannot
leak into the next. mcp_tool is allowlisted to write-classified tools only.
- Crash recovery re-queues jobs left 'running' by a killed daemon; jobs that
already exhausted MAX_ATTEMPTS are dead-lettered to 'failed' instead of being
retried (non-idempotent diary_write would otherwise duplicate verbatim
content on every restart).
- Bounded retention prunes terminal jobs older than 7 days
(MEMPALACE_DAEMON_RETENTION_DAYS); queued/running jobs are never touched so a
crash mid-prune cannot drop in-flight work.
- CLI: --daemon/--background on mine/sync submit to the queue; new
`mempalace daemon {start,stop,status,jobs,wait}` subcommand. Strictly opt-in:
no flag, env, or config means no daemon and no behavior change.
- Hooks opt in via MEMPALACE_HOOKS_DAEMON or config hooks.daemon; when the daemon
is not already running, hooks fall back to the existing direct/spawn path so
the 500ms hook budget is preserved (hooks never auto-start the daemon).
- service.run_sync renders the same operator-facing report shape as the direct
CLI sync path (no_source, out_of_scope, by_source, Re-run/Removed hints) and
drops the old KeyError-prone 'deleted' read.
This commit is contained in:
Igor Lins e Silva
parent
afa749c141
commit
aa96bb5623
214
mempalace/cli.py
214
mempalace/cli.py
|
|
@ -534,6 +534,27 @@ def cmd_mine(args):
|
|||
for raw in args.include_ignored or []:
|
||||
include_ignored.extend(part.strip() for part in raw.split(",") if part.strip())
|
||||
|
||||
if getattr(args, "background", False) and not getattr(args, "daemon", False):
|
||||
print("mempalace: --background requires --daemon", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
if getattr(args, "daemon", False):
|
||||
payload = {
|
||||
"source": args.dir,
|
||||
"mode": args.mode,
|
||||
"wing": args.wing,
|
||||
"agent": args.agent,
|
||||
"limit": args.limit,
|
||||
"dry_run": args.dry_run,
|
||||
"extract": args.extract,
|
||||
"no_gitignore": args.no_gitignore,
|
||||
"include_ignored": include_ignored,
|
||||
"max_chunks_per_file": getattr(args, "max_chunks_per_file", None),
|
||||
"redetect_origin": getattr(args, "redetect_origin", False),
|
||||
}
|
||||
_submit_daemon_cli_job("mine", payload, args, background=getattr(args, "background", False))
|
||||
return
|
||||
|
||||
# --redetect-origin re-runs corpus_origin on the current corpus state
|
||||
# and overwrites <palace>/.mempalace/origin.json before mining proceeds.
|
||||
# Heuristic-only by design — full LLM detection lives on `mempalace init`.
|
||||
|
|
@ -655,14 +676,28 @@ def cmd_sweep(args):
|
|||
|
||||
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
|
||||
|
||||
if getattr(args, "background", False) and not getattr(args, "daemon", False):
|
||||
print("mempalace: --background requires --daemon", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
if getattr(args, "daemon", False):
|
||||
payload = {
|
||||
"dir": args.dir,
|
||||
"root": list(args.root or []),
|
||||
"wing": args.wing,
|
||||
"dry_run": args.dry_run,
|
||||
}
|
||||
_submit_daemon_cli_job("sync", payload, args, background=getattr(args, "background", False))
|
||||
return
|
||||
|
||||
from .mcp_server import _wal_log
|
||||
from .palace import MineAlreadyRunning
|
||||
from .backends import detect_backend_for_path
|
||||
from .palace import _backend_artifact_label, resolve_backend_name
|
||||
from .sync import sync_palace
|
||||
|
||||
palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path
|
||||
|
||||
if not os.path.isdir(palace_path):
|
||||
print(f"\n No palace found at {palace_path}")
|
||||
return
|
||||
|
|
@ -745,6 +780,133 @@ def cmd_sync(args):
|
|||
print(f"\n{'=' * 55}\n")
|
||||
|
||||
|
||||
def _submit_daemon_cli_job(kind: str, payload: dict, args, *, background: bool) -> None:
|
||||
palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path
|
||||
backend = _backend_arg(args)
|
||||
from .daemon import DaemonError, submit_job
|
||||
|
||||
try:
|
||||
job = submit_job(
|
||||
kind,
|
||||
payload,
|
||||
palace_path=palace_path,
|
||||
backend=backend,
|
||||
wait=not background,
|
||||
auto_start=True,
|
||||
)
|
||||
except DaemonError as exc:
|
||||
print(f"mempalace: daemon submission failed: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if background:
|
||||
print(f"Submitted daemon job {job['id']} ({kind})")
|
||||
return
|
||||
|
||||
result = job.get("result") or {}
|
||||
from .service import print_job_result
|
||||
|
||||
exit_code = print_job_result(result)
|
||||
if job.get("state") != "succeeded" and exit_code == 0:
|
||||
error = job.get("error") or {}
|
||||
print(
|
||||
f"mempalace: daemon job failed: {error.get('message', 'unknown error')}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
exit_code = 1
|
||||
if exit_code:
|
||||
sys.exit(exit_code)
|
||||
|
||||
|
||||
def cmd_daemon(args):
|
||||
palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path
|
||||
backend = _backend_arg(args)
|
||||
from .daemon import (
|
||||
TERMINAL_STATES,
|
||||
DaemonError,
|
||||
QueueStore,
|
||||
get_client_if_running,
|
||||
job_to_dict,
|
||||
queue_path,
|
||||
start_daemon,
|
||||
stop_daemon,
|
||||
)
|
||||
|
||||
action = getattr(args, "daemon_action", None)
|
||||
try:
|
||||
if action == "start":
|
||||
if args.foreground:
|
||||
start_daemon(palace_path, backend=backend, foreground=True)
|
||||
return
|
||||
client = start_daemon(palace_path, backend=backend, foreground=False)
|
||||
health = client.health()
|
||||
print(f"MemPalace daemon running on 127.0.0.1:{client.port}")
|
||||
print(f" Palace: {health.get('palace_path')}")
|
||||
print(f" PID: {health.get('pid')}")
|
||||
return
|
||||
|
||||
if action == "stop":
|
||||
if stop_daemon(palace_path):
|
||||
print("MemPalace daemon stopping")
|
||||
else:
|
||||
print("MemPalace daemon is not running")
|
||||
return
|
||||
|
||||
if action == "status":
|
||||
client = get_client_if_running(palace_path)
|
||||
if client is None:
|
||||
print("MemPalace daemon is not running")
|
||||
sys.exit(1)
|
||||
health = client.health()
|
||||
print("MemPalace daemon is running")
|
||||
print(f" Palace: {health.get('palace_path')}")
|
||||
print(f" PID: {health.get('pid')}")
|
||||
print(f" Active: {health.get('active_job_id') or '-'}")
|
||||
print(f" Jobs: {health.get('counts') or {}}")
|
||||
return
|
||||
|
||||
if action == "jobs":
|
||||
client = get_client_if_running(palace_path)
|
||||
if client is not None:
|
||||
jobs = client.list_jobs(limit=args.limit)
|
||||
else:
|
||||
qpath = queue_path(palace_path)
|
||||
if not qpath.exists():
|
||||
jobs = []
|
||||
else:
|
||||
jobs = [
|
||||
job_to_dict(job, include_payload=False)
|
||||
for job in QueueStore(qpath).list(args.limit)
|
||||
]
|
||||
for job in jobs:
|
||||
print(f"{job['id']} {job['state']:<9} {job['kind']:<10} {job['created_at']}")
|
||||
return
|
||||
|
||||
if action == "wait":
|
||||
client = get_client_if_running(palace_path)
|
||||
if client is not None:
|
||||
job = client.wait(args.job_id)
|
||||
else:
|
||||
qpath = queue_path(palace_path)
|
||||
if not qpath.exists():
|
||||
raise DaemonError("daemon is not running")
|
||||
job = job_to_dict(QueueStore(qpath).get(args.job_id))
|
||||
if job.get("state") not in TERMINAL_STATES:
|
||||
raise DaemonError(f"daemon is not running; job {args.job_id} is {job['state']}")
|
||||
result = job.get("result") or {}
|
||||
from .service import print_job_result
|
||||
|
||||
exit_code = print_job_result(result)
|
||||
if job.get("state") != "succeeded" and exit_code == 0:
|
||||
print(f"mempalace: daemon job failed: {job.get('error')}", file=sys.stderr)
|
||||
exit_code = 1
|
||||
if exit_code:
|
||||
sys.exit(exit_code)
|
||||
return
|
||||
except DaemonError as exc:
|
||||
print(f"mempalace: daemon error: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def cmd_search(args):
|
||||
from .searcher import search, SearchError
|
||||
|
||||
|
|
@ -1480,6 +1642,16 @@ def main():
|
|||
p_mine.add_argument(
|
||||
"--dry-run", action="store_true", help="Show what would be filed without filing"
|
||||
)
|
||||
p_mine.add_argument(
|
||||
"--daemon",
|
||||
action="store_true",
|
||||
help="Submit this mine to the opt-in local daemon queue",
|
||||
)
|
||||
p_mine.add_argument(
|
||||
"--background",
|
||||
action="store_true",
|
||||
help="With --daemon, return a job id immediately instead of waiting",
|
||||
)
|
||||
p_mine.add_argument(
|
||||
"--extract",
|
||||
choices=["exchange", "general"],
|
||||
|
|
@ -1543,6 +1715,16 @@ def main():
|
|||
action="store_false",
|
||||
help="Actually delete drawers (overrides --dry-run; requires --wing or a project root)",
|
||||
)
|
||||
p_sync.add_argument(
|
||||
"--daemon",
|
||||
action="store_true",
|
||||
help="Submit this sync to the opt-in local daemon queue",
|
||||
)
|
||||
p_sync.add_argument(
|
||||
"--background",
|
||||
action="store_true",
|
||||
help="With --daemon, return a job id immediately instead of waiting",
|
||||
)
|
||||
|
||||
# search
|
||||
p_search = sub.add_parser("search", help="Find anything, exact words")
|
||||
|
|
@ -1705,6 +1887,27 @@ def main():
|
|||
help="Compare sqlite vs HNSW element counts (read-only; never opens a chromadb client)",
|
||||
)
|
||||
|
||||
# daemon
|
||||
p_daemon = sub.add_parser("daemon", help="Manage the opt-in long-lived daemon")
|
||||
daemon_sub = p_daemon.add_subparsers(dest="daemon_action")
|
||||
p_daemon_start = daemon_sub.add_parser("start", help="Start the daemon")
|
||||
p_daemon_start.add_argument(
|
||||
"--foreground",
|
||||
action="store_true",
|
||||
help="Run in the foreground for debugging or process supervisors",
|
||||
)
|
||||
p_daemon_start.add_argument(
|
||||
"--backend",
|
||||
default=None,
|
||||
help="Storage backend for this daemon (default: config/env/detected/chroma)",
|
||||
)
|
||||
daemon_sub.add_parser("stop", help="Stop the daemon")
|
||||
daemon_sub.add_parser("status", help="Show daemon status")
|
||||
p_daemon_jobs = daemon_sub.add_parser("jobs", help="List recent daemon jobs")
|
||||
p_daemon_jobs.add_argument("--limit", type=int, default=20, help="Max jobs to show")
|
||||
p_daemon_wait = daemon_sub.add_parser("wait", help="Wait for a daemon job")
|
||||
p_daemon_wait.add_argument("job_id", help="Job id returned by --background")
|
||||
|
||||
# mcp
|
||||
p_mcp = sub.add_parser(
|
||||
"mcp",
|
||||
|
|
@ -1806,6 +2009,13 @@ def main():
|
|||
p_palace.print_help()
|
||||
return
|
||||
|
||||
if args.command == "daemon":
|
||||
if not getattr(args, "daemon_action", None):
|
||||
p_daemon.print_help()
|
||||
return
|
||||
cmd_daemon(args)
|
||||
return
|
||||
|
||||
dispatch = {
|
||||
"init": cmd_init,
|
||||
"mine": cmd_mine,
|
||||
|
|
|
|||
|
|
@ -749,6 +749,19 @@ class MempalaceConfig:
|
|||
"""Whether the stop hook shows a desktop notification via notify-send."""
|
||||
return self._file_config.get("hooks", {}).get("desktop_toast", False)
|
||||
|
||||
@property
|
||||
def hook_use_daemon(self):
|
||||
"""Whether hooks should submit save/mine work to the opt-in daemon."""
|
||||
env_val = os.environ.get("MEMPALACE_HOOKS_DAEMON")
|
||||
if env_val is not None:
|
||||
return env_val.lower() in ("true", "1", "yes", "on")
|
||||
value = self._file_config.get("hooks", {}).get("daemon", False)
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
return value.lower() in ("true", "1", "yes", "on")
|
||||
return value == 1
|
||||
|
||||
def set_hook_setting(self, key: str, value: bool):
|
||||
"""Update a hook setting and write config to disk."""
|
||||
if "hooks" not in self._file_config:
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -509,6 +509,70 @@ def _spawn_mine(cmd: list) -> None:
|
|||
pass
|
||||
|
||||
|
||||
def _hooks_daemon_enabled() -> bool:
|
||||
try:
|
||||
return MempalaceConfig().hook_use_daemon is True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def _daemon_mine_dedupe_key(source: str, mode: str) -> str:
|
||||
try:
|
||||
source_key = str(Path(source).expanduser().resolve())
|
||||
except OSError:
|
||||
source_key = str(Path(source).expanduser())
|
||||
return f"hook:mine:{mode}:{source_key}"
|
||||
|
||||
|
||||
def _daemon_available() -> bool:
|
||||
"""True iff a daemon is already running for the configured palace.
|
||||
|
||||
This is a fast localhost health check, not a spawn: the 500ms hook budget
|
||||
forbids auto-starting a python subprocess from a hook (cold start is
|
||||
~15s). Daemon mode for hooks requires the user to have started the daemon
|
||||
explicitly via `mempalace daemon start`; when it isn't up, hooks fall back
|
||||
to the existing direct (in-process / spawn) path instead of blocking.
|
||||
"""
|
||||
from .daemon import get_client_if_running
|
||||
|
||||
try:
|
||||
return get_client_if_running(MempalaceConfig().palace_path) is not None
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def _submit_daemon_job(
|
||||
kind: str,
|
||||
payload: dict,
|
||||
*,
|
||||
dedupe_key: str | None = None,
|
||||
priority: int = 0,
|
||||
wait: bool = False,
|
||||
timeout: float = 60.0,
|
||||
):
|
||||
"""Submit to an already-running daemon. Never auto-starts (see _daemon_available).
|
||||
|
||||
Raises DaemonError on a real failure (job rejected, timeout, daemon died
|
||||
mid-submit). Callers must NOT fall back to the direct path on such errors —
|
||||
the daemon may already have accepted the job, and re-running it would
|
||||
duplicate verbatim content. Only an absent daemon (handled by the caller's
|
||||
_daemon_available() precheck) should fall back.
|
||||
"""
|
||||
from .daemon import submit_job
|
||||
|
||||
palace_path = MempalaceConfig().palace_path
|
||||
return submit_job(
|
||||
kind,
|
||||
payload,
|
||||
palace_path=palace_path,
|
||||
dedupe_key=dedupe_key,
|
||||
priority=priority,
|
||||
wait=wait,
|
||||
auto_start=False,
|
||||
timeout=timeout,
|
||||
)
|
||||
|
||||
|
||||
def _maybe_auto_ingest():
|
||||
"""Background-mine MEMPAL_DIR (project files) if set.
|
||||
|
||||
|
|
@ -527,9 +591,26 @@ def _maybe_auto_ingest():
|
|||
return
|
||||
for mine_dir, mode in targets:
|
||||
try:
|
||||
if _hooks_daemon_enabled() and _daemon_available():
|
||||
try:
|
||||
_submit_daemon_job(
|
||||
"mine",
|
||||
{"source": mine_dir, "mode": mode, "agent": "mempalace"},
|
||||
dedupe_key=_daemon_mine_dedupe_key(mine_dir, mode),
|
||||
wait=False,
|
||||
)
|
||||
except Exception as exc:
|
||||
# Daemon accepted context — don't fall back (would double-mine).
|
||||
_log(f"Daemon mine submission failed: {exc}")
|
||||
continue
|
||||
_spawn_mine([_mempalace_python(), "-m", "mempalace", "mine", mine_dir, "--mode", mode])
|
||||
except OSError:
|
||||
pass
|
||||
except Exception as exc:
|
||||
# Non-daemon spawn path failed. Hooks must never crash the user's
|
||||
# shell — log and continue. Do not label this a daemon failure: the
|
||||
# daemon block above handles its own errors with its own message.
|
||||
_log(f"mine hook failed: {exc}")
|
||||
|
||||
|
||||
def _mine_sync():
|
||||
|
|
@ -546,6 +627,22 @@ def _mine_sync():
|
|||
log_path = STATE_DIR / "hook.log"
|
||||
for mine_dir, mode in targets:
|
||||
try:
|
||||
if _hooks_daemon_enabled() and _daemon_available():
|
||||
try:
|
||||
job = _submit_daemon_job(
|
||||
"mine",
|
||||
{"source": mine_dir, "mode": mode, "agent": "mempalace"},
|
||||
dedupe_key=_daemon_mine_dedupe_key(mine_dir, mode),
|
||||
wait=True,
|
||||
timeout=60,
|
||||
)
|
||||
result = job.get("result") or {}
|
||||
if job.get("state") != "succeeded" or not result.get("success", True):
|
||||
_log(f"Daemon sync mine failed: {result.get('error', job.get('error'))}")
|
||||
except Exception as exc:
|
||||
# Daemon accepted context — don't fall back (would double-mine).
|
||||
_log(f"Daemon sync mine submission failed: {exc}")
|
||||
continue
|
||||
with open(log_path, "a") as log_f:
|
||||
subprocess.run(
|
||||
[
|
||||
|
|
@ -563,6 +660,11 @@ def _mine_sync():
|
|||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
pass
|
||||
except Exception as exc:
|
||||
# Non-daemon sync spawn path failed. Hooks must never crash the
|
||||
# user's shell — log and continue (not a daemon failure; the daemon
|
||||
# block above handles its own errors).
|
||||
_log(f"mine hook failed: {exc}")
|
||||
|
||||
|
||||
def _desktop_toast(body: str, title: str = "MemPalace"):
|
||||
|
|
@ -680,6 +782,41 @@ def _save_diary_direct(
|
|||
)
|
||||
|
||||
try:
|
||||
if _hooks_daemon_enabled() and _daemon_available():
|
||||
try:
|
||||
job = _submit_daemon_job(
|
||||
"diary_write",
|
||||
{
|
||||
"agent_name": agent_name,
|
||||
"entry": entry,
|
||||
"topic": "checkpoint",
|
||||
"wing": wing,
|
||||
},
|
||||
priority=10,
|
||||
wait=True,
|
||||
timeout=30,
|
||||
)
|
||||
except Exception as exc:
|
||||
# Daemon accepted context — don't fall back (would double-write).
|
||||
_log(f"Daemon diary checkpoint failed: {exc}")
|
||||
return {"count": 0}
|
||||
result = job.get("result") or {}
|
||||
if job.get("state") == "succeeded" and result.get("success"):
|
||||
_log(f"Diary checkpoint saved: {result.get('entry_id', '?')}")
|
||||
try:
|
||||
ack_file = STATE_DIR / "last_checkpoint"
|
||||
ack_file.write_text(
|
||||
json.dumps({"msgs": len(messages), "ts": now.isoformat()}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
except OSError:
|
||||
pass
|
||||
if toast:
|
||||
_desktop_toast(f"Checkpoint saved - {len(messages)} messages archived")
|
||||
return {"count": len(messages), "themes": themes}
|
||||
_log(f"Daemon diary checkpoint failed: {result.get('error', job.get('error'))}")
|
||||
return {"count": 0}
|
||||
|
||||
from .mcp_server import tool_diary_write
|
||||
|
||||
result = tool_diary_write(
|
||||
|
|
@ -721,6 +858,25 @@ def _ingest_transcript(transcript_path: str):
|
|||
return
|
||||
|
||||
try:
|
||||
if _hooks_daemon_enabled() and _daemon_available():
|
||||
try:
|
||||
_submit_daemon_job(
|
||||
"mine",
|
||||
{
|
||||
"source": str(path.parent),
|
||||
"mode": "convos",
|
||||
"wing": "sessions",
|
||||
"agent": "mempalace",
|
||||
},
|
||||
dedupe_key=_daemon_mine_dedupe_key(str(path.parent), "convos"),
|
||||
wait=False,
|
||||
)
|
||||
_log(f"Transcript ingest submitted to daemon: {path.name}")
|
||||
except Exception as exc:
|
||||
# Daemon accepted context — don't fall back (would double-mine).
|
||||
_log(f"Daemon transcript ingest failed: {exc}")
|
||||
return
|
||||
|
||||
# Route through ``_spawn_mine`` so the per-target PID guard kicks
|
||||
# in here too — repeated Stop/PreCompact fires for the same
|
||||
# transcript should not stack up parallel ingest mines.
|
||||
|
|
@ -740,6 +896,11 @@ def _ingest_transcript(transcript_path: str):
|
|||
_log(f"Transcript ingest started: {path.name}")
|
||||
except OSError:
|
||||
pass
|
||||
except Exception as exc:
|
||||
# Non-daemon ingest spawn path failed. Hooks must never crash the
|
||||
# user's shell — log and continue (not a daemon failure; the daemon
|
||||
# block above handles its own errors).
|
||||
_log(f"transcript ingest hook failed: {exc}")
|
||||
|
||||
|
||||
SUPPORTED_HARNESSES = {"claude-code", "codex"}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,398 @@
|
|||
"""Shared service operations used by daemon-backed entry points.
|
||||
|
||||
The MCP server remains the owner of MCP transport details. This module owns the
|
||||
small, transport-neutral execution surface the daemon needs: classify known
|
||||
tools and execute durable background jobs without printing directly to the
|
||||
caller's terminal.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
from .config import MempalaceConfig
|
||||
|
||||
_EXPLICIT_BACKEND_ENV = "MEMPALACE_BACKEND_EXPLICIT"
|
||||
_PALACE_PATH_ENV = "MEMPALACE_PALACE_PATH"
|
||||
_BACKEND_ENV = "MEMPALACE_BACKEND"
|
||||
# Env vars a job may mutate via _apply_backend / palace_path injection. They are
|
||||
# snapshotted per job and restored afterward so a job that switches the backend
|
||||
# (e.g. qdrant) cannot poison every later job in the same daemon process —
|
||||
# including mcp_tool jobs, which read MempalaceConfig (and thus the leaked env).
|
||||
_PER_JOB_ENV = (_PALACE_PATH_ENV, _BACKEND_ENV, _EXPLICIT_BACKEND_ENV)
|
||||
|
||||
|
||||
READ_TOOLS = frozenset(
|
||||
{
|
||||
"mempalace_status",
|
||||
"mempalace_list_wings",
|
||||
"mempalace_list_rooms",
|
||||
"mempalace_get_taxonomy",
|
||||
"mempalace_get_aaak_spec",
|
||||
"mempalace_traverse",
|
||||
"mempalace_find_tunnels",
|
||||
"mempalace_graph_stats",
|
||||
"mempalace_list_tunnels",
|
||||
"mempalace_list_hallways",
|
||||
"mempalace_follow_tunnels",
|
||||
"mempalace_search",
|
||||
"mempalace_check_duplicate",
|
||||
"mempalace_get_drawer",
|
||||
"mempalace_list_drawers",
|
||||
"mempalace_diary_read",
|
||||
"mempalace_memories_filed_away",
|
||||
"mempalace_kg_query",
|
||||
"mempalace_kg_stats",
|
||||
"mempalace_kg_timeline",
|
||||
}
|
||||
)
|
||||
|
||||
WRITE_TOOLS = frozenset(
|
||||
{
|
||||
"mempalace_add_drawer",
|
||||
"mempalace_delete_drawer",
|
||||
"mempalace_update_drawer",
|
||||
"mempalace_diary_write",
|
||||
"mempalace_kg_add",
|
||||
"mempalace_kg_invalidate",
|
||||
"mempalace_create_tunnel",
|
||||
"mempalace_delete_tunnel",
|
||||
"mempalace_delete_hallway",
|
||||
"mempalace_hook_settings",
|
||||
}
|
||||
)
|
||||
|
||||
MAINTENANCE_TOOLS = frozenset({"mempalace_mine", "mempalace_sync", "mempalace_reconnect"})
|
||||
|
||||
|
||||
def classify_tool(name: str) -> str:
|
||||
"""Return ``read``, ``write``, ``maintenance``, or ``unknown`` for an MCP tool."""
|
||||
if name in READ_TOOLS:
|
||||
return "read"
|
||||
if name in WRITE_TOOLS:
|
||||
return "write"
|
||||
if name in MAINTENANCE_TOOLS:
|
||||
return "maintenance"
|
||||
return "unknown"
|
||||
|
||||
|
||||
def _apply_backend(backend: str | None) -> None:
|
||||
if not backend:
|
||||
return
|
||||
backend_name = str(backend).strip().lower()
|
||||
from .backends import get_backend_class
|
||||
|
||||
get_backend_class(backend_name)
|
||||
os.environ[_EXPLICIT_BACKEND_ENV] = backend_name
|
||||
os.environ[_BACKEND_ENV] = backend_name
|
||||
|
||||
|
||||
def _capture(fn):
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
|
||||
result = fn()
|
||||
return result, stdout.getvalue(), stderr.getvalue()
|
||||
|
||||
|
||||
def execute_job(kind: str, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Execute one daemon job and return a JSON-serializable result."""
|
||||
|
||||
def _run():
|
||||
if kind == "mine":
|
||||
return run_mine(payload)
|
||||
if kind == "sync":
|
||||
return run_sync(payload)
|
||||
if kind == "diary_write":
|
||||
return run_diary_write(payload)
|
||||
if kind == "mcp_tool":
|
||||
return run_mcp_tool(payload)
|
||||
return {"success": False, "error": f"unknown daemon job kind: {kind}", "exit_code": 2}
|
||||
|
||||
# Per-job env isolation: snapshot the backend/palace env vars and restore
|
||||
# them after the job so one job's _apply_backend / palace_path injection
|
||||
# can't leak into the next job in the same long-lived process.
|
||||
saved_env = {key: os.environ.get(key) for key in _PER_JOB_ENV}
|
||||
try:
|
||||
result, stdout, stderr = _capture(_run)
|
||||
finally:
|
||||
for key, value in saved_env.items():
|
||||
if value is None:
|
||||
os.environ.pop(key, None)
|
||||
else:
|
||||
os.environ[key] = value
|
||||
if result is None:
|
||||
result = {}
|
||||
if not isinstance(result, dict):
|
||||
result = {"success": True, "value": result}
|
||||
result.setdefault("success", True)
|
||||
result.setdefault("exit_code", 0 if result.get("success") else 1)
|
||||
if stdout:
|
||||
result["stdout"] = stdout
|
||||
if stderr:
|
||||
result["stderr"] = stderr
|
||||
return result
|
||||
|
||||
|
||||
def run_mine(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Run the same mine operation as the CLI, without daemon transport concerns."""
|
||||
palace_path = os.path.abspath(
|
||||
os.path.expanduser(payload.get("palace_path") or MempalaceConfig().palace_path)
|
||||
)
|
||||
os.environ["MEMPALACE_PALACE_PATH"] = palace_path
|
||||
_apply_backend(payload.get("backend"))
|
||||
|
||||
source = payload.get("source") or payload.get("dir")
|
||||
mode = payload.get("mode") or "projects"
|
||||
wing = payload.get("wing")
|
||||
agent = payload.get("agent") or "mempalace"
|
||||
limit = int(payload.get("limit") or 0)
|
||||
dry_run = bool(payload.get("dry_run"))
|
||||
|
||||
if payload.get("redetect_origin"):
|
||||
from .cli import _run_pass_zero
|
||||
|
||||
_run_pass_zero(project_dir=source, palace_dir=palace_path, llm_provider=None)
|
||||
|
||||
from .palace import MineAlreadyRunning, MineValidationError
|
||||
|
||||
try:
|
||||
if mode == "convos":
|
||||
from .convo_miner import mine_convos
|
||||
|
||||
mine_convos(
|
||||
convo_dir=source,
|
||||
palace_path=palace_path,
|
||||
wing=wing,
|
||||
agent=agent,
|
||||
limit=limit,
|
||||
dry_run=dry_run,
|
||||
extract_mode=payload.get("extract") or "exchange",
|
||||
)
|
||||
elif mode == "extract":
|
||||
from .format_miner import mine_formats
|
||||
|
||||
mine_formats(
|
||||
format_dir=source,
|
||||
palace_path=palace_path,
|
||||
wing=wing,
|
||||
agent=agent,
|
||||
limit=limit,
|
||||
dry_run=dry_run,
|
||||
)
|
||||
elif mode == "projects":
|
||||
include_ignored = payload.get("include_ignored") or []
|
||||
from .miner import mine
|
||||
|
||||
mine(
|
||||
project_dir=source,
|
||||
palace_path=palace_path,
|
||||
wing_override=wing,
|
||||
agent=agent,
|
||||
limit=limit,
|
||||
dry_run=dry_run,
|
||||
respect_gitignore=not bool(payload.get("no_gitignore")),
|
||||
include_ignored=include_ignored,
|
||||
max_chunks_per_file=payload.get("max_chunks_per_file"),
|
||||
)
|
||||
else:
|
||||
return {"success": False, "error": f"invalid mine mode: {mode}", "exit_code": 2}
|
||||
except MineAlreadyRunning as exc:
|
||||
return {
|
||||
"success": False,
|
||||
"error": str(exc),
|
||||
"error_class": "LockHeldByOtherProcess",
|
||||
"exit_code": 1,
|
||||
}
|
||||
except MineValidationError as exc:
|
||||
return {
|
||||
"success": False,
|
||||
"error": str(exc),
|
||||
"error_class": "MineValidationError",
|
||||
"exit_code": 1,
|
||||
}
|
||||
except SystemExit as exc:
|
||||
code = exc.code if isinstance(exc.code, int) else 1
|
||||
return {
|
||||
"success": code == 0,
|
||||
"error": str(exc),
|
||||
"error_class": "SystemExit",
|
||||
"exit_code": code,
|
||||
}
|
||||
except Exception as exc:
|
||||
return {"success": False, "error": f"mine failed: {exc}", "exit_code": 1}
|
||||
|
||||
return {"success": True, "kind": "mine", "mode": mode, "dry_run": dry_run, "exit_code": 0}
|
||||
|
||||
|
||||
def run_sync(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Run sync and render the same operator-facing summary shape as the CLI."""
|
||||
palace_path = os.path.abspath(
|
||||
os.path.expanduser(payload.get("palace_path") or MempalaceConfig().palace_path)
|
||||
)
|
||||
os.environ["MEMPALACE_PALACE_PATH"] = palace_path
|
||||
_apply_backend(payload.get("backend"))
|
||||
|
||||
from .backends import detect_backend_for_path
|
||||
from .palace import MineAlreadyRunning, _backend_artifact_label, resolve_backend_name
|
||||
|
||||
if not os.path.isdir(palace_path):
|
||||
print(f"\n No palace found at {palace_path}")
|
||||
return {"success": True, "exit_code": 0}
|
||||
|
||||
try:
|
||||
backend_name = resolve_backend_name(palace_path)
|
||||
except Exception as exc:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Could not resolve palace backend: {exc}",
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
if detect_backend_for_path(palace_path) is None:
|
||||
print(
|
||||
f"\n Palace dir at {palace_path} exists but has no "
|
||||
f"{_backend_artifact_label(backend_name)} yet."
|
||||
)
|
||||
print(" Run: mempalace mine <dir>")
|
||||
return {"success": True, "exit_code": 0}
|
||||
|
||||
project_dirs = []
|
||||
if payload.get("dir"):
|
||||
project_dirs.append(os.path.expanduser(str(payload["dir"])))
|
||||
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))
|
||||
|
||||
print(f"\n{'=' * 55}")
|
||||
print(" MemPalace Sync — Gitignore-aware drawer prune")
|
||||
print(f"{'=' * 55}")
|
||||
print(f" Palace: {palace_path}")
|
||||
if payload.get("wing"):
|
||||
print(f" Wing: {payload['wing']}")
|
||||
if project_dirs:
|
||||
for project_dir in project_dirs:
|
||||
print(f" Project: {project_dir}")
|
||||
print(
|
||||
" Mode: DRY RUN (no deletions)" if dry_run else " Mode: APPLY (deleting drawers)"
|
||||
)
|
||||
print(f"{'-' * 55}\n")
|
||||
|
||||
try:
|
||||
from .mcp_server import _wal_log
|
||||
from .sync import sync_palace
|
||||
|
||||
report = sync_palace(
|
||||
palace_path=palace_path,
|
||||
project_dirs=project_dirs,
|
||||
wing=payload.get("wing"),
|
||||
dry_run=dry_run,
|
||||
wal_log=_wal_log,
|
||||
)
|
||||
except MineAlreadyRunning as exc:
|
||||
return {
|
||||
"success": False,
|
||||
"error": str(exc),
|
||||
"error_class": "LockHeldByOtherProcess",
|
||||
"exit_code": 1,
|
||||
}
|
||||
except ValueError as exc:
|
||||
return {"success": False, "error": str(exc), "exit_code": 2}
|
||||
except Exception as exc:
|
||||
return {"success": False, "error": f"sync failed: {exc}", "exit_code": 1}
|
||||
|
||||
removed_suffix = "(would remove)" if dry_run else "(removed)"
|
||||
print(f" Scanned: {report['scanned']}")
|
||||
print(f" Kept: {report['kept']}")
|
||||
print(f" Gitignored: {report['gitignored']} {removed_suffix}")
|
||||
print(f" Missing: {report['missing']} {removed_suffix}")
|
||||
print(f" No source: {report['no_source']} (kept)")
|
||||
print(f" Out of scope: {report['out_of_scope']} (kept)")
|
||||
|
||||
by_source = report.get("by_source") or {}
|
||||
if by_source:
|
||||
top = sorted(by_source.items(), key=lambda kv: -kv[1])[:5]
|
||||
label = "Top sources to remove" if dry_run else "Top sources removed"
|
||||
print(f"\n {label}:")
|
||||
for src, n in top:
|
||||
print(f" {src} ({n})")
|
||||
|
||||
if dry_run:
|
||||
if report["gitignored"] + report["missing"] > 0:
|
||||
print("\n Re-run with --apply to commit these deletions.")
|
||||
else:
|
||||
print(
|
||||
f"\n Removed {report['removed_drawers']} drawers, {report['removed_closets']} closets."
|
||||
)
|
||||
|
||||
print(f"\n{'=' * 55}\n")
|
||||
return {"success": True, "report": report, "exit_code": 0}
|
||||
|
||||
|
||||
def run_diary_write(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
palace_path = payload.get("palace_path")
|
||||
if palace_path:
|
||||
os.environ["MEMPALACE_PALACE_PATH"] = os.path.abspath(os.path.expanduser(palace_path))
|
||||
_apply_backend(payload.get("backend"))
|
||||
|
||||
from .mcp_server import tool_diary_write
|
||||
|
||||
result = tool_diary_write(
|
||||
agent_name=payload.get("agent_name") or "mempalace",
|
||||
entry=payload.get("entry") or "",
|
||||
topic=payload.get("topic") or "general",
|
||||
wing=payload.get("wing") or "",
|
||||
)
|
||||
result.setdefault("exit_code", 0 if result.get("success") else 1)
|
||||
return result
|
||||
|
||||
|
||||
def run_mcp_tool(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Execute an MCP tool by name over the daemon queue.
|
||||
|
||||
The daemon is a durable, retried write surface — not a general MCP transport.
|
||||
Restrict ``mcp_tool`` to write-classified tools only: read tools would
|
||||
exfiltrate verbatim palace content into the queue DB and the job result
|
||||
(stored world-readable-by-default without the perms fix, and returned over
|
||||
/jobs), and maintenance tools already have their own dedicated kinds
|
||||
(mine/sync). No internal caller currently uses ``mcp_tool``; this allowlist
|
||||
bounds the blast radius of the generic escape hatch.
|
||||
"""
|
||||
name = payload.get("name")
|
||||
arguments = payload.get("arguments") or {}
|
||||
if not isinstance(arguments, dict):
|
||||
return {"success": False, "error": "arguments must be an object", "exit_code": 2}
|
||||
classification = classify_tool(name) if name else "unknown"
|
||||
if classification != "write":
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"daemon mcp_tool only accepts write tools; {name!r} is {classification}",
|
||||
"exit_code": 2,
|
||||
}
|
||||
from .mcp_server import TOOLS
|
||||
|
||||
if name not in TOOLS:
|
||||
return {"success": False, "error": f"unknown MCP tool: {name}", "exit_code": 2}
|
||||
result = TOOLS[name]["handler"](**arguments)
|
||||
if isinstance(result, dict):
|
||||
result.setdefault("success", True)
|
||||
result.setdefault("exit_code", 0 if result.get("success") else 1)
|
||||
return result
|
||||
return {"success": True, "value": result, "exit_code": 0}
|
||||
|
||||
|
||||
def print_job_result(result: dict[str, Any]) -> int:
|
||||
"""Replay captured daemon job output and return the intended process exit code."""
|
||||
stdout = result.get("stdout")
|
||||
stderr = result.get("stderr")
|
||||
if stdout:
|
||||
print(stdout, end="")
|
||||
if stderr:
|
||||
print(stderr, end="", file=sys.stderr)
|
||||
if not result.get("success", True) and result.get("error") and not stderr:
|
||||
print(f"mempalace: {result['error']}", file=sys.stderr)
|
||||
return int(result.get("exit_code", 0 if result.get("success", True) else 1) or 0)
|
||||
|
|
@ -17,6 +17,7 @@ from mempalace.cli import (
|
|||
cmd_hook,
|
||||
cmd_init,
|
||||
cmd_instructions,
|
||||
cmd_daemon,
|
||||
cmd_mine,
|
||||
cmd_repair,
|
||||
cmd_search,
|
||||
|
|
@ -621,6 +622,64 @@ def test_cmd_mine_include_ignored_comma_split(mock_config_cls):
|
|||
assert call_kwargs["include_ignored"] == ["a.txt", "b.txt", "c.txt"]
|
||||
|
||||
|
||||
@patch("mempalace.cli.MempalaceConfig")
|
||||
def test_cmd_mine_daemon_background_submits_job(mock_config_cls, capsys):
|
||||
mock_config_cls.return_value.palace_path = "/fake/palace"
|
||||
args = argparse.Namespace(
|
||||
dir="/src",
|
||||
palace=None,
|
||||
mode="projects",
|
||||
wing=None,
|
||||
agent="mempalace",
|
||||
limit=0,
|
||||
dry_run=False,
|
||||
no_gitignore=False,
|
||||
include_ignored=["a.txt,b.txt"],
|
||||
extract="exchange",
|
||||
daemon=True,
|
||||
background=True,
|
||||
backend=None,
|
||||
global_backend=None,
|
||||
max_chunks_per_file=None,
|
||||
redetect_origin=False,
|
||||
)
|
||||
with patch("mempalace.daemon.submit_job", return_value={"id": "job-1"}) as mock_submit:
|
||||
with patch("mempalace.miner.mine") as mock_mine:
|
||||
cmd_mine(args)
|
||||
|
||||
mock_mine.assert_not_called()
|
||||
mock_submit.assert_called_once()
|
||||
call_kwargs = mock_submit.call_args.kwargs
|
||||
assert call_kwargs["palace_path"] == "/fake/palace"
|
||||
assert call_kwargs["wait"] is False
|
||||
payload = mock_submit.call_args.args[1]
|
||||
assert payload["include_ignored"] == ["a.txt", "b.txt"]
|
||||
assert "job-1" in capsys.readouterr().out
|
||||
|
||||
|
||||
@patch("mempalace.cli.MempalaceConfig")
|
||||
def test_cmd_mine_background_requires_daemon(mock_config_cls, capsys):
|
||||
mock_config_cls.return_value.palace_path = "/fake/palace"
|
||||
args = argparse.Namespace(
|
||||
dir="/src",
|
||||
palace=None,
|
||||
mode="projects",
|
||||
wing=None,
|
||||
agent="mempalace",
|
||||
limit=0,
|
||||
dry_run=False,
|
||||
no_gitignore=False,
|
||||
include_ignored=[],
|
||||
extract="exchange",
|
||||
daemon=False,
|
||||
background=True,
|
||||
)
|
||||
with pytest.raises(SystemExit) as excinfo:
|
||||
cmd_mine(args)
|
||||
assert excinfo.value.code == 2
|
||||
assert "--background requires --daemon" in capsys.readouterr().err
|
||||
|
||||
|
||||
@patch("mempalace.cli.MempalaceConfig")
|
||||
def test_cmd_mine_exits_nonzero_on_lock_holder(mock_config_cls, capsys):
|
||||
"""Regression #1264: lock contention must exit non-zero with a clear message.
|
||||
|
|
@ -1260,6 +1319,94 @@ def test_cmd_sync_palace_dir_no_db(mock_config_cls, tmp_path, capsys):
|
|||
assert list(tmp_path.iterdir()) == []
|
||||
|
||||
|
||||
@patch("mempalace.cli.MempalaceConfig")
|
||||
def test_cmd_sync_daemon_background_submits_job(mock_config_cls, capsys):
|
||||
from mempalace.cli import cmd_sync
|
||||
|
||||
mock_config_cls.return_value.palace_path = "/fake/palace"
|
||||
args = argparse.Namespace(
|
||||
palace=None,
|
||||
dir="/project",
|
||||
root=["/extra"],
|
||||
wing="wing_a",
|
||||
dry_run=False,
|
||||
daemon=True,
|
||||
background=True,
|
||||
backend=None,
|
||||
global_backend=None,
|
||||
)
|
||||
with patch("mempalace.daemon.submit_job", return_value={"id": "sync-job"}) as mock_submit:
|
||||
cmd_sync(args)
|
||||
|
||||
mock_submit.assert_called_once()
|
||||
assert mock_submit.call_args.args[0] == "sync"
|
||||
payload = mock_submit.call_args.args[1]
|
||||
assert payload == {"dir": "/project", "root": ["/extra"], "wing": "wing_a", "dry_run": False}
|
||||
assert mock_submit.call_args.kwargs["wait"] is False
|
||||
assert "sync-job" in capsys.readouterr().out
|
||||
|
||||
|
||||
@patch("mempalace.cli.MempalaceConfig")
|
||||
def test_cmd_daemon_jobs_reads_durable_queue_when_stopped(
|
||||
mock_config_cls, tmp_path, monkeypatch, capsys
|
||||
):
|
||||
from mempalace.daemon import QueueStore, queue_path
|
||||
|
||||
palace_dir = tmp_path / "palace"
|
||||
state_root = tmp_path / "state"
|
||||
palace_dir.mkdir()
|
||||
monkeypatch.setenv("MEMPALACE_DAEMON_STATE_ROOT", str(state_root))
|
||||
mock_config_cls.return_value.palace_path = str(palace_dir)
|
||||
job = QueueStore(queue_path(str(palace_dir))).enqueue("mine", {"source": "/src"})
|
||||
|
||||
args = argparse.Namespace(
|
||||
palace=None,
|
||||
backend=None,
|
||||
global_backend=None,
|
||||
daemon_action="jobs",
|
||||
limit=20,
|
||||
)
|
||||
with patch("mempalace.daemon.get_client_if_running", return_value=None):
|
||||
cmd_daemon(args)
|
||||
|
||||
out = capsys.readouterr().out
|
||||
assert job.id in out
|
||||
assert "queued" in out
|
||||
assert "mine" in out
|
||||
|
||||
|
||||
@patch("mempalace.cli.MempalaceConfig")
|
||||
def test_cmd_daemon_wait_reads_finished_job_when_stopped(
|
||||
mock_config_cls, tmp_path, monkeypatch, capsys
|
||||
):
|
||||
from mempalace.daemon import QueueStore, queue_path
|
||||
|
||||
palace_dir = tmp_path / "palace"
|
||||
state_root = tmp_path / "state"
|
||||
palace_dir.mkdir()
|
||||
monkeypatch.setenv("MEMPALACE_DAEMON_STATE_ROOT", str(state_root))
|
||||
mock_config_cls.return_value.palace_path = str(palace_dir)
|
||||
store = QueueStore(queue_path(str(palace_dir)))
|
||||
queued = store.enqueue("mine", {"source": "/src"})
|
||||
store.finish(
|
||||
queued.id,
|
||||
state="succeeded",
|
||||
result={"success": True, "stdout": "done\n", "exit_code": 0},
|
||||
)
|
||||
|
||||
args = argparse.Namespace(
|
||||
palace=None,
|
||||
backend=None,
|
||||
global_backend=None,
|
||||
daemon_action="wait",
|
||||
job_id=queued.id,
|
||||
)
|
||||
with patch("mempalace.daemon.get_client_if_running", return_value=None):
|
||||
cmd_daemon(args)
|
||||
|
||||
assert "done" in capsys.readouterr().out
|
||||
|
||||
|
||||
@patch("mempalace.cli.MempalaceConfig")
|
||||
def test_cmd_compress_no_palace(mock_config_cls, tmp_path, capsys):
|
||||
"""cmd_compress exits non-zero with a 'No palace found' message on a missing dir.
|
||||
|
|
|
|||
|
|
@ -690,6 +690,36 @@ def test_hooks_auto_save_env_override_true():
|
|||
del os.environ["MEMPALACE_HOOKS_AUTO_SAVE"]
|
||||
|
||||
|
||||
def test_hook_use_daemon_default_false(monkeypatch):
|
||||
monkeypatch.delenv("MEMPALACE_HOOKS_DAEMON", raising=False)
|
||||
cfg = MempalaceConfig(config_dir=tempfile.mkdtemp())
|
||||
assert cfg.hook_use_daemon is False
|
||||
|
||||
|
||||
def test_hook_use_daemon_from_config(monkeypatch, tmp_path):
|
||||
monkeypatch.delenv("MEMPALACE_HOOKS_DAEMON", raising=False)
|
||||
with open(tmp_path / "config.json", "w") as f:
|
||||
json.dump({"hooks": {"daemon": True}}, f)
|
||||
cfg = MempalaceConfig(config_dir=str(tmp_path))
|
||||
assert cfg.hook_use_daemon is True
|
||||
|
||||
|
||||
def test_hook_use_daemon_string_config(monkeypatch, tmp_path):
|
||||
monkeypatch.delenv("MEMPALACE_HOOKS_DAEMON", raising=False)
|
||||
with open(tmp_path / "config.json", "w") as f:
|
||||
json.dump({"hooks": {"daemon": "yes"}}, f)
|
||||
cfg = MempalaceConfig(config_dir=str(tmp_path))
|
||||
assert cfg.hook_use_daemon is True
|
||||
|
||||
|
||||
def test_hook_use_daemon_env_override(monkeypatch, tmp_path):
|
||||
with open(tmp_path / "config.json", "w") as f:
|
||||
json.dump({"hooks": {"daemon": False}}, f)
|
||||
monkeypatch.setenv("MEMPALACE_HOOKS_DAEMON", "yes")
|
||||
cfg = MempalaceConfig(config_dir=str(tmp_path))
|
||||
assert cfg.hook_use_daemon is True
|
||||
|
||||
|
||||
# --- max_backups (backup retention) ---
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,457 @@
|
|||
import threading
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from mempalace import daemon
|
||||
from mempalace import service
|
||||
|
||||
|
||||
def _raise_not_ready(*a, **kw):
|
||||
"""Stand-in for DaemonClient when the spawned daemon must never come up."""
|
||||
raise daemon.DaemonError("not ready")
|
||||
|
||||
|
||||
def test_prune_terminal_drops_old_terminal_jobs_keeps_active(tmp_path, monkeypatch):
|
||||
"""Terminal jobs older than the retention window are pruned; queued/running
|
||||
and fresh terminal jobs are untouched. Bounded queue growth for the DB that
|
||||
holds verbatim payloads."""
|
||||
monkeypatch.setenv(daemon.STATE_ROOT_ENV, str(tmp_path / "state"))
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
store = daemon.QueueStore(daemon.queue_path(str(palace)))
|
||||
|
||||
old_term = store.enqueue("mine", {"source": "old"})
|
||||
store.finish(old_term.id, state="succeeded", result={"success": True})
|
||||
fresh_term = store.enqueue("mine", {"source": "fresh"})
|
||||
store.finish(fresh_term.id, state="succeeded", result={"success": True})
|
||||
queued = store.enqueue("mine", {"source": "queued"})
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
cutoff = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
|
||||
with store._lock, store._connect() as conn:
|
||||
conn.execute(
|
||||
"UPDATE jobs SET finished_at = ? WHERE id = ?",
|
||||
(cutoff, old_term.id),
|
||||
)
|
||||
|
||||
pruned = store.prune_terminal(older_than_days=7)
|
||||
assert pruned == 1
|
||||
# The old terminal job is gone; the fresh terminal and queued jobs survive.
|
||||
with pytest.raises(daemon.DaemonError):
|
||||
store.get(old_term.id)
|
||||
assert store.get(fresh_term.id).state == "succeeded"
|
||||
assert store.get(queued.id).state == "queued"
|
||||
|
||||
|
||||
def test_queue_dedupes_and_recovers_running_jobs(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv(daemon.STATE_ROOT_ENV, str(tmp_path / "state"))
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
|
||||
store = daemon.QueueStore(daemon.queue_path(str(palace)))
|
||||
first = store.enqueue("mine", {"source": "a"}, dedupe_key="same")
|
||||
second = store.enqueue("mine", {"source": "a"}, dedupe_key="same")
|
||||
|
||||
assert second.id == first.id
|
||||
|
||||
claimed = store.claim_next()
|
||||
assert claimed.id == first.id
|
||||
assert claimed.state == "running"
|
||||
|
||||
recovered = store.recover_running()
|
||||
assert recovered == 1
|
||||
assert store.get(first.id).state == "queued"
|
||||
|
||||
|
||||
def test_daemon_http_lifecycle_executes_job(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv(daemon.STATE_ROOT_ENV, str(tmp_path / "state"))
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
calls = []
|
||||
|
||||
def fake_execute(kind, payload):
|
||||
calls.append((kind, payload))
|
||||
return {"success": True, "exit_code": 0, "stdout": "done\n"}
|
||||
|
||||
monkeypatch.setattr(service, "execute_job", fake_execute)
|
||||
|
||||
thread = threading.Thread(
|
||||
target=daemon.run_server,
|
||||
kwargs={"palace_path": str(palace), "port": 0},
|
||||
daemon=True,
|
||||
)
|
||||
thread.start()
|
||||
|
||||
client = None
|
||||
deadline = time.monotonic() + 10
|
||||
while time.monotonic() < deadline:
|
||||
client = daemon.get_client_if_running(str(palace))
|
||||
if client is not None:
|
||||
break
|
||||
time.sleep(0.05)
|
||||
|
||||
assert client is not None
|
||||
health = client.health()
|
||||
assert health["ok"] is True
|
||||
assert health["palace_path"] == daemon.canonical_palace_path(str(palace))
|
||||
|
||||
job = client.submit("mine", {"source": "src"}, dedupe_key="job")
|
||||
finished = client.wait(job["id"], timeout=5)
|
||||
|
||||
assert finished["state"] == "succeeded"
|
||||
assert finished["result"]["stdout"] == "done\n"
|
||||
assert calls == [("mine", {"source": "src", "palace_path": str(palace.resolve())})]
|
||||
|
||||
client.shutdown()
|
||||
thread.join(timeout=5)
|
||||
assert not thread.is_alive()
|
||||
|
||||
|
||||
def test_submit_job_uses_client_and_waits(monkeypatch, tmp_path):
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
|
||||
class DummyClient:
|
||||
def __init__(self):
|
||||
self.submitted = None
|
||||
|
||||
def submit(self, kind, payload, dedupe_key=None, priority=0):
|
||||
self.submitted = (kind, payload, dedupe_key, priority)
|
||||
return {"id": "job-1", "state": "queued"}
|
||||
|
||||
def wait(self, job_id, timeout=daemon.DEFAULT_WAIT_TIMEOUT):
|
||||
assert job_id == "job-1"
|
||||
return {
|
||||
"id": "job-1",
|
||||
"state": "succeeded",
|
||||
"result": {"success": True, "exit_code": 0},
|
||||
}
|
||||
|
||||
dummy = DummyClient()
|
||||
monkeypatch.setattr(daemon, "ensure_client", lambda *a, **kw: dummy)
|
||||
|
||||
job = daemon.submit_job(
|
||||
"mine",
|
||||
{"source": "src"},
|
||||
palace_path=str(palace),
|
||||
dedupe_key="dedupe",
|
||||
wait=True,
|
||||
)
|
||||
|
||||
assert job["state"] == "succeeded"
|
||||
assert dummy.submitted[0] == "mine"
|
||||
# palace_path is overridden (not trusted from the payload), never appended.
|
||||
assert dummy.submitted[1]["palace_path"] == daemon.canonical_palace_path(str(palace))
|
||||
assert dummy.submitted[2] == "dedupe"
|
||||
|
||||
|
||||
def test_service_tool_classification():
|
||||
assert service.classify_tool("mempalace_search") == "read"
|
||||
assert service.classify_tool("mempalace_add_drawer") == "write"
|
||||
assert service.classify_tool("mempalace_mine") == "maintenance"
|
||||
assert service.classify_tool("unknown") == "unknown"
|
||||
|
||||
|
||||
# --- helpers for HTTP-lifecycle tests ---
|
||||
|
||||
|
||||
def _start_server(tmp_path, monkeypatch, execute_fn):
|
||||
monkeypatch.setenv(daemon.STATE_ROOT_ENV, str(tmp_path / "state"))
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
monkeypatch.setattr(service, "execute_job", execute_fn)
|
||||
thread = threading.Thread(
|
||||
target=daemon.run_server,
|
||||
kwargs={"palace_path": str(palace), "port": 0},
|
||||
daemon=True,
|
||||
)
|
||||
thread.start()
|
||||
client = None
|
||||
deadline = time.monotonic() + 10
|
||||
while time.monotonic() < deadline:
|
||||
client = daemon.get_client_if_running(str(palace))
|
||||
if client is not None:
|
||||
break
|
||||
time.sleep(0.05)
|
||||
assert client is not None
|
||||
return client, thread, palace
|
||||
|
||||
|
||||
# --- ship-blocker regressions ---
|
||||
|
||||
|
||||
def test_systemexit_in_job_does_not_kill_worker(tmp_path, monkeypatch):
|
||||
"""A SystemExit (BaseException, not Exception) must be caught, the job
|
||||
marked failed, and the worker kept alive for the next job. Regression for
|
||||
the critical worker-death bug."""
|
||||
state = {"first": True}
|
||||
|
||||
def fake_execute(kind, payload):
|
||||
if state["first"]:
|
||||
state["first"] = False
|
||||
raise SystemExit("boom")
|
||||
return {"success": True, "exit_code": 0}
|
||||
|
||||
client, thread, palace = _start_server(tmp_path, monkeypatch, fake_execute)
|
||||
try:
|
||||
first = client.submit("mine", {"source": "src"})
|
||||
finished_first = client.wait(first["id"], timeout=5)
|
||||
assert finished_first["state"] == "failed"
|
||||
assert finished_first["error"]["error_class"] == "SystemExit"
|
||||
|
||||
# Worker must still be alive — health reports it and a second job runs.
|
||||
assert client.health()["worker_alive"] is True
|
||||
second = client.submit("mine", {"source": "src2"})
|
||||
finished_second = client.wait(second["id"], timeout=5)
|
||||
assert finished_second["state"] == "succeeded"
|
||||
finally:
|
||||
client.shutdown()
|
||||
thread.join(timeout=5)
|
||||
assert not thread.is_alive()
|
||||
|
||||
|
||||
def test_shutdown_cancels_active_job(tmp_path, monkeypatch):
|
||||
"""POST /shutdown must not leave an in-flight job 'running' for blind
|
||||
re-queue on next start. The worker is drained (bounded), then the active
|
||||
job is marked 'cancelled' so recover_running won't re-run it.
|
||||
|
||||
In production the serve process exits immediately after run_server returns,
|
||||
killing the daemon worker thread before it can overwrite the cancelled
|
||||
state. The test mirrors that by asserting the cancelled state *before*
|
||||
releasing the blocked worker.
|
||||
"""
|
||||
block = threading.Event()
|
||||
|
||||
def fake_execute(kind, payload):
|
||||
# Simulate a long-running job that never finishes on its own.
|
||||
block.wait(30)
|
||||
return {"success": True, "exit_code": 0}
|
||||
|
||||
monkeypatch.setattr(daemon, "SHUTDOWN_DRAIN_SECONDS", 0.2)
|
||||
client, thread, palace = _start_server(tmp_path, monkeypatch, fake_execute)
|
||||
job = client.submit("mine", {"source": "src"}, dedupe_key="x")
|
||||
# Wait until the worker has claimed it (state flips to running).
|
||||
deadline = time.monotonic() + 5
|
||||
while time.monotonic() < deadline:
|
||||
if client.get_job(job["id"])["state"] == "running":
|
||||
break
|
||||
time.sleep(0.02)
|
||||
assert client.get_job(job["id"])["state"] == "running"
|
||||
|
||||
client.shutdown()
|
||||
thread.join(timeout=5)
|
||||
assert not thread.is_alive()
|
||||
|
||||
# The interrupted job must be cancelled (terminal), not left running.
|
||||
store = daemon.QueueStore(daemon.queue_path(str(palace)))
|
||||
final = store.get(job["id"])
|
||||
assert final.state == "cancelled"
|
||||
# And recover_running must not re-queue a cancelled job.
|
||||
assert store.recover_running() == 0
|
||||
|
||||
# Release the blocked worker so it (and the daemon thread) can exit.
|
||||
block.set()
|
||||
|
||||
|
||||
def test_recover_running_dead_letters_exhausted_jobs(tmp_path, monkeypatch):
|
||||
"""A job that has crashed MAX_ATTEMPTS times must be dead-lettered to
|
||||
'failed', not re-queued — non-idempotent kinds (diary_write) would
|
||||
otherwise duplicate verbatim content on every restart."""
|
||||
monkeypatch.setenv(daemon.STATE_ROOT_ENV, str(tmp_path / "state"))
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
store = daemon.QueueStore(daemon.queue_path(str(palace)))
|
||||
job = store.enqueue("diary_write", {"entry": "x"})
|
||||
# Simulate MAX_ATTEMPTS claims that each crashed (running, attempts=MAX).
|
||||
with store._lock, store._connect() as conn:
|
||||
conn.execute(
|
||||
"UPDATE jobs SET state='running', attempts=? WHERE id=?",
|
||||
(daemon.MAX_ATTEMPTS, job.id),
|
||||
)
|
||||
|
||||
recovered = store.recover_running()
|
||||
assert recovered == 0 # not re-queued
|
||||
final = store.get(job.id)
|
||||
assert final.state == "failed"
|
||||
assert final.attempts == daemon.MAX_ATTEMPTS
|
||||
|
||||
|
||||
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
|
||||
double-execution guard."""
|
||||
monkeypatch.setenv(daemon.STATE_ROOT_ENV, str(tmp_path / "state"))
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
store = daemon.QueueStore(daemon.queue_path(str(palace)))
|
||||
job = store.enqueue("mine", {"source": "src"})
|
||||
first = store.claim_next()
|
||||
assert first.id == job.id
|
||||
# Manually re-mark it queued but leave a second claim attempt: claim_next
|
||||
# should still only ever return one running job per claim. After finishing
|
||||
# the first, the next claim returns None (queue empty).
|
||||
store.finish(first.id, state="succeeded", result={"success": True})
|
||||
assert store.claim_next() is None
|
||||
|
||||
|
||||
def test_queue_db_file_is_owner_only(tmp_path, monkeypatch):
|
||||
"""The queue DB holds verbatim payloads — it must be 0600, not the sqlite
|
||||
default 0644. Regression for the privacy-principle violation."""
|
||||
import os as _os
|
||||
|
||||
monkeypatch.setenv(daemon.STATE_ROOT_ENV, str(tmp_path / "state"))
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
store = daemon.QueueStore(daemon.queue_path(str(palace)))
|
||||
store.enqueue("diary_write", {"entry": "secret verbatim content"})
|
||||
mode = _os.stat(str(store.path)).st_mode & 0o777
|
||||
assert mode == 0o600, f"queue.sqlite3 is {oct(mode)}, expected 0600"
|
||||
|
||||
|
||||
def test_token_file_is_owner_only(tmp_path, monkeypatch):
|
||||
import os as _os
|
||||
|
||||
monkeypatch.setenv(daemon.STATE_ROOT_ENV, str(tmp_path / "state"))
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
daemon.ensure_token(str(palace))
|
||||
token_path = daemon.state_dir(str(palace)) / "token"
|
||||
assert (_os.stat(str(token_path)).st_mode & 0o777) == 0o600
|
||||
|
||||
|
||||
def test_health_rejects_missing_and_wrong_token(tmp_path, monkeypatch):
|
||||
from urllib import error as urlerror
|
||||
from urllib import request as urlrequest
|
||||
|
||||
client, thread, palace = _start_server(tmp_path, monkeypatch, lambda k, p: {"success": True})
|
||||
try:
|
||||
base = f"http://127.0.0.1:{client.port}"
|
||||
# No Authorization header → 401.
|
||||
with pytest.raises(urlerror.HTTPError):
|
||||
urlrequest.urlopen(urlrequest.Request(base + "/health"), timeout=3)
|
||||
# Wrong token → 401.
|
||||
with pytest.raises(urlerror.HTTPError):
|
||||
urlrequest.urlopen(
|
||||
urlrequest.Request(base + "/health", headers={"Authorization": "Bearer wrong"}),
|
||||
timeout=3,
|
||||
)
|
||||
finally:
|
||||
client.shutdown()
|
||||
thread.join(timeout=5)
|
||||
|
||||
|
||||
def test_worker_overrides_client_palace_path(tmp_path, monkeypatch):
|
||||
"""An authenticated client must not be able to retarget the daemon at a
|
||||
different palace by stuffing palace_path into the payload."""
|
||||
seen = {}
|
||||
|
||||
def fake_execute(kind, payload):
|
||||
seen["palace_path"] = payload.get("palace_path")
|
||||
return {"success": True, "exit_code": 0}
|
||||
|
||||
client, thread, palace = _start_server(tmp_path, monkeypatch, fake_execute)
|
||||
try:
|
||||
job = client.submit(
|
||||
"mine", {"source": "src", "palace_path": "/tmp/other-palace"}, dedupe_key="p"
|
||||
)
|
||||
client.wait(job["id"], timeout=5)
|
||||
finally:
|
||||
client.shutdown()
|
||||
thread.join(timeout=5)
|
||||
assert seen["palace_path"] == daemon.canonical_palace_path(str(palace))
|
||||
assert seen["palace_path"] != "/tmp/other-palace"
|
||||
|
||||
|
||||
def test_mcp_tool_allowlist_rejects_non_write_tools(tmp_path, monkeypatch):
|
||||
"""The daemon queue is a durable write surface; read/maintenance/unknown
|
||||
tools must be rejected so verbatim content can't be exfiltrated into the
|
||||
queue or retried destructively."""
|
||||
# read tool → rejected
|
||||
out = service.run_mcp_tool({"name": "mempalace_search", "arguments": {}})
|
||||
assert out["success"] is False
|
||||
assert "only accepts write tools" in out["error"]
|
||||
# maintenance tool → rejected (has its own kinds: mine/sync)
|
||||
out = service.run_mcp_tool({"name": "mempalace_mine", "arguments": {}})
|
||||
assert out["success"] is False
|
||||
# unknown tool → rejected
|
||||
out = service.run_mcp_tool({"name": "mempalace_bogus", "arguments": {}})
|
||||
assert out["success"] is False
|
||||
# write tool → passes the allowlist (handler not called here since TOOLS
|
||||
# won't have it under the test name; but classification must let it through)
|
||||
assert service.classify_tool("mempalace_add_drawer") == "write"
|
||||
|
||||
|
||||
def test_execute_job_isolates_env_per_job(monkeypatch):
|
||||
"""A job that mutates MEMPALACE_BACKEND must not leak into the next job's
|
||||
env. Regression for the per-job isolation bug (_apply_backend poisoning)."""
|
||||
import os as _os
|
||||
|
||||
monkeypatch.delenv("MEMPALACE_BACKEND", raising=False)
|
||||
monkeypatch.delenv("MEMPALACE_PALACE_PATH", raising=False)
|
||||
|
||||
def fake_mine(payload):
|
||||
_os.environ["MEMPALACE_BACKEND"] = "leaked-backend"
|
||||
return {"success": True, "exit_code": 0}
|
||||
|
||||
monkeypatch.setattr(service, "run_mine", fake_mine)
|
||||
service.execute_job("mine", {"palace_path": "/tmp/p", "source": "s"})
|
||||
assert _os.environ.get("MEMPALACE_BACKEND") is None
|
||||
|
||||
|
||||
def test_daemon_client_raises_on_endpoint_missing_port(tmp_path, monkeypatch):
|
||||
"""A malformed endpoint.json must raise DaemonError, not a bare KeyError."""
|
||||
import json as _json
|
||||
|
||||
monkeypatch.setenv(daemon.STATE_ROOT_ENV, str(tmp_path / "state"))
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
daemon.ensure_token(str(palace))
|
||||
# endpoint with no port
|
||||
daemon.state_dir(str(palace)).mkdir(parents=True, exist_ok=True)
|
||||
(daemon.state_dir(str(palace)) / "endpoint.json").write_text(
|
||||
_json.dumps({"host": "127.0.0.1", "pid": 1}) + "\n", encoding="utf-8"
|
||||
)
|
||||
|
||||
with pytest.raises(daemon.DaemonError):
|
||||
daemon.DaemonClient(str(palace))
|
||||
|
||||
|
||||
def test_start_daemon_kills_orphan_on_readiness_timeout(tmp_path, monkeypatch):
|
||||
"""If the spawned daemon never becomes ready, start_daemon must kill and
|
||||
reap the orphaned subprocess rather than leaking it with the port/token."""
|
||||
monkeypatch.setenv(daemon.STATE_ROOT_ENV, str(tmp_path / "state"))
|
||||
palace = tmp_path / "palace"
|
||||
palace.mkdir()
|
||||
daemon.ensure_token(str(palace))
|
||||
|
||||
monkeypatch.setattr(daemon, "get_client_if_running", lambda *a, **kw: None)
|
||||
|
||||
class FakeProc:
|
||||
def __init__(self):
|
||||
self.killed = False
|
||||
self.returncode = None
|
||||
|
||||
def poll(self):
|
||||
return self.returncode # None == still alive
|
||||
|
||||
def kill(self):
|
||||
self.killed = True
|
||||
|
||||
def wait(self):
|
||||
self.returncode = -9
|
||||
return self.returncode
|
||||
|
||||
fake = FakeProc()
|
||||
|
||||
def fake_popen(*a, **kw):
|
||||
return fake
|
||||
|
||||
monkeypatch.setattr(daemon.subprocess, "Popen", fake_popen)
|
||||
monkeypatch.setattr(daemon, "DaemonClient", _raise_not_ready)
|
||||
monkeypatch.setattr(daemon.time, "sleep", lambda *a, **kw: None)
|
||||
|
||||
with pytest.raises(daemon.DaemonError):
|
||||
daemon.start_daemon(str(palace), timeout=0.05)
|
||||
assert fake.killed is True
|
||||
|
|
@ -16,6 +16,7 @@ from mempalace.hooks_cli import (
|
|||
_diary_agent_for_harness,
|
||||
_extract_recent_messages,
|
||||
_get_mine_targets,
|
||||
_hooks_daemon_enabled,
|
||||
_log,
|
||||
_maybe_auto_ingest,
|
||||
_mempalace_python,
|
||||
|
|
@ -467,6 +468,45 @@ def test_stop_hook_checkpoint_visible_to_diary_read(monkeypatch, config, palace_
|
|||
assert legacy.get("entries") == []
|
||||
|
||||
|
||||
def test_save_diary_direct_daemon_opt_in_submits_job(tmp_path):
|
||||
transcript = tmp_path / "session.jsonl"
|
||||
palace_dir = tmp_path / "palace"
|
||||
palace_dir.mkdir()
|
||||
_write_transcript(
|
||||
transcript,
|
||||
[{"message": {"role": "user", "content": f"message {i}"}} for i in range(3)],
|
||||
)
|
||||
env = {"MEMPALACE_HOOKS_DAEMON": "yes", "MEMPALACE_PALACE_PATH": str(palace_dir)}
|
||||
job = {"id": "job", "state": "succeeded", "result": {"success": True, "entry_id": "e1"}}
|
||||
|
||||
with patch.dict("os.environ", env):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
with patch("mempalace.hooks_cli._daemon_available", return_value=True):
|
||||
with patch("mempalace.daemon.submit_job", return_value=job) as mock_submit:
|
||||
result = _save_diary_direct(
|
||||
str(transcript),
|
||||
"sess1",
|
||||
wing="wing_project",
|
||||
agent_name="claude",
|
||||
)
|
||||
|
||||
assert result["count"] == 3
|
||||
mock_submit.assert_called_once()
|
||||
assert mock_submit.call_args.args[0] == "diary_write"
|
||||
payload = mock_submit.call_args.args[1]
|
||||
assert payload["agent_name"] == "claude"
|
||||
assert payload["wing"] == "wing_project"
|
||||
assert payload["topic"] == "checkpoint"
|
||||
assert (tmp_path / "last_checkpoint").exists()
|
||||
|
||||
|
||||
def test_hooks_daemon_enabled_requires_explicit_true():
|
||||
with patch("mempalace.hooks_cli.MempalaceConfig") as mock_cfg_cls:
|
||||
assert _hooks_daemon_enabled() is False
|
||||
mock_cfg_cls.return_value.hook_use_daemon = True
|
||||
assert _hooks_daemon_enabled() is True
|
||||
|
||||
|
||||
# --- hook_session_start ---
|
||||
|
||||
|
||||
|
|
@ -788,6 +828,33 @@ def test_maybe_auto_ingest_with_env(tmp_path):
|
|||
assert cmd[cmd.index("--mode") + 1] == "projects"
|
||||
|
||||
|
||||
def test_maybe_auto_ingest_daemon_opt_in_submits_job(tmp_path):
|
||||
"""Daemon-enabled hooks submit a background mine instead of spawning one."""
|
||||
mempal_dir = tmp_path / "project"
|
||||
palace_dir = tmp_path / "palace"
|
||||
mempal_dir.mkdir()
|
||||
palace_dir.mkdir()
|
||||
env = {
|
||||
"MEMPAL_DIR": str(mempal_dir),
|
||||
"MEMPALACE_HOOKS_DAEMON": "yes",
|
||||
"MEMPALACE_PALACE_PATH": str(palace_dir),
|
||||
}
|
||||
with patch.dict("os.environ", env):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
with patch("mempalace.hooks_cli._daemon_available", return_value=True):
|
||||
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
||||
with patch(
|
||||
"mempalace.daemon.submit_job", return_value={"id": "job"}
|
||||
) as mock_submit:
|
||||
_maybe_auto_ingest()
|
||||
|
||||
mock_popen.assert_not_called()
|
||||
mock_submit.assert_called_once()
|
||||
assert mock_submit.call_args.args[0] == "mine"
|
||||
assert mock_submit.call_args.args[1]["source"] == str(mempal_dir.resolve())
|
||||
assert mock_submit.call_args.kwargs["wait"] is False
|
||||
|
||||
|
||||
def test_maybe_auto_ingest_uses_mempalace_python(tmp_path):
|
||||
"""Spawned mine command uses _mempalace_python(), not bare sys.executable.
|
||||
|
||||
|
|
@ -1114,6 +1181,31 @@ def test_ingest_transcript_uses_detached_kwargs(tmp_path):
|
|||
assert kwargs.get("close_fds") is True
|
||||
|
||||
|
||||
def test_ingest_transcript_daemon_opt_in_submits_job(tmp_path):
|
||||
transcript = tmp_path / "session.jsonl"
|
||||
palace_dir = tmp_path / "palace"
|
||||
palace_dir.mkdir()
|
||||
transcript.write_text("x" * 200)
|
||||
env = {"MEMPALACE_HOOKS_DAEMON": "yes", "MEMPALACE_PALACE_PATH": str(palace_dir)}
|
||||
with patch.dict("os.environ", env):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
with patch("mempalace.hooks_cli._daemon_available", return_value=True):
|
||||
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
||||
with patch(
|
||||
"mempalace.daemon.submit_job", return_value={"id": "job"}
|
||||
) as mock_submit:
|
||||
from mempalace.hooks_cli import _ingest_transcript
|
||||
|
||||
_ingest_transcript(str(transcript))
|
||||
|
||||
mock_popen.assert_not_called()
|
||||
mock_submit.assert_called_once()
|
||||
payload = mock_submit.call_args.args[1]
|
||||
assert payload["source"] == str(tmp_path)
|
||||
assert payload["mode"] == "convos"
|
||||
assert payload["wing"] == "sessions"
|
||||
|
||||
|
||||
def test_ingest_transcript_skips_when_target_running(tmp_path):
|
||||
"""Repeated transcript ingests for the same transcript should dedup."""
|
||||
transcript = tmp_path / "session.jsonl"
|
||||
|
|
|
|||
|
|
@ -11,6 +11,12 @@ from pathlib import Path
|
|||
import chromadb
|
||||
import pytest
|
||||
|
||||
# run_sync imports mempalace.mcp_server lazily; that import initializes the
|
||||
# embedder, which rebinds sys.stdout and defeats capsys/redirect_stdout for any
|
||||
# prints after sync_palace returns. Importing it here makes the lazy import a
|
||||
# cached no-op so the daemon-path report tests can capture run_sync's output.
|
||||
import mempalace.mcp_server # noqa: F401
|
||||
|
||||
|
||||
def _seed_drawers(palace_path, repo_path, deleted_path, elsewhere_path):
|
||||
"""Populate the drawers collection with 6 entries covering all buckets."""
|
||||
|
|
@ -1397,3 +1403,81 @@ class TestSyncCli:
|
|||
with pytest.raises(SystemExit) as exc_info:
|
||||
cli.main()
|
||||
assert exc_info.value.code == 2
|
||||
|
||||
|
||||
class TestServiceRunSyncReport:
|
||||
"""Daemon path (service.run_sync) must render the same report shape as the
|
||||
direct CLI path, with no KeyError on report['deleted'] (regression: the old
|
||||
code read a non-existent 'deleted' key and dropped no_source/out_of_scope/
|
||||
by_source and the Re-run/Removed hints).
|
||||
|
||||
sync_palace is mocked so the test exercises only run_sync's report
|
||||
formatting — opening the real Chroma collection reinitializes the embedder,
|
||||
which disturbs sys.stdout and defeats capsys.
|
||||
"""
|
||||
|
||||
def _fake_report(self, **overrides):
|
||||
report = {
|
||||
"scanned": 6,
|
||||
"kept": 1,
|
||||
"gitignored": 2,
|
||||
"missing": 1,
|
||||
"no_source": 1,
|
||||
"out_of_scope": 1,
|
||||
"removed_drawers": 0,
|
||||
"removed_closets": 0,
|
||||
"dry_run": True,
|
||||
"by_source": {"src/a.py": 2, "src/b.py": 1},
|
||||
}
|
||||
report.update(overrides)
|
||||
return report
|
||||
|
||||
def test_dry_run_renders_full_report(self, monkeypatch, tmp_dir, capsys):
|
||||
import mempalace.sync as sync_module
|
||||
from mempalace import service
|
||||
|
||||
palace = os.path.join(tmp_dir, "palace")
|
||||
os.makedirs(palace)
|
||||
# Satisfy run_sync's detect_backend_for_path guard without spinning up
|
||||
# the real Chroma/embedder stack (which would disturb sys.stdout).
|
||||
Path(palace, "chroma.sqlite3").touch()
|
||||
monkeypatch.setattr(
|
||||
sync_module,
|
||||
"sync_palace",
|
||||
lambda **kw: self._fake_report(dry_run=True),
|
||||
)
|
||||
result = service.run_sync({"palace_path": palace, "dir": tmp_dir, "dry_run": True})
|
||||
assert result["success"] is True
|
||||
out = capsys.readouterr().out
|
||||
# The fields the stripped daemon report used to drop.
|
||||
assert "No source:" in out
|
||||
assert "Out of scope:" in out
|
||||
# by_source top sources block.
|
||||
assert "Top sources to remove" in out
|
||||
assert "src/a.py (2)" in out
|
||||
# Re-run hint fires when there is something to remove.
|
||||
assert "Re-run with --apply" in out
|
||||
# The old KeyError line must not be present.
|
||||
assert "Deleted:" not in out
|
||||
|
||||
def test_apply_renders_removed_counts(self, monkeypatch, tmp_dir, capsys):
|
||||
import mempalace.sync as sync_module
|
||||
from mempalace import service
|
||||
|
||||
palace = os.path.join(tmp_dir, "palace")
|
||||
os.makedirs(palace)
|
||||
Path(palace, "chroma.sqlite3").touch()
|
||||
monkeypatch.setattr(
|
||||
sync_module,
|
||||
"sync_palace",
|
||||
lambda **kw: self._fake_report(
|
||||
dry_run=False, removed_drawers=3, removed_closets=2, by_source={"src/a.py": 3}
|
||||
),
|
||||
)
|
||||
result = service.run_sync({"palace_path": palace, "dir": tmp_dir, "dry_run": False})
|
||||
assert result["success"] is True
|
||||
out = capsys.readouterr().out
|
||||
# Apply mode prints the removed-drawers/closets line, not the Re-run hint.
|
||||
assert "Removed 3 drawers, 2 closets" in out
|
||||
assert "Top sources removed" in out
|
||||
assert "Re-run with --apply" not in out
|
||||
|
|
|
|||
Loading…
Reference in New Issue