Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
213 changes: 187 additions & 26 deletions src/lemoncrow/gateway/adapters/mcp_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -3772,23 +3772,19 @@ def _record_session_cwd(name: str, args: Any) -> None:
_last_session_cwd = cwd.strip()


def _session_worktree_root(workspace_root: Path) -> Path | None:
"""The linked git worktree the session is working in, else None.
def _linked_worktree_root(workspace_root: Path, candidate_dir: Path) -> Path | None:
"""The linked worktree of ``workspace_root`` that contains ``candidate_dir``, else None.

Detected without spawning git: a linked worktree's ``.git`` is a *file*
holding ``gitdir: <path>``, and for a worktree of THIS repo that path lives
under ``<workspace_root>/.git/worktrees/``. A normal checkout has ``.git``
as a directory, which ends the walk immediately.

Returns None for every uncertain case -- no recorded cwd, a plain
directory, a worktree belonging to a different repo -- so resolution falls
back to the workspace root exactly as before.
Returns None for every uncertain case -- a plain directory, a worktree
belonging to a different repo, the workspace root itself.
"""
recorded = _last_session_cwd
if not recorded:
return None
try:
candidate = Path(recorded).expanduser().resolve()
candidate = candidate_dir.expanduser().resolve()
if not candidate.is_dir():
return None
root = workspace_root.resolve()
Expand All @@ -3811,6 +3807,19 @@ def _session_worktree_root(workspace_root: Path) -> Path | None:
return None


def _session_worktree_root(workspace_root: Path) -> Path | None:
"""The linked git worktree the session is working in, else None.

Returns None for every uncertain case -- no recorded cwd, a plain
directory, a worktree belonging to a different repo -- so resolution falls
back to the workspace root exactly as before.
"""
recorded = _last_session_cwd
if not recorded:
return None
return _linked_worktree_root(workspace_root, Path(recorded))


# Thread-local slot for passing real tokens_saved from tool handlers to the
# budget recorder without polluting the LLM-facing response dict.
# _tool_call_tokens_saved moved to mcp.smart_state (imported/re-exported).
Expand Down Expand Up @@ -5530,7 +5539,7 @@ def render_tool_result_text(name: str, result: Any) -> str | None:
# `resolved_against` rides as a one-liner suffix like vcs_status rather
# than tripping the JSON fallback -- a redirected edit is still a clean
# success and should not render as a structured dump.
base_keys = keys - {"vcs_status", "resolved_against"}
base_keys = keys - {"vcs_status", "resolved_against", "resolved_against_source"}
if base_keys <= {"calls_saved"}:
text = "ok"
elif base_keys <= {"applied", "calls_saved"}:
Expand All @@ -5542,7 +5551,10 @@ def render_tool_result_text(name: str, result: Any) -> str | None:
if text:
resolved_against = payload.get("resolved_against")
if isinstance(resolved_against, str) and resolved_against:
text = f"{text} | resolved against worktree {resolved_against} (from last bash cwd)"
if payload.get("resolved_against_source") == "explicit":
text = f"{text} | resolved against root {resolved_against} (explicit root argument)"
else:
text = f"{text} | resolved against worktree {resolved_against} (from last bash cwd)"
vcs_raw = payload.get("vcs_status")
vcs = vcs_raw if isinstance(vcs_raw, dict) else {}
vcs_lines = vcs.get("lines")
Expand Down Expand Up @@ -6894,6 +6906,71 @@ def _collect_touched_paths(edits: list[dict[str, Any]], *, repo_root: str | Path
return dict(sorted(paths.items()))


# Scratch directories writes are allowed into besides the workspace and any
# opted-in additional directory: staging a file before moving it in is ordinary
# tool work, and refusing it would break callers that predate `root=`. One
# literal covers macOS's /tmp -> /private/tmp symlink, because every root is
# resolved before it is compared.
_SCRATCH_EDIT_ROOTS: tuple[Path, ...] = (Path("/tmp"),)


def _resolve_explicit_edit_root(raw_root: str, *, workspace_root: Path, extra_roots: list[Path]) -> Path | None:
"""Validate an explicit ``root=`` for edit resolution; None when out of bounds.

