refactor(terminal): introduce per-session cwd records (step 1: dual-write)

First step of the cwd rearchitecture (see PR #65185 for the targeted
leak fixes this will eventually supersede, and
.hermes/plans/cwd-rearch-audit.md for the full audit + sequencing).

The root cause of the wrong-worktree bug class is that cwd lives on the
SHARED terminal env — a global mutable timeshared between sessions.
env.cwd_owner stamping, _last_known_cwd, and file_tools' ownership
ladder are all patches over that misplacement.

This adds the replacement store: _session_cwd, keyed by the raw
session/task key, with record/get/clear accessors. Step 1 is dual-write
only — every site that learns a session's live cwd also records it:

- terminal_tool foreground path: after env.execute() the env's own
  post-command tracking has updated env.cwd; mirror it under the
  session key that drove the command
- register_task_env_overrides: a registered workspace cwd (ACP/TUI/
  desktop) seeds the session record
- clear_task_env_overrides: drops the record on teardown

Readers are untouched — behavior is identical. Later steps flip
file_tools resolution and _resolve_command_cwd to read this store,
then delete env-side tracking, cwd_owner, and _last_known_cwd.

Also hardens terminal_tool's env acquisition with an explicit
env-is-None guard (previously implicitly unbound on an unreachable
branch, flagged by pyright once the dual-write read env post-loop).

tests/tools/test_session_cwd_store.py

@@ -0,0 +1,114 @@
+"""Session-cwd record store (cwd rearchitecture, step 1: dual-write).
+
+The store is the future single source of truth for per-session working
+directories. Step 1 only guarantees the WRITE side: every path that learns a
+session's live cwd must record it under the raw session key. Readers still
+use the legacy env.cwd ladder; these tests pin the invariants the later
+read-side flip will rely on.
+"""
+
+import pytest
+
+import tools.terminal_tool as tt
+
+
+@pytest.fixture(autouse=True)
+def _clean_store(monkeypatch):
+    monkeypatch.setattr(tt, "_session_cwd", {})
+    monkeypatch.setattr(tt, "_task_env_overrides", {})
+
+
+class TestRecordSemantics:
+    def test_records_are_keyed_by_raw_session_key(self):
+        tt.record_session_cwd("sess-a", "/wt/a")
+        tt.record_session_cwd("sess-b", "/wt/b")
+        # No cross-talk: each session reads back exactly its own record.
+        assert tt.get_session_cwd("sess-a") == "/wt/a"
+        assert tt.get_session_cwd("sess-b") == "/wt/b"
+        assert tt.get_session_cwd("sess-c") is None
+
+    def test_none_and_empty_keys_collapse_to_default(self):
+        tt.record_session_cwd(None, "/somewhere")
+        assert tt.get_session_cwd(None) == "/somewhere"
+        assert tt.get_session_cwd("") == "/somewhere"
+        assert tt.get_session_cwd("default") == "/somewhere"
+
+    def test_invalid_cwd_values_are_ignored(self):
+        tt.record_session_cwd("sess-a", None)
+        tt.record_session_cwd("sess-a", "")
+        tt.record_session_cwd("sess-a", "   ")
+        tt.record_session_cwd("sess-a", 123)  # type: ignore[arg-type]
+        assert tt.get_session_cwd("sess-a") is None
+
+    def test_clear_drops_only_the_named_session(self):
+        tt.record_session_cwd("sess-a", "/wt/a")
+        tt.record_session_cwd("sess-b", "/wt/b")
+        tt.clear_session_cwd("sess-a")
+        assert tt.get_session_cwd("sess-a") is None
+        assert tt.get_session_cwd("sess-b") == "/wt/b"
+
+
+class TestDualWriteSites:
+    def test_register_cwd_override_seeds_the_session_record(self):
+        """A registered workspace cwd IS the session's cwd until a `cd`."""
+        tt.register_task_env_overrides("desktop-sess", {"cwd": "/wt/desktop"})
+        assert tt.get_session_cwd("desktop-sess") == "/wt/desktop"
+
+    def test_register_without_cwd_does_not_touch_the_record(self):
+        tt.register_task_env_overrides("rl-42", {"docker_image": "x:y"})
+        assert tt.get_session_cwd("rl-42") is None
+
+    def test_clear_task_env_overrides_drops_the_record(self):
+        tt.register_task_env_overrides("desktop-sess", {"cwd": "/wt/desktop"})
+        tt.clear_task_env_overrides("desktop-sess")
+        assert tt.get_session_cwd("desktop-sess") is None
+
+    def test_reregistration_updates_the_record(self):
+        """ACP session/load switching project roots mid-session."""
+        tt.register_task_env_overrides("acp-sess", {"cwd": "/proj/one"})
+        tt.register_task_env_overrides("acp-sess", {"cwd": "/proj/two"})
+        assert tt.get_session_cwd("acp-sess") == "/proj/two"
+
+
+class TestPostCommandDualWrite:
+    """The env's post-command cwd tracking must mirror into the session record."""
+
+    def _run(self, monkeypatch, task_id, env):
+        import json
+        monkeypatch.setattr(tt, "_active_environments", {task_id: env})
+        monkeypatch.setattr(tt, "_last_activity", {})
+        monkeypatch.setattr(
+            tt, "_get_env_config",
+            lambda: {"env_type": "local", "cwd": "/default", "timeout": 60,
+                     "lifetime_seconds": 3600},
+        )
+        monkeypatch.setattr(
+            tt, "_check_all_guards",
+            lambda command, env_type, **kwargs: {"approved": True},
+        )
+        return json.loads(tt.terminal_tool(command="cd /new/dir", task_id=task_id))
+
+    def test_cd_result_is_recorded_under_the_session_key(self, monkeypatch):
+        class FakeEnv:
+            env = {}
+            cwd = "/start"
+            def execute(self, command, **kwargs):
+                # Simulate the env's own post-command tracking (marker parse).
+                self.cwd = "/new/dir"
+                return {"output": "", "returncode": 0}
+
+        result = self._run(monkeypatch, "sess-a", FakeEnv())
+        assert result["exit_code"] == 0
+        assert tt.get_session_cwd("sess-a") == "/new/dir"
+        # And ONLY that session's record was touched.
+        assert tt.get_session_cwd("sess-b") is None
+
+    def test_envs_without_cwd_tracking_record_nothing(self, monkeypatch):
+        class FakeEnv:
+            env = {}
+            def execute(self, command, **kwargs):
+                return {"output": "", "returncode": 0}
+
+        result = self._run(monkeypatch, "sess-a", FakeEnv())
+        assert result["exit_code"] == 0
+        assert tt.get_session_cwd("sess-a") is None

