ISS-10812: route each session's calls to its own worktree (PLN-2070 PR 2)

integrations/claude/plugin/hooks/mcp_read_allow.py

@@ -30,13 +30,24 @@
 user's own deny rule for something outside this list.
 
 Fail-open; opt-out via LEMONCROW_MCP_READ_ALLOW=0.
+
+It also records the session's working directory for every lc call, read-only or
+not and whatever the opt-out says: ``<store root>/session_cwd/<session id>``
+holds the payload's ``cwd``, which follows ``EnterWorktree``. The shared MCP
+daemon reads it to route the session's calls to the worktree it works in
+(``lemoncrow.gateway.adapters.mcp.session_root``). The record never changes
+this hook's output and never raises; SessionStart prunes records older than
+7 days.
 """
 
 from __future__ import annotations
 
 import json
 import os
+import re
 import sys
+import tempfile
+from pathlib import Path
 from typing import Any
 
 # MCP server key as registered by install_claude.sh (user scope: "lc",
@@ -122,15 +133,61 @@ def _allow(reason: str) -> None:
     )
 
 
+# A session id names a file, so it must never carry a path separator or a dot.
+_SESSION_ID_RE = re.compile(r"[A-Za-z0-9_-]{1,128}")
+
+
+def _session_cwd_dir() -> Path:
+    """``<store root>/session_cwd``, the store root resolved as ``lemoncrow`` resolves it."""
+    configured = os.environ.get("LEMONCROW_ROOT", "").strip()
+    root = Path(configured).expanduser() if configured else Path.home() / ".lemoncrow"
+    return root / "session_cwd"
+
+
+def _record_session_cwd(payload: dict[str, Any]) -> None:
+    """Write the payload's cwd to its session's file when it changed; never raise."""
+    session_id = payload.get("session_id")
+    cwd = payload.get("cwd")
+    if not isinstance(session_id, str) or not _SESSION_ID_RE.fullmatch(session_id):
+        return
+    if not isinstance(cwd, str) or not cwd.strip():
+        return
+    try:
+        directory = _session_cwd_dir()
+        if not directory.is_dir():
+            directory.mkdir(parents=True, exist_ok=True)
+            os.chmod(directory, 0o700)
+        target = directory / session_id
+        try:
+            if target.read_text(encoding="utf-8") == cwd:
+                return
+        except (OSError, UnicodeDecodeError):
+            pass
+        fd, tmp = tempfile.mkstemp(dir=directory, prefix=f".{session_id}.", suffix=".tmp")
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as handle:
+                handle.write(cwd)
+            os.replace(tmp, target)
+        except BaseException:
+            try:
+                os.unlink(tmp)
+            except OSError:
+                pass
+            raise
+    except Exception as exc:  # the permission decision must survive any failure here
+        print(f"lc: session cwd not recorded: {exc}", file=sys.stderr)
+
+
 def main() -> int:
-    if os.environ.get("LEMONCROW_MCP_READ_ALLOW", "1") == "0":
-        return 0
     try:
         payload: dict[str, Any] = json.loads(sys.stdin.read() or "{}")
     except (json.JSONDecodeError, TypeError, OSError):
         return 0
     if not isinstance(payload, dict):
         return 0
+    _record_session_cwd(payload)
+    if os.environ.get("LEMONCROW_MCP_READ_ALLOW", "1") == "0":
+        return 0
     tool = _read_only_tool(str(payload.get("tool_name") or ""), payload.get("tool_input"))
     if tool is None:
         return 0

integrations/claude/plugin/hooks/session_start.py

@@ -27,6 +27,7 @@
 import re
 import sys
 import tempfile
+import time
 from contextlib import suppress
 from pathlib import Path
 from typing import Any
@@ -85,6 +86,29 @@ def _lemoncrow_root() -> Path:
     return Path.home() / ".lemoncrow"
 
 
+_SESSION_CWD_MAX_AGE_S = 7 * 24 * 3600
+
+
+def _prune_session_cwds(now: float | None = None) -> None:
+    """Delete session cwd records untouched for 7 days.
+
+    ``mcp_read_allow.py`` writes one per session, under the same store root, and
+    rewrites it only when the cwd changes; a live session whose record is pruned
+    gets it back on its next lc call.
+    """
+    configured = os.environ.get("LEMONCROW_ROOT", "").strip()
+    root = Path(configured).expanduser() if configured else Path.home() / ".lemoncrow"
+    cutoff = (time.time() if now is None else now) - _SESSION_CWD_MAX_AGE_S
+    try:
+        entries = list(os.scandir(root / "session_cwd"))
+    except OSError:
+        return
+    for entry in entries:
+        with suppress(OSError):
+            if entry.is_file(follow_symlinks=False) and entry.stat(follow_symlinks=False).st_mtime < cutoff:
+                os.unlink(entry.path)
+
+
 def _active_session_id() -> str | None:
     state = _read_session_state()
     return state.get("session_id") or state.get("active_session_id")
@@ -263,6 +287,9 @@ def main() -> int:
         with suppress(Exception):
             _emit_env_context(payload["cwd"])
 
+    with suppress(Exception):
+        _prune_session_cwds()
+
     session_id_raw: str = payload.get("session_id", "") or ""
     source: str = payload.get("source", "startup") or "startup"
     model: str = payload.get("model", "") or ""

pyproject.toml

@@ -164,6 +164,7 @@ pythonpath = ["src", "."]
 markers = [
     "slow: marks tests as slow",
     "ab: real A/B benchmark for MCP tool savings",
+    "real_code_autosync: keep CodeContextEngine autosync on (opts out of tests/gateway/conftest.py's forced-off patch)",
     ]
 
 [tool.mypy]

src/lemoncrow/gateway/adapters/mcp/bash.py

@@ -21,6 +21,7 @@
 from pathlib import Path
 from typing import Any, Literal, cast
 
+from lemoncrow.gateway.adapters.mcp import session_root
 from lemoncrow.gateway.adapters.mcp.deferral import (
     _defer_bash_enabled,
     _deferral_supported,
@@ -442,15 +443,8 @@ def _render_grep_stdout(payload: dict[str, Any]) -> str:
             return json.dumps(matches, ensure_ascii=False)
         return json.dumps(payload, ensure_ascii=False)
 
-    workspace = os.environ.get("CLAUDE_WORKSPACE_ROOT", os.getcwd())
-    # A misconfigured CLAUDE_WORKSPACE_ROOT -- e.g. a host path that leaked into
-    # a container via the environment and does not exist here -- would make every
-    # cwd-less command fail with a raw FileNotFoundError from Popen, surfaced as
-    # an opaque "MCP error -32000". Fall back to the process cwd (always a real
-    # directory) so the command still runs instead of hard-failing.
-    if not Path(workspace).is_dir():
-        workspace = os.getcwd()
-    effective_cwd = cwd or workspace
+    workspace = _bash_workspace()
+    effective_cwd = cwd or _cwd_less_run_dir(workspace)
 
     if action in {"poll", "kill", "status", "update", "send"}:
         if not session_id:
@@ -1064,6 +1058,25 @@ def _render_bash_text(result: dict[str, Any]) -> str:
     return ""
 
 
+def _bash_workspace() -> str:
+    """The workspace root ``bash`` works under: CLAUDE_WORKSPACE_ROOT, else the process cwd.
+
+    A misconfigured CLAUDE_WORKSPACE_ROOT -- e.g. a host path that leaked into a
+    container via the environment and does not exist here -- would make every
+    cwd-less command fail with a raw FileNotFoundError from Popen, surfaced as an
+    opaque "MCP error -32000". It falls back to the process cwd (always a real
+    directory) so the command still runs instead of hard-failing.
+    """
+    workspace = os.environ.get("CLAUDE_WORKSPACE_ROOT", os.getcwd())
+    return workspace if Path(workspace).is_dir() else os.getcwd()
+
+
+def _cwd_less_run_dir(workspace: str) -> str:
+    """Where a command with no ``cwd`` runs: the session's linked worktree, else *workspace*."""
+    worktree = session_root.session_worktree(Path(workspace))
+    return workspace if worktree is None else str(worktree)
+
+
 def _lift_bash_command_list(args: dict[str, Any], known_params: frozenset[str]) -> dict[str, Any]:
     """Recover list-for-string args: command=[...] -> commands, id=[...] -> ids."""
     cmd = args.get("command")
@@ -1255,7 +1268,7 @@ def tool_bash(
         # bash(id=x) with no explicit action = wait for the run to finish.
         action = "poll"
     if action == "run" and command and not bg and not interactive:
-        _dump_notice = _check_redundant_file_dump(command, cwd)
+        _dump_notice = _check_redundant_file_dump(command, cwd or _cwd_less_run_dir(_bash_workspace()))
         if _dump_notice:
             return _dump_notice
     result = _run_bash_tool(