Accepts the workspace root (and anything under it), a linked worktree of it
-- worktrees often live outside the repo directory -- and any directory
already opted in for writes. Anything else is refused rather than honored:
a root naming a foreign checkout would misdirect writes exactly the way the
inference this argument exists to override can.
"""
candidate = Path(raw_root).expanduser()
if not candidate.is_absolute():
candidate = workspace_root / candidate
try:
candidate = candidate.resolve()
if not candidate.is_dir():
return None
except OSError:
return None
root = workspace_root.resolve()
if candidate == root or candidate.is_relative_to(root):
return candidate
if _linked_worktree_root(workspace_root, candidate) is not None:
return candidate
# Resolved on both sides: an additional directory reached through a symlink
# (or the /tmp literal on macOS) contains nothing lexically.
if any(candidate == r or candidate.is_relative_to(r) for r in (extra.resolve() for extra in extra_roots)):
return candidate
return None


def _ambiguous_relative_edit_paths(
edits: list[dict[str, Any]], *, inferred_root: Path, workspace_root: Path
) -> list[tuple[str, list[Path]]]:
"""Relative edit paths naming an existing file under more than one root.

Returns ``(raw_path, [existing candidate, ...])`` for each edit whose
relative path resolves onto a real file in both the inferred worktree and
the workspace root. A path that exists in one root is unambiguous, and one
that exists in neither is a create -- neither is reported.
"""
roots = [inferred_root.resolve(), workspace_root.resolve()]
if roots[0] == roots[1]:
return []
ambiguous: list[tuple[str, list[Path]]] = []
for edit in edits:
raw = str(edit.get("file_path") or edit.get("path") or "")
if not raw:
continue
candidate = Path(_snapshot_path(raw))
if candidate.is_absolute():
continue
hits = [root / candidate for root in roots if (root / candidate).is_file()]
if len(hits) > 1:
ambiguous.append((raw, hits))
return ambiguous


def _snapshot_paths(paths: dict[str, Path]) -> dict[str, tuple[Path, bool, str | None]]:
"""Snapshot each file's pre-edit state for rollback.

Expand Down Expand Up @@ -7397,6 +7474,13 @@ def _attach_contract_literal_review(
},
},
},
"root": {
"type": "string",
"description": (
"Directory relative paths resolve against — the workspace root, one of its "
"worktrees, or an allowed dir. Beats the inferred worktree."
),
},
},
}