tools/terminal_tool.py

@@ -1069,6 +1069,58 @@ def _maybe_reap_docker_orphans(container_config: Dict[str, Any]) -> None:
 # Thread-safe because each task_id is unique per rollout.
 _task_env_overrides: Dict[str, Dict[str, Any]] = {}
 
+# ── Per-session cwd records (cwd rearchitecture, step 1) ────────────────────
+#
+# The durable source of truth for "which directory is THIS session working
+# in". Keyed by the raw session/task key (NOT the collapsed container id):
+# the terminal env is shared across sessions, so any cwd state stored on the
+# env is a global mutable timeshared between sessions — the root cause of the
+# wrong-worktree bug class (env.cwd_owner stamping, _last_known_cwd, and the
+# ownership ladder in file_tools are all patches over that misplacement).
+#
+# Step 1 (this change): dual-write only. Every site that learns a session's
+# live cwd (post-command tracking, cwd-override registration) also records it
+# here. Readers still use the legacy env.cwd ladder. Later steps flip
+# file_tools and _resolve_command_cwd to read this store, then delete the
+# env-side tracking + ownership guards. See .hermes/plans/cwd-rearch-audit.md.
+_session_cwd: Dict[str, str] = {}
+_session_cwd_lock = threading.Lock()
+
+
+def record_session_cwd(session_key: Optional[str], cwd: Optional[str]) -> None:
+    """Record *cwd* as the working directory of *session_key*.
+
+    Called wherever a session's live cwd becomes known: after a terminal
+    command completes (the env's post-command tracking has just parsed the
+    resulting cwd) and when a surface registers a workspace cwd override.
+    Empty/None session keys collapse to ``"default"`` (single-session CLI).
+    Non-string / empty cwds are ignored.
+    """
+    if not isinstance(cwd, str) or not cwd.strip():
+        return
+    key = str(session_key or "default")
+    with _session_cwd_lock:
+        if _session_cwd.get(key) != cwd:
+            _session_cwd[key] = cwd
+
+
+def get_session_cwd(session_key: Optional[str]) -> Optional[str]:
+    """Return the recorded working directory for *session_key*, if any.
+
+    No fallback chain here on purpose: callers decide what an absent record
+    means (config default, TERMINAL_CWD seed, process cwd). ``None``/empty
+    keys read the ``"default"`` record.
+    """
+    key = str(session_key or "default")
+    with _session_cwd_lock:
+        return _session_cwd.get(key)
+
+
+def clear_session_cwd(session_key: str) -> None:
+    """Drop a session's cwd record (session teardown)."""
+    with _session_cwd_lock:
+        _session_cwd.pop(session_key, None)
+
 
 def register_task_env_overrides(task_id: str, overrides: Dict[str, Any]):
     """
@@ -1099,6 +1151,9 @@ def register_task_env_overrides(task_id: str, overrides: Dict[str, Any]):
     # while letting an explicit ACP cwd change win, as the client expects.
     new_cwd = overrides.get("cwd")
     if isinstance(new_cwd, str) and new_cwd.strip():
+        # Dual-write (cwd rearch step 1): a registered workspace cwd IS the
+        # session's working directory until a `cd` changes it.
+        record_session_cwd(task_id, new_cwd)
         # The live env is cached under the raw task_id for per-session surfaces
         # (ACP/gateway/dashboard) and under the collapsed container id for
         # isolation-keyed rollouts. Try the raw id first, then the container id,
@@ -1118,6 +1173,7 @@ def clear_task_env_overrides(task_id: str):
     Called during cleanup to avoid stale entries accumulating.
     """
     _task_env_overrides.pop(task_id, None)
