fix(file-tools): stop cwd resolution leaking across sessions

Two cross-session cwd leaks in _resolve_path_for_task, both reproduced
with failing tests before the fix:

Leak A: the shared terminal env's cwd_owner is stamped "" or "default"
whenever it is driven without a session key (top-level CLI turn, cron
tick). _live_cwd_if_owned treated that as trusted-by-everyone, and since
the live cwd is rung 1 of the resolution ladder it outranked the
resolving session's OWN registered worktree override — a desktop/TUI
session's relative edits silently landed in whatever directory the last
session-key-less command cd'd to.

Fix: when the resolving session has a registered cwd override, the live
cwd must be owned by EXACTLY that session to win (strict ownership).
Sessions without a registered override keep the prior permissive
behavior, so single-session CLI is unchanged.

Leak B: the durable _last_known_cwd registry (#26211) is keyed by the
collapsed container id ("default" for everyone) with no record of which
session produced the entry. A session with no live cwd and no registered
override inherited whatever directory the LAST session navigated to.

Fix: entries are now (cwd, owner) tuples; the read side only returns an
entry to the session that produced it (or session-agnostic "default"
entries, preserving the single-session #26211 behavior). Legacy
bare-string entries are read as session-agnostic.

Also routes the two remaining direct _last_known_cwd accesses in
_get_file_ops through the owner-aware helpers so the write/read sides
can't drift apart.

tests/tools/test_file_tools.py

@@ -797,8 +797,8 @@ class TestLastKnownCwd:
         live = _get_live_tracking_cwd(task_id)
 
         assert live == "/Users/user/project"
-        # The read mirrored the live cwd into the durable registry.
-        assert _last_known_cwd.get(task_id) == "/Users/user/project"
+        # The read mirrored the live cwd into the durable registry (owner-tagged).
+        assert _last_known_cwd.get(task_id) == ("/Users/user/project", "default")
         _last_known_cwd.pop(task_id, None)
 
     @patch("tools.terminal_tool._active_environments", new_callable=dict)
@@ -828,7 +828,7 @@ class TestLastKnownCwd:
         cached.env.cwd_owner = "default"
         mock_cache[task_id] = cached
         assert _get_live_tracking_cwd(task_id) == "/Users/user/project"
-        assert _last_known_cwd.get(task_id) == "/Users/user/project"
+        assert _last_known_cwd.get(task_id) == ("/Users/user/project", "default")
 
         # 2) Cleanup thread kills the env AND clears the cache before the next
         #    file write — so _get_file_ops' save-old-cwd branch never runs.

tests/tools/test_file_tools_cwd_resolution.py

@@ -477,11 +477,126 @@ def test_preserved_cwd_does_not_override_non_owning_sessions_worktree(
     wt_a, wt_b, _main = _two_worktree_sessions
     monkeypatch.setattr(ft, "_last_known_cwd", {})
 
-    # Owner B resolves first — this mirrors wt_b into _last_known_cwd['default'].
+    # Owner B resolves first — this mirrors wt_b into _last_known_cwd['default']
+    # (owner-tagged with sess-b so other sessions can't inherit it).
     assert ft._resolve_path_for_task("target.py", task_id="sess-b") == (wt_b / "target.py")
-    assert ft._last_known_cwd.get("default") == str(wt_b)
+    assert ft._last_known_cwd.get("default") == (str(wt_b), "sess-b")
 
     # A still routes to its own registered worktree despite the shared anchor.
     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))
+
+
+# ── Fix E: unowned/default-owned shared cwd must not beat a registered override ─
+# (July 2026: the shared env's cwd_owner is stamped "" or "default" whenever it
+# is driven without a session key — a top-level CLI turn, a cron tick. The
+# ownership guard treated that as "trusted by everyone", and since the live cwd
+# is rung 1 of the resolution ladder it outranked the resolving session's OWN
+# registered worktree override. Sibling leak: _last_known_cwd entries carried no
+# owner at all, so a session with no anchor of its own inherited whatever
+# directory the LAST session navigated to.)
+
+
+def test_default_owned_live_cwd_does_not_beat_registered_override(
+    _two_worktree_sessions, monkeypatch
+):
+    """An env last driven without a session key must not hijack a registered worktree."""
+    wt_a, wt_b, _main = _two_worktree_sessions
+    monkeypatch.setattr(ft, "_last_known_cwd", {})
+    # Shared env re-stamped by a session-key-less driver (CLI turn / cron tick),
+    # its cwd pointing at some other checkout.
+    monkeypatch.setattr(
+        terminal_tool,
+        "_active_environments",
+        {"default": _FakeOwnedEnv(str(wt_b), "default")},
+    )
+    resolved = ft._resolve_path_for_task("target.py", task_id="sess-a")
+    assert resolved == (wt_a / "target.py")
+    assert not str(resolved).startswith(str(wt_b))
+
+
+def test_empty_owned_live_cwd_does_not_beat_registered_override(
+    _two_worktree_sessions, monkeypatch
+):
+    """Same leak with owner stamped '' (get_current_session_key default)."""
+    wt_a, wt_b, _main = _two_worktree_sessions
+    monkeypatch.setattr(ft, "_last_known_cwd", {})
+    monkeypatch.setattr(
+        terminal_tool,
+        "_active_environments",
+        {"default": _FakeOwnedEnv(str(wt_b), "")},
+    )
+    resolved = ft._resolve_path_for_task("target.py", task_id="sess-a")
+    assert resolved == (wt_a / "target.py")
+    assert not str(resolved).startswith(str(wt_b))
+
+
+def test_unowned_live_cwd_still_wins_without_registered_override(tmp_path, monkeypatch):
+    """Single-session CLI behavior unchanged: no override → unowned live cwd is used."""
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.delenv("TERMINAL_CWD", raising=False)
+    monkeypatch.setattr(terminal_tool, "_task_env_overrides", {})
+    monkeypatch.setattr(ft, "_file_ops_cache", {})
+    monkeypatch.setattr(ft, "_last_known_cwd", {})
+    monkeypatch.setattr(
+        terminal_tool,
+        "_active_environments",
+        {"default": _FakeOwnedEnv(str(ws), "")},
+    )
+    assert ft._resolve_path_for_task("target.py", task_id="some-session") == (ws / "target.py")
+
+
+def test_preserved_cwd_does_not_leak_to_session_without_any_anchor(
+    _two_worktree_sessions, monkeypatch
+):
+    """A preserved anchor recorded by session B must not be handed to session C.
+
+    Session C has NO live cwd, NO registered override — before the fix it fell
+    through to _last_known_cwd['default'] and inherited B's worktree.
+    Now the owner-tagged entry is refused and C falls back to the process cwd.
+    """
+    wt_a, wt_b, main = _two_worktree_sessions
+    monkeypatch.setattr(ft, "_last_known_cwd", {})
+
+    # B (owner) does a live read → mirrors wt_b into the shared registry.
+    assert ft._resolve_path_for_task("x.py", task_id="sess-b") == (wt_b / "x.py")
+    # Env cleaned up.
+    monkeypatch.setattr(terminal_tool, "_active_environments", {})
+
+    # C: not registered, never drove the env. Must NOT inherit wt_b.
+    resolved = ft._resolve_path_for_task("target.py", task_id="sess-c")
+    assert not str(resolved).startswith(str(wt_b))
+    assert resolved == (main / "target.py").resolve()
+
+    # B itself still gets its preserved anchor back (that's the #26211 fix).
+    assert ft._resolve_path_for_task("y.py", task_id="sess-b") == (wt_b / "y.py")
+
+
+def test_default_task_still_reads_preserved_cwd(tmp_path, monkeypatch):
+    """#26211 unchanged for the single-session default task."""
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.delenv("TERMINAL_CWD", raising=False)
+    monkeypatch.setattr(terminal_tool, "_task_env_overrides", {})
+    monkeypatch.setattr(ft, "_file_ops_cache", {})
+    monkeypatch.setattr(ft, "_last_known_cwd", {})
+    monkeypatch.setattr(terminal_tool, "_active_environments", {})
+
+    ft._remember_last_known_cwd("default", str(ws), owner="default")
+    assert ft._resolve_path_for_task("target.py", task_id="default") == (ws / "target.py")
+
+
+def test_legacy_bare_string_registry_entry_still_readable(tmp_path, monkeypatch):
+    """Old-format bare-string entries are treated as session-agnostic."""
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    monkeypatch.setattr(terminal_tool, "_task_env_overrides", {})
+    monkeypatch.setattr(ft, "_file_ops_cache", {})
+    monkeypatch.setattr(ft, "_last_known_cwd", {"default": str(ws)})
+    monkeypatch.setattr(terminal_tool, "_active_environments", {})
+    assert ft._last_known_cwd_for("default") == str(ws)
+    assert ft._last_known_cwd_for("any-session") == str(ws)

tools/file_tools.py

@@ -269,7 +269,7 @@ 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:
+def _live_cwd_if_owned(env, task_id: str, strict: bool = False) -> 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
@@ -280,6 +280,13 @@ def _live_cwd_if_owned(env, task_id: str) -> str | None:
     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).