Expand Down Expand Up @@ -7622,6 +7706,7 @@ def _silence_clean_edit_result(result: dict[str, Any]) -> dict[str, Any]:
# which is exactly where a wrong worktree inference would hide.
if "resolved_against" in result:
silent["resolved_against"] = result["resolved_against"]
silent["resolved_against_source"] = result.get("resolved_against_source", "inferred")
return silent
for key in _EDIT_NOISE_KEYS:
result.pop(key, None)
Expand Down Expand Up @@ -7710,7 +7795,8 @@ def _anchor_snippet(text: str, limit: int = _RETRY_ANCHOR_CHARS) -> str:
"call, even same-file (ranges use the original snapshot). {path, old, new} "
"only without a fresh range. Whole or brand-new file: {path, new, "
"replace:true}. Minified-view line numbers → add :minified "
"('f.py:minified:L10-L14'). No re-read after success."
"('f.py:minified:L10-L14'). root='/abs/dir' pins where relative paths "
"resolve (worktrees). No re-read after success."
),
param_aliases={"post_edit_hooks": "hooks"},
# Policy knobs, not agent choices: accepted by name (tests, power use) but
Expand All @@ -7728,6 +7814,7 @@ def _anchor_snippet(text: str, limit: int = _RETRY_ANCHOR_CHARS) -> str:
)
def tool_smart_edit(
edits: list[dict[str, Any]],
root: str | None = None,
atomic: bool = True,
hooks: bool = True,
post_edit_timeout_ms: int = 30_000,
Expand Down Expand Up @@ -7765,22 +7852,84 @@ def tool_smart_edit(
# `resolved_against` rides on the result, survives the clean-success
# squelch, and renders as a suffix on the one-liner. Absolute paths are
# unaffected -- _resolve_snapshot_path only applies a root to relative ones.
# A caller who knows the checkout passes `root` and skips the guessing
# entirely; see _resolve_explicit_edit_root.
_session_worktree = _session_worktree_root(repo_root)
_edit_root = _session_worktree or repo_root
edits = [_normalize_edit_aliases(e) for e in edits]
_require_edits(edits)

paths = _collect_touched_paths(edits, repo_root=_edit_root)
# Confine writes to the workspace root plus any additional directories from
# Claude Code's additionalDirectories setting or LEMONCROW_ADDITIONAL_DIRS env.
# Read tools accept any absolute path; writes need explicit opt-in.
# Path("/tmp").resolve() as well as "/tmp": on macOS /tmp is a symlink to
# /private/tmp, and the candidates below are resolved, so the bare literal
# never matched and the /tmp allowance was dead on that platform.
_extra_roots = [*_claude_additional_dirs(repo_root), Path("/tmp"), Path("/tmp").resolve()]
# _SCRATCH_EDIT_ROOTS carries the scratch allowance ("/tmp") and needs no
# twin "/private/tmp" entry: _allowed_edit_roots resolves every root before
# comparing, so one literal covers macOS's symlink.
_extra_roots = [*_claude_additional_dirs(repo_root), *_SCRATCH_EDIT_ROOTS]
if _session_worktree is not None:
_extra_roots.append(_session_worktree)
_allowed_edit_roots = [repo_root, _edit_root, *_extra_roots]

# An explicit `root` is the caller NAMING the checkout, so it outranks both
# the inference and the workspace root, and nothing is guessed underneath
# it. Out of bounds is refused, never quietly ignored.
_explicit_root: Path | None = None
if isinstance(root, str) and root.strip():
_explicit_root = _resolve_explicit_edit_root(root, workspace_root=repo_root, extra_roots=_extra_roots)
if _explicit_root is None:
return {
"failed": [
{
"paths": [root],
"error": (
f"root {root} is not this workspace ({repo_root}), one of its linked "
"worktrees, or an allowed additional directory -- pass one of those, or "
"drop root to resolve against the workspace root"
),
}
],
"rolled_back": True,
}
_extra_roots.append(_explicit_root)
_edit_root = _explicit_root or _session_worktree or repo_root
# Resolved against resolved. Touched paths arrive through
# _resolve_snapshot_path's .resolve(), while _workspace_root() hands back
# whatever the env or CLI gave it -- a macOS /tmp or /var path, a home
# reached through a symlink -- and is_relative_to is purely lexical, so an
# unresolved root lexically contains none of its own files. Compare the
# resolved forms; the escape error still prints the caller's own path.
_allowed_edit_roots = [_candidate.resolve() for _candidate in (repo_root, _edit_root, *_extra_roots)]
# Every later membership test below takes a RESOLVED path, so it needs the
# resolved root for the same reason -- under a symlinked workspace an
# unresolved one silently drops all hook diagnostics and every path the
# contract review would have read.
_repo_root_resolved = repo_root.resolve()

# A relative path naming an existing file in BOTH the inferred worktree and
# the workspace root has no right answer: the worktree came from another
# tool's cwd, not from the caller, so choosing one writes a correct change
# into a checkout nobody named (observed: an edit landing in an unrelated
# task's worktree). Refuse and make the caller say which. A path that exists
# under exactly one root, or under neither (a create), still resolves.
if _explicit_root is None and _session_worktree is not None:
_ambiguous = _ambiguous_relative_edit_paths(edits, inferred_root=_edit_root, workspace_root=repo_root)
if _ambiguous:
_collisions = "; ".join(
f"{raw} exists in " + " and ".join(str(hit) for hit in hits) for raw, hits in _ambiguous
)
return {
"failed": [
{
"paths": [raw for raw, _hits in _ambiguous],
"error": (
f"ambiguous relative edit path: {_collisions}. The worktree was inferred "
"from the last bash cwd, not named by you -- pass an absolute path, or "
"root=<dir>, to say which checkout you mean"
),
}
],
"rolled_back": True,
}

paths = _collect_touched_paths(edits, repo_root=_edit_root)

_escaped_edit_paths = [
str(_p) for _p in paths.values() if not any(_p == _r or _p.is_relative_to(_r) for _r in _allowed_edit_roots)
Expand Down Expand Up @@ -8004,7 +8153,11 @@ def tool_smart_edit(

from lemoncrow.pro.capabilities.tool_supervision.rich_edit import apply_rich_edits

result = apply_rich_edits(edits, atomic=atomic, repo_root=_edit_root, allowed_roots=_extra_roots)
# The SAME roots the confinement check above allows, not just the extras:
# rich_edit._resolve confines to [repo_root=_edit_root, *allowed_roots], so
# passing only _extra_roots dropped the main checkout and refused every
# absolute path under it while a worktree was inferred.
result = apply_rich_edits(edits, atomic=atomic, repo_root=_edit_root, allowed_roots=_allowed_edit_roots)
_phase_write_ms = int((time.monotonic() - _phase_start) * 1000)

# Sync the long-lived engine's index-version cache so the next explore
Expand Down Expand Up @@ -8145,7 +8298,7 @@ def _diag_in_repo_root(d: dict[str, Any], root: Path) -> bool:
result["diagnostics"] = [
d
for d in result["diagnostics"]
if d.get("severity") in ("error", "warning") and _diag_in_repo_root(d, repo_root)
if d.get("severity") in ("error", "warning") and _diag_in_repo_root(d, _repo_root_resolved)
]
if not result["diagnostics"]:
result.pop("diagnostics")
Expand All @@ -8162,7 +8315,7 @@ def _fmt_diag(d: dict[str, Any], root: Path) -> str:
msg = d.get("message", "")
return f"{loc} {code}: {msg}" if code else f"{loc}: {msg}"

_diag_lines = [_fmt_diag(d, repo_root) for d in result.pop("diagnostics")]
_diag_lines = [_fmt_diag(d, _repo_root_resolved) for d in result.pop("diagnostics")]
# Cap: a touched file with many pre-existing findings must not dump
# an unbounded lint report into the edit result.
if len(_diag_lines) > _EDIT_DIAG_CAP:
Expand Down Expand Up @@ -8208,7 +8361,9 @@ def _fmt_diag(d: dict[str, Any], root: Path) -> str:
result,
edits,
repo_root=repo_root,
touched_paths=[str(p.relative_to(repo_root)) for p in paths.values() if p.is_relative_to(repo_root)],
touched_paths=[
str(p.relative_to(_repo_root_resolved)) for p in paths.values() if p.is_relative_to(_repo_root_resolved)
],
)
_phase_contract_ms = int((time.monotonic() - _contract_start) * 1000)
# Incremental: refresh the shared index for the touched files now, so a
Expand All @@ -8220,10 +8375,16 @@ def _fmt_diag(d: dict[str, Any], root: Path) -> str:
"hooks": _phase_hooks_ms,
"contract": _phase_contract_ms,
}
# Disclose the worktree redirect. It is an inference, so a wrong one has to
# be visible in the same breath as the write it misdirected.
if _session_worktree is not None:
# Disclose which root the relative paths resolved against, and whether the
# caller named it: an inference that went wrong has to be visible in the
# same breath as the write it misdirected, and an explicit root carries the
# trust an inference does not.
if _explicit_root is not None:
result["resolved_against"] = str(_explicit_root)
result["resolved_against_source"] = "explicit"
elif _session_worktree is not None:
result["resolved_against"] = str(_session_worktree)
result["resolved_against_source"] = "inferred"
return _silence_clean_edit_result(result)


Expand Down
6 changes: 5 additions & 1 deletion src/lemoncrow/pro/capabilities/tool_supervision/rich_edit.py
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,11 @@ def _resolve(root: Path, raw_path: str, allowed_roots: list[Path] | None = None)
path = Path(spec.path)
resolved = path if path.is_absolute() else root / path
resolved = resolved.resolve()
roots = [root, *(allowed_roots or [])]
# `resolved` is resolved, so the roots must be too: is_relative_to is
# lexical, and a root carrying a symlink (macOS /tmp, /var, a symlinked
# home) lexically contains none of its own files. The message below still
# names `root` as the caller passed it.
roots = [Path(r).resolve() for r in (root, *(allowed_roots or []))]
if not any(resolved == r or resolved.is_relative_to(r) for r in roots):
raise ValueError(
f"path escape denied: {raw_path} is outside the workspace root {root} — "
Expand Down
Loading
Loading