Merge pull request #38 from isidentical/batuhan/fea-496-make-isolate-…

…agents-detectable

feat: Add `is_agent()` to detect whether we are in an agent

src/isolate/connections/_local/_base.py

@@ -19,6 +19,7 @@
 )
 
 from isolate.backends.common import get_executable_path, logged_io
+from isolate.connections.common import AGENT_SIGNATURE
 from isolate.logs import LogLevel
 
 if TYPE_CHECKING:
@@ -95,6 +96,7 @@ def get_env_vars(self) -> Dict[str, str]:
         immediately (so that we can seamlessly transfer logs)."""
 
         custom_vars = {}
+        custom_vars[AGENT_SIGNATURE] = "1"
         custom_vars["PYTHONUNBUFFERED"] = "1"
         if self.extra_inheritance_paths:
             # The order here should reflect the order of the inheritance

src/isolate/connections/common.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import importlib
+import os
 from contextlib import contextmanager
 from typing import TYPE_CHECKING, Any, Iterator, cast
 
@@ -15,6 +16,9 @@ def dumps(self, obj: Any) -> bytes:
             ...
 
 
+AGENT_SIGNATURE = "IS_ISOLATE_AGENT"
+
+
 class SerializationError(Exception):
     """An error that happened during the serialization process."""
 
@@ -82,3 +86,8 @@ def serialize_object(serialization_method: str, object: Any) -> bytes:
 
     with _step("serializing the given object"):
         return serialization_backend.dumps(object)
+
+
+def is_agent() -> bool:
+    """Returns true if the current process is an isolate agent."""
+    return os.environ.get(AGENT_SIGNATURE) == "1"

tests/test_connections.py

@@ -11,6 +11,7 @@
 from isolate.backends.settings import IsolateSettings
 from isolate.backends.virtualenv import VirtualPythonEnvironment
 from isolate.connections import LocalPythonGRPC, PythonIPC
+from isolate.connections.common import is_agent
 
 REPO_DIR = Path(__file__).parent.parent
 assert (
@@ -133,6 +134,16 @@ def test_extra_inheritance_paths(self, tmp_path: Any) -> None:
             # This comes from the third_env
             assert self.check_version(conn, "emoji") == "2.0.0"
 
+    def test_is_agent(self):
+        local_env = LocalPythonEnvironment()
+
+        assert not is_agent()
+        with self.open_connection(local_env, local_env.create()) as conn:
+            assert not is_agent()
+            assert conn.run(is_agent)
+            assert not is_agent()
+        assert not is_agent()
+
 
 class TestPythonIPC(GenericPythonConnectionTests):
     def open_connection(