+
+    ``strict=True`` requires the owner to be EXACTLY the resolving session —
+    an unowned or ``"default"``-owned cwd is rejected too. Used when the
+    resolving session has its own registered workspace override: a shared env
+    last driven without a session key (top-level CLI turn, cron tick) is
+    stamped ``""``/``"default"`` and must not outrank the session's own
+    registered worktree.
     """
     if env is None:
         return None
@@ -288,13 +295,20 @@ def _live_cwd_if_owned(env, task_id: str) -> str | None:
         return None
     owner = str(getattr(env, "cwd_owner", "") or "")
     tid = str(task_id or "")
+    if strict:
+        return live if (owner and owner == tid) else None
     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."""
+def _get_live_tracking_cwd(task_id: str = "default", strict: bool = False) -> str | None:
+    """Return the task's live terminal cwd for bookkeeping when available.
+
+    ``strict`` is forwarded to :func:`_live_cwd_if_owned` — when the resolving
+    session has its own registered workspace override, only an env explicitly
+    owned by that exact session may outrank it.
+    """
     try:
         from tools.terminal_tool import _resolve_container_task_id
         container_key = _resolve_container_task_id(task_id)
@@ -305,14 +319,15 @@ def _get_live_tracking_cwd(task_id: str = "default") -> str | None:
         cached = _file_ops_cache.get(container_key) or _file_ops_cache.get(task_id)
     if cached is not None:
         env = getattr(cached, "env", None)
