diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py index 5c9eca58596..8875b3a5d5b 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -412,6 +412,25 @@ def _resolve_checkpoint_root(is_hosted: bool) -> str: return "/home/session/.checkpoints" + @staticmethod + def _resolve_approval_storage_path(is_hosted: bool) -> str: + """Resolve function approval storage path. + + Hosted: $HOME/.function_approvals/approval_requests.json. + Local: {cwd}/.function_approvals/approval_requests.json. + """ + if not is_hosted: + return os.path.join(os.getcwd(), ".function_approvals", "approval_requests.json") + home = os.environ.get("HOME", "").strip() + if home and home != "/": + try: + resolved = Path(home).resolve() + if str(resolved) != str(resolved.root): + return str(resolved / ".function_approvals" / "approval_requests.json") + except (OSError, ValueError): + pass + return "/home/session/.function_approvals/approval_requests.json" + def __init__( self, agent: SupportsAgentRun, @@ -454,6 +473,7 @@ def __init__( self._is_workflow_agent = False self._checkpoint_storage_path = None + self._approval_storage_path = self._resolve_approval_storage_path(self.config.is_hosted) if isinstance(agent, WorkflowAgent): if agent.workflow._runner_context.has_checkpointing(): # pyright: ignore[reportPrivateUsage] raise RuntimeError( @@ -465,7 +485,7 @@ def __init__( self._agent = agent self._approval_storage = ( - FileBasedFunctionApprovalStorage(self.FUNCTION_APPROVAL_STORAGE_PATH) + FileBasedFunctionApprovalStorage(self._approval_storage_path) if self.config.is_hosted else InMemoryFunctionApprovalStorage() ) @@ -532,7 +552,7 @@ def _approval_storage_for_user(self, user_id: str | None) -> 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 diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 1dd847a7081..1948f3827cc 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -48,6 +48,7 @@ from typing_extensions import Any from agent_framework_foundry_hosting import ResponsesHostServer +from agent_framework_foundry_hosting._responses import FileBasedFunctionApprovalStorage from agent_framework_foundry_hosting._responses import ( _AZURE_RESPONSES_MESSAGE_ROLE_TYPE, # pyright: ignore[reportPrivateUsage] CONSENT_ERROR_CODE, @@ -3450,7 +3451,6 @@ def _helper() -> Callable[..., str]: return _approval_storage_path_for_user def test_user_id_scopes_path_under_base_directory(self, tmp_path: Any) -> None: - from pathlib import Path helper = self._helper() base = tmp_path / "approvals" / "requests.json" @@ -4376,7 +4376,7 @@ def _patched_isinstance(obj: Any, cls: Any) -> bool: ): server = ResponsesHostServer(mock_agent, store=InMemoryResponseProvider()) - checkpoint_path = server._checkpoint_storage_path + checkpoint_path = server._checkpoint_storage_path assert checkpoint_path is not None actual_normalized = checkpoint_path.replace("\\", "/") assert actual_normalized.endswith("/home/testuser/.checkpoints") @@ -4414,4 +4414,56 @@ def test_hosted_with_unusable_home_falls_back_to_default( assert server._checkpoint_storage_path == "/home/session/.checkpoints" assert server._checkpoint_storage_path != "/.checkpoints" + # endregion + + +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +class TestApprovalStoragePath: + """ + In hosted mode, function approval storage must be stored under + $HOME/.function_approvals (durable across compute recreation), not + /.function_approvals (ephemeral root path that is wiped on idle). + """ + + def test_local_approval_path_uses_cwd(self) -> None: + """In local mode, approval storage should be under cwd, NOT root `/`.""" + server = _make_server(MagicMock()) + expected = os.path.join(os.getcwd(), ".function_approvals", "approval_requests.json") + assert server._approval_storage_path == expected + + def test_hosted_approval_path_uses_home(self, monkeypatch: pytest.MonkeyPatch) -> None: + """In hosted mode with valid HOME, approvals must be under $HOME/.function_approvals/.""" + monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "true") + monkeypatch.setenv("HOME", "/home/testuser") + server = ResponsesHostServer(MagicMock(context_providers=[]), store=InMemoryResponseProvider()) + actual_normalized = server._approval_storage_path.replace("\\", "/") + assert actual_normalized.endswith("/home/testuser/.function_approvals/approval_requests.json") + assert not actual_normalized.startswith("/.function_approvals") + assert isinstance(server._approval_storage, FileBasedFunctionApprovalStorage) + storage_normalized = server._approval_storage._storage_path.replace("\\", "/") + assert storage_normalized.endswith("/home/testuser/.function_approvals/approval_requests.json") + user_storage = server._approval_storage_for_user("test-user") # pyright: ignore[reportPrivateUsage] + assert isinstance(user_storage, FileBasedFunctionApprovalStorage) + user_normalized = user_storage._storage_path.replace("\\", "/") # pyright: ignore[reportPrivateUsage] + assert user_normalized.endswith("/home/testuser/.function_approvals/test-user/approval_requests.json") + + def test_hosted_without_home_env_uses_default_session_dir(self, monkeypatch: pytest.MonkeyPatch) -> None: + """When HOME is unset in hosted mode, fall back to /home/session/.function_approvals/.""" + monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "true") + monkeypatch.delenv("HOME", raising=False) + server = ResponsesHostServer(MagicMock(), store=InMemoryResponseProvider()) + expected = "/home/session/.function_approvals/approval_requests.json" + assert server._approval_storage_path == expected + + @pytest.mark.parametrize("bad_home", ["/", "", " "]) + def test_hosted_with_unusable_home_falls_back_to_default( + self, monkeypatch: pytest.MonkeyPatch, bad_home: str + ) -> None: + """Filesystem-root or empty HOME must NOT produce /.function_approvals/.""" + monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "true") + monkeypatch.setenv("HOME", bad_home) + server = ResponsesHostServer(MagicMock(), store=InMemoryResponseProvider()) + expected = "/home/session/.function_approvals/approval_requests.json" + assert server._approval_storage_path == expected + assert not server._approval_storage_path.startswith("/.function_approvals")