feat(tools): add named and dynamic execution targets

agent/chat_completion_helpers.py

@@ -45,7 +45,7 @@
 )
 from agent.reasoning_summaries import separate_glued_reasoning_blocks
 from agent.stream_single_writer import claim_stream_writer, stream_writer_is_current
-from tools.terminal_tool import is_persistent_env
+from tools.terminal_tool import is_persistent_env  # Backward-compatible test/patch seam.
 from utils import base_url_host_matches, base_url_hostname, env_float, env_int
 
 logger = logging.getLogger(__name__)
@@ -3191,9 +3191,9 @@ def _managed_summary_call(request, callback, *, retry_count: int):
 def cleanup_task_resources(agent, task_id: str) -> None:
     """Clean up VM and browser resources for a given task.
 
-    Skips ``cleanup_vm`` when the active terminal environment is marked
-    persistent (``persistent_filesystem=True``) so that long-lived sandbox
-    containers survive between turns. The idle reaper in
+    Removes each non-persistent target while keeping persistent named sibling
+    environments live so long-lived containers survive between turns. The
+    idle reaper in
     ``terminal_tool._cleanup_inactive_envs`` still tears them down once
     ``terminal.lifetime_seconds`` is exceeded. Non-persistent backends are
     torn down per-turn as before to prevent resource leakage (the original
@@ -3205,14 +3205,26 @@ def cleanup_task_resources(agent, task_id: str) -> None:
     idle sessions.
     """
     try:
-        if is_persistent_env(task_id):
-            if agent.verbose_logging:
-                logging.debug(
-                    f"Skipping per-turn cleanup_vm for persistent env {task_id}; "
-                    f"idle reaper will handle it."
-                )
-        else:
-            _ra().cleanup_vm(task_id)
+        from tools.terminal_tool import (
+            active_environment_turns,
+            defer_environment_turn_cleanup,
+            release_logical_environment_turn_for_cleanup,
+        )
+
+        # The current logical turn must stop counting itself before deciding
+        # whether it is the last user of the collapsed shared environment.
+        # The lease is idempotent, so duplicate finalization paths and the
+        # outer exception fallback cannot decrement another overlapping turn.
+        release_logical_environment_turn_for_cleanup(task_id)
+        remaining_turns = active_environment_turns(task_id)
+        include_collapsed = remaining_turns == 0
+        if not include_collapsed:
+            defer_environment_turn_cleanup(task_id)
+        _ra().cleanup_vm(
+            task_id,
+            preserve_persistent=True,
+            include_collapsed=include_collapsed,
+        )
     except Exception as e:
         if agent.verbose_logging:
             logger.warning("Failed to cleanup VM for task %s: %s", task_id, e)

agent/file_safety.py

@@ -59,6 +59,10 @@ def build_write_denied_paths(home: str) -> set[str]:
             # Bitwarden Secrets Manager encrypted disk cache.
             str(hermes_home / "cache" / "bws_cache.enc.json"),
             str(hermes_root / "cache" / "bws_cache.enc.json"),
+            # Keyed execution-target fingerprint secret. Exposing or replacing
+            # it would let an untrusted tool forge persisted target identities.
+            str(hermes_home / ".execution-target-fingerprint-key"),
+            str(hermes_root / ".execution-target-fingerprint-key"),
             os.path.join(home, ".netrc"),
             os.path.join(home, ".pgpass"),
             os.path.join(home, ".npmrc"),
@@ -329,6 +333,7 @@ def get_read_block_error(path: str) -> Optional[str]:
         "auth.lock",
         ".anthropic_oauth.json",
         ".env",
+        ".execution-target-fingerprint-key",
         "webhook_subscriptions.json",
         os.path.join("auth", "google_oauth.json"),
         # Bitwarden Secrets Manager disk cache: stores plaintext secret values

agent/prompt_builder.py

