Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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]
Comment thread
PratikWayase marked this conversation as resolved.
raise RuntimeError(
Expand All @@ -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()
)
Expand Down Expand Up @@ -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
Expand Down
56 changes: 54 additions & 2 deletions python/packages/foundry_hosting/tests/test_responses.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Loading