-        live_cwd = _live_cwd_if_owned(env, task_id)
+        live_cwd = _live_cwd_if_owned(env, task_id, strict=strict)
         if live_cwd:
-            _remember_last_known_cwd(container_key, live_cwd)
+            _remember_last_known_cwd(container_key, live_cwd, owner=task_id)
             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):
+        # An ownerless legacy cwd can't satisfy strict ownership.
+        if not strict and env is None and getattr(cached, "cwd", None):
             legacy_cwd = getattr(cached, "cwd", None)
-            _remember_last_known_cwd(container_key, legacy_cwd)
+            _remember_last_known_cwd(container_key, legacy_cwd, owner=task_id)
             return legacy_cwd
 
     try:
@@ -320,9 +335,9 @@ def _get_live_tracking_cwd(task_id: str = "default") -> str | None:
 
         with _env_lock:
             env = _active_environments.get(container_key) or _active_environments.get(task_id)
-        live_cwd = _live_cwd_if_owned(env, task_id)
+        live_cwd = _live_cwd_if_owned(env, task_id, strict=strict)
         if live_cwd:
-            _remember_last_known_cwd(container_key, live_cwd)
+            _remember_last_known_cwd(container_key, live_cwd, owner=task_id)
             return live_cwd
     except Exception:
         pass
@@ -345,8 +360,6 @@ def _authoritative_workspace_root(task_id: str = "default") -> str | None:
     case callers fall back to the process cwd.
     """
     live = _get_live_tracking_cwd(task_id)
-    if live:
-        return live
     # A session-specific registered override (TUI/Desktop/ACP workspace cwd)
     # is more authoritative than the shared last-known anchor: it is keyed by
     # the raw session id, so when two worktree sessions share the single
@@ -354,6 +367,15 @@ def _authoritative_workspace_root(task_id: str = "default") -> str | None:
     # registered worktree — never the other session's leftover cwd. (Checked
     # before _last_known_cwd, which is keyed by the shared container id.)
     registered = _registered_task_cwd_override(task_id)
+    if live and registered:
+        # Both anchors exist: the shared env's live cwd may only outrank this
+        # session's own registered worktree when the env is owned by EXACTLY
+        # this session. An unowned/"default"-owned cwd (last driven by a
+        # top-level CLI turn or cron tick with no session key) is some other
+        # context's leftover `cd` — the registered override wins.
+        live = _get_live_tracking_cwd(task_id, strict=True)
+    if live:
+        return live
     if registered:
         return registered
     # When the terminal env was cleaned up mid-conversation, the live cwd is
@@ -797,10 +819,15 @@ _file_ops_cache: dict = {}
 # Per-task last-known CWD — preserved across env re-creation so
 # relative-path file writes land in the right directory after the
 # terminal environment is cleaned up and rebuilt (root cause of #26211).
+# Values are ``(cwd, owner)`` tuples: the registry is keyed by the shared
+# container id (usually "default"), so without an owner tag any session
+# would inherit whatever directory the LAST session navigated to — the
+# cross-session leak. ``owner`` is the raw session/task id that produced
+# the cwd ("default" for single-session/CLI, which every session may use).
 _last_known_cwd: dict = {}
 
 
-def _remember_last_known_cwd(task_id: str, cwd: str | None) -> None:
+def _remember_last_known_cwd(task_id: str, cwd: str | None, owner: str | None = None) -> None:
     """Mirror a live terminal cwd into the durable ``_last_known_cwd`` registry.
 
     Belt-and-suspenders for #26211: the cleanup thread can pop BOTH
