feat(cli): apply daemon write-routing policy

Author: fatkobra

docs/cli-write-routing.md

@@ -0,0 +1,107 @@
+# CLI write routing
+
+Routine CLI writes consume the shared routing policy introduced for the Tier 3
+rollout tracked in #1963.
+
+## Routed commands
+
+The policy applies to:
+
+- `mempalace mine`;
+- `mempalace sweep`;
+- `mempalace sync`;
+- the optional post-setup mine run by `mempalace init`.
+
+## Policies
+
+### `direct`
+
+Use the existing direct in-process execution path.
+
+### `prefer`
+
+Submit through the local daemon. Interactive CLI commands are allowed to start
+the daemon when it is not already running.
+
+### `require`
+
+Submit through the local daemon. Interactive CLI commands are allowed to start
+the daemon when it is not already running.
+
+For CLI commands, `prefer` and `require` both select the daemon because daemon
+startup is permitted. Their difference remains meaningful to hook callers,
+which cannot cold-start the daemon.
+
+## Explicit flags
+
+Force daemon execution:
+
+    mempalace mine ./project --daemon
+
+Force direct execution:
+
+    mempalace mine ./project --direct
+
+The flags are mutually exclusive and override environment/config policy.
+
+Background execution:
+
+    mempalace mine ./project --background
+
+`--background` is valid when the selected route is the daemon. A direct route
+with `--background` exits with a configuration error.
+
+## Configuration
+
+Use the daemon for routine CLI writes:
+
+    MEMPALACE_CLI_WRITE_ROUTING=prefer
+
+Prohibit an accidental direct route:
+
+    MEMPALACE_CLI_WRITE_ROUTING=require
+
+Retain direct behavior:
+
+    MEMPALACE_CLI_WRITE_ROUTING=direct
+
+Config file:
+
+    {
+      "write_routing": {
+        "cli": "require"
+      }
+    }
+
+## Defaults and rollout
+
+The default remains `direct` in this PR.
+
+This means existing users receive no silent execution-topology change. A
+supervised Tier 3 deployment enables `prefer` or `require` explicitly. The
+production default can be changed later after the stacked rollout is reviewed.
+
+## Safety properties
+
+- Daemon submission errors never trigger direct fallback.
+- A failed or ambiguous submission may already have created a durable job;
+  retrying directly could duplicate content.
+- Post-init mining forwards the already-scanned file list into the daemon, so
+  the project is not scanned twice.
+- `sweep` is now a first-class daemon job.
+- `--direct` remains an explicit emergency/debug escape hatch.
+- No low-level lock behavior is changed.
+
+## Maintenance exclusions
+
+These remain outside ordinary routing:
+
+- repair;
+- migration and wing migration;
+- index rebuild;
+- closet compression;
+- embedder identity changes.
+
+They can replace indexes, rewrite broad metadata sets, or require all cached
+handles to close. They need an exclusive-maintenance protocol rather than an
+ordinary queued-write job.

docs/write-routing-policy.md

@@ -147,10 +147,11 @@ operations offline, with no writable service running.
 
 ## Follow-up PRs
 
-Hook-triggered writes now consume this policy; see
+Hook-triggered writes consume this policy; see
 `docs/hook-write-routing.md`.
 
-The remaining rollout PR will apply the policy to routine CLI writes.
+Routine CLI writes also consume this policy; see
+`docs/cli-write-routing.md`.
 
 Maintenance operations such as repair, migration, and index rebuild are not
 ordinary routed writes. They require a separate exclusive-maintenance policy.

mempalace/cli.py

@@ -40,6 +40,11 @@
 from pathlib import Path
 
 from .config import MempalaceConfig
+from .cli_write_routing import (
+    add_cli_write_routing_flags,
+    resolve_cli_write_routing,
+)
+from .write_routing import WriteRoutingError
 from .corpus_origin import detect_origin_heuristic, detect_origin_llm
 from .llm_client import LLMError, get_provider
 from .version import __version__
@@ -102,6 +107,21 @@ def _maintenance_requires_chroma(palace_path: str, command_name: str) -> bool:
     return False
 
 
