fix(tools): isolate per-session worktree cwd

tests/tools/test_file_tools_cwd_resolution.py

@@ -314,3 +314,83 @@ def test_patch_reports_resolved_absolute_path(_isolated_cwd, monkeypatch):
     assert "WORKSPACE_PATCHED" in (workspace / "target.py").read_text()
     # And the decoy copy is untouched.
     assert (decoy / "target.py").read_text() == "DECOY_ORIGINAL\n"
+
+
+# ── Fix D: shared terminal env must not leak its cwd across worktree sessions ─
+# (June 2026: two desktop sessions, each on its own worktree, share the single
+# "default" terminal environment. Its `cwd` tracks whichever session ran the
+# last command, so a file edit from the OTHER session resolved against that
+# foreign cwd and silently landed in the wrong worktree. terminal_tool now
+# stamps env.cwd_owner with the driving session; file tools trust the shared
+# env's live cwd only when the resolving session owns it.)
+
+
+class _FakeOwnedEnv:
+    def __init__(self, cwd: str, cwd_owner: str):
+        self.cwd = cwd
+        self.cwd_owner = cwd_owner
+
+
+@pytest.fixture
+def _two_worktree_sessions(tmp_path, monkeypatch):
+    """Two worktree sessions sharing one terminal env owned by session B."""
+    wt_a = tmp_path / "wt_a"
+    wt_b = tmp_path / "wt_b"
+    main = tmp_path / "main"
+    for d in (wt_a, wt_b, main):
+        d.mkdir()
+        (d / "target.py").write_text(f"{d.name}\n")
+    monkeypatch.chdir(main)
+    monkeypatch.delenv("TERMINAL_CWD", raising=False)
+    monkeypatch.setattr(terminal_tool, "_task_env_overrides", {})
+    monkeypatch.setattr(ft, "_file_ops_cache", {})
+    # Both sessions register their worktree cwd (TUI/desktop registration path).
+    terminal_tool.register_task_env_overrides("sess-a", {"cwd": str(wt_a)})
+    terminal_tool.register_task_env_overrides("sess-b", {"cwd": str(wt_b)})
+    # The shared "default" env: session B ran the last command, so its live cwd
+    # is wt_b and B owns it.
+    monkeypatch.setattr(
+        terminal_tool,
+        "_active_environments",
+        {"default": _FakeOwnedEnv(str(wt_b), "sess-b")},
+    )
+    return wt_a, wt_b, main
+
+
+def test_live_cwd_ignored_for_non_owning_session(_two_worktree_sessions):
+    wt_a, wt_b, _main = _two_worktree_sessions
+    # Owner sees the live cwd; the other session must NOT inherit it.
+    assert ft._get_live_tracking_cwd("sess-b") == str(wt_b)
+    assert ft._get_live_tracking_cwd("sess-a") is None
+
+
+def test_resolution_routes_to_resolving_sessions_worktree(_two_worktree_sessions):
+    """The wrong-worktree fix: A resolves into wt_a, not the shared env's wt_b."""
+    wt_a, wt_b, _main = _two_worktree_sessions
+    # Session A does not own the shared env → falls back to its own registered
+    # worktree cwd instead of B's live cwd.
+    resolved_a = ft._resolve_path_for_task("target.py", task_id="sess-a")
+    assert resolved_a == (wt_a / "target.py")
+    assert not str(resolved_a).startswith(str(wt_b))
+
+
+def test_owning_session_still_resolves_against_live_cwd(_two_worktree_sessions):
+    """No regression: the owner keeps resolving against the live cwd."""
+    wt_a, wt_b, _main = _two_worktree_sessions
+    resolved_b = ft._resolve_path_for_task("target.py", task_id="sess-b")
+    assert resolved_b == (wt_b / "target.py")
+    assert not str(resolved_b).startswith(str(wt_a))
+
+
+def test_unknown_owner_keeps_prior_single_session_behavior(tmp_path, monkeypatch):
+    """An env with no owner (CLI / legacy) still yields its live cwd."""
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    monkeypatch.setattr(ft, "_file_ops_cache", {})
+    monkeypatch.setattr(
+        terminal_tool,
+        "_active_environments",
+        {"default": _FakeOwnedEnv(str(ws), "")},
+    )
+    assert ft._get_live_tracking_cwd("default") == str(ws)
+    assert ft._get_live_tracking_cwd("any-session") == str(ws)