@@ -4,6 +4,7 @@
 assemble pieces, then combines them with memory and ephemeral prompts.
 """
 
+import hashlib
 import json
 import logging
 import os
@@ -1128,7 +1129,7 @@ def hud_surface_note(valid_tool_names: "set[str] | None" = None) -> str:
 # a mid-process backend switch rebuilds the string. Kept in-module (not on
 # disk) because the probe captures live backend state that may change
 # across Hermes restarts.
-_BACKEND_PROBE_CACHE: dict[tuple[str, str], str] = {}
+_BACKEND_PROBE_CACHE: dict[tuple[str, str, str], str] = {}
 
 
 def _windows_marketing_version() -> str:
@@ -1175,16 +1176,26 @@ def _windows_marketing_version() -> str:
 )
 
 
-def _probe_remote_backend(env_type: str) -> str | None:
+def _probe_remote_backend(
+    env_type: str,
+    terminal_config: dict | None = None,
+    target_name: str = "",
+) -> str | None:
     """Run a tiny introspection command inside the active terminal backend.
 
     Returns a pre-formatted multi-line string describing the backend's OS,
     $HOME, cwd, and user — or None if the probe failed. Result is cached
     per process. Used only for non-local backends where the agent's tools
     operate on a different machine than the host Hermes runs on.
     """
-    cwd_hint = os.getenv("TERMINAL_CWD", "")
-    cache_key = (env_type, cwd_hint)
+    if terminal_config is None:
+        config_identity = os.getenv("TERMINAL_CWD", "")
+    else:
+        serialized = json.dumps(
+            terminal_config, sort_keys=True, default=str, ensure_ascii=True,
+        )
+        config_identity = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
+    cache_key = (env_type, target_name, config_identity)
     cached = _BACKEND_PROBE_CACHE.get(cache_key)
     if cached is not None:
         return cached or None
@@ -1199,7 +1210,11 @@ def _probe_remote_backend(env_type: str) -> str | None:
         return None
 
     try:
-        config = _get_env_config()
+        config = (
+            _get_env_config(dict(terminal_config))
+            if terminal_config is not None
+            else _get_env_config()
+        )
         # Build the environment the same way tools/terminal_tool.py does for a
         # live command: select the backend image, then assemble ssh/container
         # config from the env-derived dict. (There is no `get_environment`
@@ -1251,7 +1266,12 @@ def _probe_remote_backend(env_type: str) -> str | None:
             timeout=config.get("timeout", 180),
             ssh_config=ssh_config,
             container_config=container_config,
-            task_id="prompt-backend-probe",
+            task_id=(
+                "prompt-backend-probe-"
+                + hashlib.sha256(
+                    f"{env_type}:{target_name}:{config_identity}".encode("utf-8")
+                ).hexdigest()[:12]
+            ),
             host_cwd=config.get("host_cwd"),
         )
         # Single-line POSIX probe — works on any Unixy backend. Wrapped in
@@ -1329,7 +1349,26 @@ def build_environment_hints() -> str:
 
     hints: list[str] = []
 
-    backend = (os.getenv("TERMINAL_ENV") or "local").strip().lower()
+    target_inventory = ()
+    default_target = None
+    try:
+        from tools.execution_targets import list_execution_targets
+
+        resolved_targets = list_execution_targets()
+        if resolved_targets and resolved_targets[0].named:
+            target_inventory = resolved_targets
+            default_target = next(
+                (item for item in target_inventory if item.is_default),
+                target_inventory[0],
+            )
+    except Exception as e:
+        logger.debug("Could not resolve named execution targets for prompt: %s", e)
+
+    backend = (
+        default_target.backend
+        if default_target is not None
+        else (os.getenv("TERMINAL_ENV") or "local")
+    ).strip().lower()
     is_remote_backend = backend in _REMOTE_TERMINAL_BACKENDS
 
     if not is_remote_backend:
@@ -1347,8 +1386,15 @@ def build_environment_hints() -> str:
 
         host_lines.append(f"User home directory: {os.path.expanduser('~')}")
         try:
-            host_lines.append(f"Current working directory: {resolve_agent_cwd()}")
-        except OSError:
+            if default_target is not None:
+                from tools.terminal_tool import _get_env_config
+
+                cwd_hint = _get_env_config(dict(default_target.config)).get("cwd")
+            else:
+                cwd_hint = resolve_agent_cwd()
+            if cwd_hint:
+                host_lines.append(f"Current working directory: {cwd_hint}")
+        except (OSError, TypeError, ValueError):
             pass
 
         if sys.platform == "win32" and not is_wsl():
@@ -1366,11 +1412,32 @@ def build_environment_hints() -> str:
             hints.append(_WINDOWS_BASH_SHELL_HINT)
     else:
         # --- Remote backend block (host info suppressed) ---
-        probe = _probe_remote_backend(backend)
+        probe = (
+            _probe_remote_backend(
+                backend,
+                terminal_config=dict(default_target.config),
+                target_name=(
+                    f"{default_target.profile_scope}:{default_target.target}"
+                    if default_target.profile_scope
+                    else default_target.target
+                ),
+            )
+            if default_target is not None
+            else _probe_remote_backend(backend)
+        )
+        tool_scope = (
+            "Calls to `terminal`, `read_file`, `write_file`, `patch`, "
+            "`search_files`, and `execute_code` that omit an execution-target "
+            "selector operate"
+            if default_target is not None
+            else (
+                "Your `terminal`, `read_file`, `write_file`, `patch`, "
+                "`search_files`, and `execute_code` tools all operate"
+            )
+        )
         if probe:
             hints.append(
-                f"Terminal backend: {backend}. Your `terminal`, `read_file`, "
-                f"`write_file`, `patch`, and `search_files` tools all operate "
+                f"Terminal backend: {backend}. {tool_scope} "
                 f"inside this {backend} environment — NOT on the machine "
                 f"where Hermes itself is running. The host OS, home, and cwd "
                 f"of the Hermes process are irrelevant; only the following "
@@ -1381,8 +1448,7 @@ def build_environment_hints() -> str:
                 backend, f"a {backend} environment (likely Linux)"
             )
             hints.append(
-                f"Terminal backend: {backend}. Your `terminal`, `read_file`, "
-                f"`write_file`, `patch`, and `search_files` tools all operate "
+                f"Terminal backend: {backend}. {tool_scope} "
                 f"inside {description} — NOT on the machine where Hermes "
                 f"itself runs. The backend probe didn't respond at "
                 f"prompt-build time, so the sandbox's current user, $HOME, "
@@ -1391,6 +1457,23 @@ def build_environment_hints() -> str:
                 f"`uname -a && whoami && pwd`."
             )
 
+    if default_target is not None and default_target.named:
+        target_summary = ", ".join(
+            f"{json.dumps(item.target, ensure_ascii=True)} ({item.backend}"
+            f"{', default' if item.is_default else ''})"
+            for item in target_inventory
+        )
+        hints.append(
+            "Configured execution targets: " + target_summary + ". "
+            "`terminal`, `read_file`, `write_file`, `patch`, `search_files`, and "
+            "`execute_code` select one with `execution_target`. "
+            "`search_files.target` remains its content/files search-mode selector. "
+            "Omitting the selector uses the default target. Environment facts "
+            "above describe only the default target "
+            f"{json.dumps(default_target.target, ensure_ascii=True)}; "
+            "tool results report the resolved target, backend, and cwd."
+        )
+
     if is_wsl():
         hints.append(WSL_ENVIRONMENT_HINT)
 

agent/tool_dispatch_helpers.py

@@ -114,6 +114,18 @@ def _is_mcp_tool_parallel_safe(tool_name: str) -> bool:
         return False
 
 
+def _named_execution_targets_enabled() -> bool:
+    """Return whether omitted file selectors route through a named default."""
+    try:
+        from tools.execution_targets import list_execution_targets
+
+        return any(target.named for target in list_execution_targets())
+    except Exception:
+        # The tools report malformed target config. The planner must still fail
+        # conservative rather than race path mutations before that happens.
+        return True
+
+
 def _plan_tool_batch_segments(tool_calls, *, execution_cwd: Optional[Path] = None) -> List[tuple]:
     """Split a tool-call batch into ordered ``(kind, calls)`` segments.
 