+def _resolve_cli_write_routing_or_exit(args, operation: str):
+    """Resolve routine CLI routing and render configuration errors."""
+    try:
+        return resolve_cli_write_routing(
+            args,
+            operation=operation,
+        )
+    except WriteRoutingError as exc:
+        print(
+            f"mempalace: invalid CLI write routing: {exc}",
+            file=sys.stderr,
+        )
+        raise SystemExit(2) from exc
+
+
 def _gather_origin_samples(project_dir) -> list:
     """Collect Tier-1 samples for corpus-origin detection.
 
@@ -565,6 +585,39 @@ def _maybe_run_mine_after_init(args, cfg) -> None:
             return
 
     palace_path = cfg.palace_path
+    routing = _resolve_cli_write_routing_or_exit(
+        args,
+        "init auto-mine",
+    )
+
+    if routing.use_daemon:
+        payload = {
+            "source": project_dir,
+            "mode": "projects",
+            "wing": None,
+            "agent": "mempalace",
+            "limit": 0,
+            "dry_run": False,
+            "extract": "exchange",
+            "no_gitignore": False,
+            "include_ignored": [],
+            "max_chunks_per_file": None,
+            "redetect_origin": False,
+            "files": (
+                [str(file_path) for file_path in scanned_files]
+                if scanned_files is not None
+                else None
+            ),
+        }
+        _submit_daemon_cli_job(
+            "mine",
+            payload,
+            args,
+            background=False,
+            auto_start=routing.decision.auto_start_daemon,
+        )
+        return
+
     try:
         mine(
             project_dir=project_dir,
@@ -722,27 +775,43 @@ 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)
+    payload = {
+        "source": args.dir,
+        "mode": 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,
+        ),
+    }
 
-    if getattr(args, "daemon", False):
-        payload = {
-            "source": args.dir,
-            "mode": 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),
-        }
-        if source_adapter:
-            payload["source_adapter"] = source_adapter
-        _submit_daemon_cli_job("mine", payload, args, background=getattr(args, "background", False))
+    if source_adapter:
+        payload["source_adapter"] = source_adapter
+
+    routing = _resolve_cli_write_routing_or_exit(
+        args,
+        "mine",
+    )
+    if routing.use_daemon:
+        _submit_daemon_cli_job(
+            "mine",
+            payload,
+            args,
+            background=bool(getattr(args, "background", False)),
+            auto_start=routing.decision.auto_start_daemon,
+        )
         return
 
     from .palace import MineAlreadyRunning, MineValidationError
@@ -1026,6 +1095,19 @@ def cmd_sweep(args):
 
     palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path
     target = os.path.expanduser(args.target)
+    routing = _resolve_cli_write_routing_or_exit(
+        args,
+        "sweep",
+    )
+    if routing.use_daemon:
+        _submit_daemon_cli_job(
+            "sweep",
+            {"target": target},
+            args,
+            background=bool(getattr(args, "background", False)),
+            auto_start=routing.decision.auto_start_daemon,
+        )
+        return
 
     if os.path.isfile(target):
         result = sweep(target, palace_path)
@@ -1058,18 +1140,25 @@ 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)
+    payload = {
+        "dir": args.dir,
+        "root": list(args.root or []),
+        "wing": args.wing,
+        "dry_run": args.dry_run,
+    }
 
-    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))
+    routing = _resolve_cli_write_routing_or_exit(
+        args,
+        "sync",
+    )
+    if routing.use_daemon:
+        _submit_daemon_cli_job(
+            "sync",
+            payload,
+            args,
+            background=bool(getattr(args, "background", False)),
+            auto_start=routing.decision.auto_start_daemon,
+        )
         return
 
     from .palace import MineAlreadyRunning
@@ -1160,7 +1249,14 @@ def cmd_sync(args):
     print(f"\n{'=' * 55}\n")
 
 
-def _submit_daemon_cli_job(kind: str, payload: dict, args, *, background: bool) -> None:
+def _submit_daemon_cli_job(
+    kind: str,
+    payload: dict,
+    args,
+    *,
+    background: bool,
+    auto_start: bool = True,
+) -> 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
@@ -1172,7 +1268,7 @@ def _submit_daemon_cli_job(kind: str, payload: dict, args, *, background: bool)
             palace_path=palace_path,
             backend=backend,
             wait=not background,
-            auto_start=True,
+            auto_start=auto_start,
             # A job refused the palace lock is deferred, not failed (#2014), so
             # it never becomes terminal while the holder lives. Waiting it out
             # would strand this terminal behind a peer that can outlive the
@@ -2520,6 +2616,11 @@ def main():
         ),
     )
 
+    add_cli_write_routing_flags(
+        p_init,
+        allow_background=False,
+    )
+
     # mine
     p_mine = sub.add_parser("mine", help="Mine files into the palace")
     p_mine.add_argument(
@@ -2582,16 +2683,7 @@ 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",
-    )
+    add_cli_write_routing_flags(p_mine)
     p_mine.add_argument(
         "--extract",
         choices=["exchange", "general"],
@@ -2634,6 +2726,8 @@ def main():
         help="A .jsonl transcript file, or a directory to scan recursively",
     )
 
+    add_cli_write_routing_flags(p_sweep)
+
     # sync
     p_sync = sub.add_parser(
         "sync",
@@ -2665,16 +2759,7 @@ 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",
-    )
+    add_cli_write_routing_flags(p_sync)
 
     # search
     p_search = sub.add_parser("search", help="Find anything, exact words")