tests/tools/test_terminal_task_cwd.py

@@ -149,11 +149,14 @@ def test_background_command_prefers_live_env_cwd_over_init_time_cwd(monkeypatch)
     )
 
     assert result["exit_code"] == 0
+    # session_key falls back to the raw task_id when no gateway contextvar is set
+    # (it doesn't propagate to tool-worker threads), so process.kill / stop can
+    # still find and terminate this background process.
     assert registry.calls == [{
         "command": "sleep 1",
         "cwd": "/workspace/live",
         "task_id": task_id,
-        "session_key": "",
+        "session_key": task_id,
         "env_vars": {},
         "use_pty": False,
     }]

tools/file_tools.py

@@ -166,6 +166,30 @@ def _registered_task_cwd_override(task_id: str = "default") -> str | None:
     return _sentinel_free_abs_cwd(overrides.get("cwd"))
 
 
+def _live_cwd_if_owned(env, task_id: str) -> str | None:
+    """The env's live cwd, but only when THIS session owns it.
+
+    The terminal env is shared (collapsed to the ``"default"`` container), so its
+    ``cwd`` tracks the LAST session that ran a command. With two worktree
+    sessions open, trusting it blindly routes one session's edits into the other
+    session's checkout (the wrong-worktree-patch bug). ``terminal_tool`` stamps
+    ``env.cwd_owner`` with the session that last drove the env; return its cwd
+    only when that owner matches the resolving session, else ``None`` so the
+    caller falls through to this session's own registered cwd override. Unknown
+    owner / ``default`` keys keep the prior behavior (single-session / CLI).
+    """
+    if env is None:
+        return None
+    live = getattr(env, "cwd", None)
+    if not live:
+        return None
+    owner = str(getattr(env, "cwd_owner", "") or "")
+    tid = str(task_id or "")
+    if owner and tid and owner != "default" and tid != "default" and owner != tid:
+        return None
+    return live
+
+
 def _get_live_tracking_cwd(task_id: str = "default") -> str | None:
     """Return the task's live terminal cwd for bookkeeping when available."""
     try:
@@ -177,18 +201,20 @@ def _get_live_tracking_cwd(task_id: str = "default") -> str | None:
     with _file_ops_lock:
         cached = _file_ops_cache.get(container_key) or _file_ops_cache.get(task_id)
     if cached is not None:
-        live_cwd = getattr(getattr(cached, "env", None), "cwd", None) or getattr(
-            cached, "cwd", None
-        )
+        env = getattr(cached, "env", None)
+        live_cwd = _live_cwd_if_owned(env, task_id)
         if live_cwd:
             return live_cwd
+        # Legacy: a cache entry carrying its own cwd with no env to own it.
+        if env is None and getattr(cached, "cwd", None):
+            return getattr(cached, "cwd", None)
 
     try:
         from tools.terminal_tool import _active_environments, _env_lock
 
         with _env_lock:
             env = _active_environments.get(container_key) or _active_environments.get(task_id)
-            live_cwd = getattr(env, "cwd", None) if env is not None else None
+        live_cwd = _live_cwd_if_owned(env, task_id)
         if live_cwd:
             return live_cwd
     except Exception:

tools/terminal_tool.py

@@ -2290,14 +2290,27 @@ def terminal_tool(
                 "EOF."
             )
 
+        # Claim the (shared "default") terminal env for the session driving this
+        # command. File tools read env.cwd_owner to decide whether the env's live
+        # cwd is THIS session's `cd` or a different worktree session's — without
+        # it, two open worktree sessions sharing the env route each other's edits
+        # to the wrong checkout. get_current_session_key()'s contextvar doesn't
+        # cross tool-worker threads, so fall back to the raw task_id (which IS the
+        # session_key for the top-level agent) — a stable, thread-safe anchor.
+        from tools.approval import get_current_session_key
+
+        session_key = get_current_session_key(default="") or (task_id or "")
+        try:
+            env.cwd_owner = session_key
+        except Exception:
+            pass
+
         if background:
             # Spawn a tracked background process via the process registry.
             # For local backends: uses subprocess.Popen with output buffering.
             # For non-local backends: runs inside the sandbox via env.execute().
-            from tools.approval import get_current_session_key
             from tools.process_registry import process_registry
 
-            session_key = get_current_session_key(default="")
             effective_cwd = _resolve_command_cwd(
                 workdir=workdir,
                 env=env,