Python: Move approval storage to platform supplied durable dir

python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

@@ -133,7 +133,7 @@
 _HOSTED_RESPONSES_HISTORY_SOURCE_ID = "_foundry_responses_history"
 
 
-# region Approval Storage
+# region Storage
 class ApprovalStorage(Protocol):
     """Storage for saving function approval requests."""
 
@@ -300,7 +300,43 @@ def _approval_storage_path_for_user(base_path: str, user_id: str) -> str:
     return str(user_dir / filename)
 
 
-# endregion Approval Storage
+_HOME_DIR_ENV_VAR = "HOME"
+_HOME_DIR_FALLBACK = "/home/session"
+
+
+def _resolve_storage_path(storage_path: str, *, is_hosted: bool) -> str:
+    """Resolve file storage beneath the durable home directory when hosted.
+
+    Hosted paths use ``$HOME`` or fall back to ``/home/session``. Local paths
+    use the current working directory.
+    """
+    relative_path = storage_path.lstrip("/")
+    if not is_hosted:
+        return str(Path.cwd() / relative_path)
+
+    home = os.environ.get(_HOME_DIR_ENV_VAR, "").strip()
+    if home and home != "/":
+        try:
+            home_path = Path(home)
+            if not home_path.is_absolute():
+                raise ValueError(f"{_HOME_DIR_ENV_VAR} must be an absolute path: {home!r}")
+            resolved = home_path.resolve()
+            # Make sure the resolved path is not the root directory, which would allow
+            # writing to arbitrary locations on the host filesystem.
+            if resolved.parent != resolved:
+                return str(resolved / relative_path)
+        except (OSError, RuntimeError, ValueError):
+            logger.warning(
+                "Failed to resolve $HOME=%r, falling back to %s",
+                home,
+                _HOME_DIR_FALLBACK,
+                exc_info=True,
+            )
+
+    return f"{_HOME_DIR_FALLBACK}/{relative_path}"
+
+
+# endregion Storage
 
 # Foundry Toolbox Auth integration
 # Consent-URL error code returned by the Foundry MCP gateway when calling `/list`
@@ -386,27 +422,6 @@ class ResponsesHostServer(ResponsesAgentServerHost):
     FUNCTION_APPROVAL_STORAGE_PATH = "/.function_approvals/approval_requests.json"
     SESSION_STORAGE_PATH = "/.sessions"
 
-    @staticmethod
-    def _resolve_checkpoint_root(is_hosted: bool) -> str:
-        """Resolve checkpoint storage path.
-
-        Hosted: $HOME/.checkpoints (or /home/session/.checkpoints).
-        Local: {cwd}/.checkpoints.
-        """
-        if not is_hosted:
-            return os.path.join(os.getcwd(), ".checkpoints")
-
-        home = os.environ.get("HOME", "").strip()
-        if home and home != "/":
-            try:
-                resolved = Path(home).resolve()
-                if str(resolved) != str(resolved.root):
-                    return str(resolved / ".checkpoints")
-            except (OSError, ValueError):
-                pass
-
-        return "/home/session/.checkpoints"
-
     def __init__(
         self,
         agent: SupportsAgentRun,
@@ -457,7 +472,9 @@ def __init__(
                     "There should not be a checkpoint storage already present in the workflow agent. "
                     "The hosting infrastructure will manage checkpoints instead."
                 )
-            self._checkpoint_storage_path = self._resolve_checkpoint_root(self.config.is_hosted)
+            self._checkpoint_storage_path = _resolve_storage_path(
+                self.CHECKPOINT_STORAGE_PATH, is_hosted=self.config.is_hosted
+            )
             self._is_workflow_agent = True
 
         self._uses_hosted_responses_history = False
@@ -479,21 +496,23 @@ def __init__(
                 )
 
         self._agent: SupportsAgentRun = agent
+        self._session_storage_path = _resolve_storage_path(self.SESSION_STORAGE_PATH, is_hosted=self.config.is_hosted)
         self._session_store: SessionStore | None = (
-            (
-                FoundrySessionStore(Path.home() / self.SESSION_STORAGE_PATH.lstrip("/"))
-                if self.config.is_hosted
-                else SessionStore()
-            )
+            (FoundrySessionStore(self._session_storage_path) if self.config.is_hosted else SessionStore())
             if not self._is_workflow_agent
             else None
         )
+
+        self._approval_storage_path = _resolve_storage_path(
+            self.FUNCTION_APPROVAL_STORAGE_PATH, is_hosted=self.config.is_hosted
+        )
         self._approval_storage: ApprovalStorage = (
-            FileBasedFunctionApprovalStorage(self.FUNCTION_APPROVAL_STORAGE_PATH)
+            FileBasedFunctionApprovalStorage(self._approval_storage_path)
             if self.config.is_hosted
             else InMemoryFunctionApprovalStorage()
         )
         self._approval_storages_by_user: dict[str, ApprovalStorage] = {}
+
         # Lazy agent lifecycle: the agent (and any MCP tools it owns) is entered on
         # the first request rather than at server startup, so that authentication
         # failures during MCP connect can be surfaced to the client as an
@@ -540,7 +559,7 @@ def _approval_storage_for_request(self) -> ApprovalStorage:
         storage = self._approval_storages_by_user.get(user_id)
         if storage is None:
             storage = FileBasedFunctionApprovalStorage(
-                _approval_storage_path_for_user(self.FUNCTION_APPROVAL_STORAGE_PATH, user_id)
+                _approval_storage_path_for_user(self._approval_storage_path, user_id)
             )
             self._approval_storages_by_user[user_id] = storage
         return storage

python/packages/foundry_hosting/tests/test_responses.py

@@ -375,16 +375,17 @@ def test_init_uses_foundry_session_store_by_default_when_hosted(
         monkeypatch: pytest.MonkeyPatch,
     ) -> None:
         monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
+        monkeypatch.setenv("HOME", str(tmp_path))
         agent = _make_agent(
             response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("hi")])])
         )
 