+    clear_session_cwd(task_id)
 
 
 def _resolve_container_task_id(task_id: Optional[str]) -> str:
@@ -2164,6 +2220,7 @@ def terminal_tool(
         # Use a per-task creation lock so concurrent tool calls for the same
         # task_id wait for the first one to finish creating the sandbox,
         # instead of each creating their own (wasting Modal resources).
+        env = None
         with _env_lock:
             # Prefer the collapsed container id, but fall back to an env cached
             # under the raw task_id. Per-session surfaces (ACP/gateway/dashboard)
@@ -2265,6 +2322,16 @@ def terminal_tool(
                         env = new_env
                     logger.info("%s environment ready for task %s", env_type, effective_task_id[:8])
 
+        if env is None:
+            # Unreachable in practice (either the cached branch or the creation
+            # branch assigned env above); guard for type-safety and so a future
+            # refactor of the branches can't fall through to an AttributeError.
+            return json.dumps({
+                "output": "",
+                "exit_code": -1,
+                "error": "Terminal environment unavailable (creation raced cleanup)",
+            }, ensure_ascii=False)
+
         # Hard-block: gateway lifecycle commands (systemctl/launchctl/hermes
         # restart|stop targeting hermes-gateway) must never run inside the
         # gateway process itself. The restart would SIGTERM the gateway, which
@@ -2693,7 +2760,15 @@ def terminal_tool(
                 
                 # Got a result
                 break
-            
+
+            # Dual-write (cwd rearch step 1): the env's post-command tracking
+            # (marker parse / local sync) has just updated env.cwd with the
+            # directory this command finished in. That cwd belongs to THIS
+            # session — record it under the session key so the durable record
+            # never depends on the shared env surviving or on who drives the
+            # env next.
+            record_session_cwd(session_key, getattr(env, "cwd", None))
+
             # Extract output
             output = result.get("output", "")
             returncode = result.get("returncode", 0)