@@ -128,18 +140,10 @@ def _plan_tool_batch_segments(tool_calls, *, execution_cwd: Optional[Path] = Non
 
     * ``_NEVER_PARALLEL_TOOLS`` (interactive tools) → barrier.
     * Unparseable / non-dict arguments → barrier.
-    * Path-scoped tools (``read_file``/``search_files``/``write_file``/
-      ``patch``) join a parallel run only when their target path(s) do not
-      CONFLICT with a path already reserved in the same run.  Reservations
-      carry a reader/writer role: reader↔reader overlap is harmless (two
-      reads of the same file commute) and stays parallel; any overlap
-      involving a writer closes the run so the conflicting call starts a
-      NEW run after the first completes.  ``search_files`` reserves its
-      search root (default ``.``) as a reader — a search batched after a
-      write into the searched subtree is ordered behind that write instead
-      of racing it.  For V4A ``patch(mode="patch")`` the reserved paths are
-      the file headers in the patch body, not a possibly-stale ``path=``
-      argument.
+    * Once named execution targets are enabled, stateful tools become
+      conservative barriers. Target-specific cwd/path identity is resolved
+      lazily by the tool layer and cannot be compared safely in this lexical
+      planner.
     * Anything not in ``_PARALLEL_SAFE_TOOLS`` and not an opted-in MCP
       tool → barrier.
 
@@ -151,6 +155,7 @@ def _plan_tool_batch_segments(tool_calls, *, execution_cwd: Optional[Path] = Non
     current: list = []
     # (canonical_path, is_writer) reservations for the current parallel run.
     reserved_paths: list[tuple[Path, bool]] = []
+    named_default = _named_execution_targets_enabled()
 
     def _close_parallel() -> None:
         nonlocal current, reserved_paths
@@ -194,6 +199,10 @@ def _add_sequential(tc) -> None:
             continue
 
         if tool_name in _PATH_SCOPED_TOOLS:
+            target_selected = function_args.get("execution_target") is not None
+            if named_default or target_selected:
+                _add_sequential(tool_call)
+                continue
             scoped_paths = _extract_parallel_scope_paths(
                 tool_name, function_args, execution_cwd=execution_cwd
             )