@@ -811,12 +838,29 @@ def _remember_last_known_cwd(task_id: str, cwd: str | None) -> None:
     read (which happens on every relative-path file resolution while the env is
     alive), the durable anchor no longer depends on the cleanup-detection
     branch firing, so it survives recreation regardless of pop ordering.
+
+    ``owner`` tags the entry with the session that produced the cwd so the
+    read side (:func:`_last_known_cwd_for`) can refuse to hand one session's
+    preserved directory to a different session.
     """
     if not cwd:
         return
+    entry = (cwd, str(owner or task_id or "default"))
     with _file_ops_lock:
-        if _last_known_cwd.get(task_id) != cwd:
-            _last_known_cwd[task_id] = cwd
+        if _last_known_cwd.get(task_id) != entry:
+            _last_known_cwd[task_id] = entry
+
+
+def _last_known_cwd_entry(key: str) -> tuple[str, str] | None:
+    """Read a raw registry entry, tolerating legacy bare-string values."""
+    value = _last_known_cwd.get(key)
+    if value is None:
+        return None
+    if isinstance(value, tuple):
+        return value
+    # Legacy bare-string entry (written by older code / direct test presets):
+    # no owner recorded — treat as session-agnostic, matching prior behavior.
+    return (value, "default")
 
 
 def _last_known_cwd_for(task_id: str = "default") -> str | None:
@@ -825,14 +869,28 @@ def _last_known_cwd_for(task_id: str = "default") -> str | None:
     The registry is keyed by the resolved container id (the same key used by
     the save sites in ``_get_file_ops`` / ``_get_live_tracking_cwd``), so look
     up the resolved key first and fall back to the raw task id.
+
+    Ownership guard: an entry is returned only when it was produced by THIS
+    session, or is session-agnostic (owner ``"default"``, i.e. single-session
+    CLI / legacy). A preserved cwd recorded by a *different* session must not
+    leak here — the whole point of the shared-container key is convenience,
+    not shared workspace state.
     """
     try:
         from tools.terminal_tool import _resolve_container_task_id
         container_key = _resolve_container_task_id(task_id)
     except Exception:
         container_key = task_id
+    tid = str(task_id or "default")
     with _file_ops_lock:
-        return _last_known_cwd.get(container_key) or _last_known_cwd.get(task_id)
+        for key in (container_key, task_id):
+            entry = _last_known_cwd_entry(key)
+            if entry is None:
+                continue
+            cwd, owner = entry
+            if owner == "default" or tid == "default" or owner == tid:
+                return cwd
+        return None
 
 # Track files read per task to detect re-read loops and deduplicate reads.
 # Per task_id we store:
@@ -1074,8 +1132,7 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
                 # file-creation failures in long-running conversations).
                 old_cwd = getattr(cached, "cwd", None)
                 if old_cwd:
-                    with _file_ops_lock:
-                        _last_known_cwd[task_id] = old_cwd
+                    _remember_last_known_cwd(task_id, old_cwd, owner=raw_task_id)
                 with _file_ops_lock:
                     _file_ops_cache.pop(task_id, None)
 
@@ -1113,7 +1170,7 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
             else:
                 image = ""
 
-            cwd = overrides.get("cwd") or _last_known_cwd.get(task_id) or config["cwd"]
+            cwd = overrides.get("cwd") or _last_known_cwd_for(raw_task_id) or config["cwd"]
             # Re-apply the container cwd guard that _get_env_config() already
             # ran on config["cwd"] (see #50636).  A per-task cwd override
             # registered by the gateway/TUI/ACP for workspace tracking is a