-        with patch.object(Path, "home", return_value=tmp_path):
-            server = ResponsesHostServer(agent, store=InMemoryResponseProvider())
+        server = ResponsesHostServer(agent, store=InMemoryResponseProvider())
 
         session_store = server._session_store  # pyright: ignore[reportPrivateUsage]
         assert isinstance(session_store, FoundrySessionStore)
         assert session_store.storage_path == tmp_path / ".sessions"
+        assert server._session_storage_path == str(tmp_path / ".sessions")  # pyright: ignore[reportPrivateUsage]
         assert server.SESSION_STORAGE_PATH == "/.sessions"
 
     def test_init_rejects_history_provider_with_load_messages(self) -> None:
@@ -5189,4 +5190,54 @@ def test_hosted_with_unusable_home_falls_back_to_default(
         assert server._checkpoint_storage_path != "/.checkpoints"
 
 
+@pytest.mark.filterwarnings("ignore::DeprecationWarning")
+class TestHostedFileStoragePath:
+    def test_hosted_paths_use_home(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "true")
+        monkeypatch.setenv("HOME", str(tmp_path))
+
+        server = ResponsesHostServer(MagicMock(context_providers=[]), store=InMemoryResponseProvider())
+
+        expected_session_path = tmp_path / ".sessions"
+        expected_approval_path = tmp_path / ".function_approvals" / "approval_requests.json"
+        assert server._session_storage_path == str(expected_session_path)  # pyright: ignore[reportPrivateUsage]
+        assert server._approval_storage_path == str(expected_approval_path)  # pyright: ignore[reportPrivateUsage]
+        assert isinstance(server._session_store, FoundrySessionStore)  # pyright: ignore[reportPrivateUsage]
+        assert server._session_store.storage_path == expected_session_path  # pyright: ignore[reportPrivateUsage]
+        assert isinstance(server._approval_storage, FileBasedFunctionApprovalStorage)  # pyright: ignore[reportPrivateUsage]
+        assert server._approval_storage._storage_path == str(expected_approval_path)  # pyright: ignore[reportPrivateUsage]
+
+        with _request_context(user_id="user-A"):
+            user_storage = server._approval_storage_for_request()  # pyright: ignore[reportPrivateUsage]
+        assert isinstance(user_storage, FileBasedFunctionApprovalStorage)
+        assert user_storage._storage_path == str(  # pyright: ignore[reportPrivateUsage]
+            tmp_path / ".function_approvals" / "user-A" / "approval_requests.json"
+        )
+
+    @pytest.mark.parametrize("home", [None, "/", "", "   "])
+    def test_hosted_paths_fall_back_to_default_session_directory(
+        self, monkeypatch: pytest.MonkeyPatch, home: str | None
+    ) -> None:
+        monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "true")
+        if home is None:
+            monkeypatch.delenv("HOME", raising=False)
+        else:
+            monkeypatch.setenv("HOME", home)
+
+        server = ResponsesHostServer(MagicMock(context_providers=[]), store=InMemoryResponseProvider())
+
+        assert server._session_storage_path == "/home/session/.sessions"  # pyright: ignore[reportPrivateUsage]
+        assert (  # pyright: ignore[reportPrivateUsage]
+            server._approval_storage_path == "/home/session/.function_approvals/approval_requests.json"
+        )
+
+    def test_local_paths_use_current_working_directory(self) -> None:
+        server = _make_server(MagicMock(context_providers=[]))
+
+        assert server._session_storage_path == os.path.join(os.getcwd(), ".sessions")  # pyright: ignore[reportPrivateUsage]
+        assert server._approval_storage_path == os.path.join(  # pyright: ignore[reportPrivateUsage]
+            os.getcwd(), ".function_approvals", "approval_requests.json"
+        )
+
+
 # endregion