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 49882462ce..30ab7ab10e 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/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 diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index cc1d863304..ed61144b00 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/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