Skip to content
Merged
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 @@ -391,6 +391,27 @@ class ResponsesHostServer(ResponsesAgentServerHost):
CHECKPOINT_STORAGE_PATH = "/.checkpoints"
FUNCTION_APPROVAL_STORAGE_PATH = "/.function_approvals/approval_requests.json"

@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,
Expand Down Expand Up @@ -439,11 +460,7 @@ 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.CHECKPOINT_STORAGE_PATH
if self.config.is_hosted
else os.path.join(os.getcwd(), self.CHECKPOINT_STORAGE_PATH.lstrip("/"))
)
self._checkpoint_storage_path = self._resolve_checkpoint_root(self.config.is_hosted)
self._is_workflow_agent = True

self._agent = agent
Expand Down
97 changes: 97 additions & 0 deletions python/packages/foundry_hosting/tests/test_responses.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,11 @@
from __future__ import annotations

import json
import os
import uuid
from collections.abc import AsyncIterator, Awaitable, Callable, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Literal, overload
from unittest.mock import AsyncMock, MagicMock, patch

Expand Down Expand Up @@ -4317,4 +4319,99 @@ async def test_round_trip_approval_response_rejected(self) -> None:
assert approval_responses[0].approved is False # type: ignore[attr-defined]


@pytest.mark.filterwarnings("ignore::DeprecationWarning")
class TestCheckpointStoragePath:
"""
In hosted mode, WorkflowAgent checkpoints must be stored under
$HOME/.checkpoints (durable across compute recreation), not
/.checkpoints (ephemeral root path that is wiped on idle).
"""

@staticmethod
def _make_mock_workflow_agent() -> MagicMock:
"""Create a mock WorkflowAgent for path-only tests."""
mock_workflow = MagicMock()
mock_workflow._runner_context.has_checkpointing.return_value = False
mock_workflow.name = "test-checkpoint-path"

mock_agent = MagicMock(spec=WorkflowAgent)
mock_agent.workflow = mock_workflow
mock_agent.context_providers = []

return mock_agent

def test_local_checkpoint_path_uses_cwd(self) -> None:
"""In local mode, checkpoints should be under cwd, NOT root `/`."""
mock_agent = self._make_mock_workflow_agent()
_original_isinstance = isinstance

def _patched_isinstance(obj: Any, cls: Any) -> bool:
if cls is WorkflowAgent:
return True
return _original_isinstance(obj, cls)

with patch(
"agent_framework_foundry_hosting._responses.isinstance",
side_effect=_patched_isinstance,
):
server = _make_server(mock_agent)

assert server._checkpoint_storage_path == os.path.join(os.getcwd(), ".checkpoints")

def test_hosted_checkpoint_path_uses_home(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""In hosted mode with valid HOME, checkpoints must be under $HOME/.checkpoints."""
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "true")
monkeypatch.setenv("HOME", "/home/testuser")
mock_agent = self._make_mock_workflow_agent()
_original_isinstance = isinstance

def _patched_isinstance(obj: Any, cls: Any) -> bool:
if cls is WorkflowAgent:
return True
return _original_isinstance(obj, cls)

with patch(
"agent_framework_foundry_hosting._responses.isinstance",
side_effect=_patched_isinstance,
):
server = ResponsesHostServer(mock_agent, store=InMemoryResponseProvider())

checkpoint_path = server._checkpoint_storage_path
assert checkpoint_path is not None
actual_normalized = checkpoint_path.replace("\\", "/")
assert actual_normalized.endswith("/home/testuser/.checkpoints")
assert not actual_normalized.startswith("/.checkpoints")

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/.checkpoints."""
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "true")
monkeypatch.delenv("HOME", raising=False)
mock_agent = self._make_mock_workflow_agent()

with patch(
"agent_framework_foundry_hosting._responses.isinstance",
side_effect=lambda o, c: c is WorkflowAgent or isinstance(o, c),
):
server = ResponsesHostServer(mock_agent, store=InMemoryResponseProvider())

assert server._checkpoint_storage_path == "/home/session/.checkpoints"

@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 /.checkpoints."""
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "true")
monkeypatch.setenv("HOME", bad_home)
mock_agent = self._make_mock_workflow_agent()

with patch(
"agent_framework_foundry_hosting._responses.isinstance",
side_effect=lambda o, c: c is WorkflowAgent or isinstance(o, c),
):
server = ResponsesHostServer(mock_agent, store=InMemoryResponseProvider())

assert server._checkpoint_storage_path == "/home/session/.checkpoints"
assert server._checkpoint_storage_path != "/.checkpoints"

# endregion
8 changes: 8 additions & 0 deletions python/packages/foundry_hosting/tests/test_responses_int.py
Original file line number Diff line number Diff line change
Expand Up @@ -259,6 +259,14 @@ async def test_input_image_url(self, server: ResponsesHostServer) -> None:
output_text = output_messages[0]["content"][0]["text"].lower()
assert "cat" in output_text

@pytest.mark.xfail(
reason=(
"Foundry Responses API rejects inline base64 data URIs in image_url with "
"'invalid_payload: ... is not a valid absolute URI'. It requires an absolute "
"http(s) URI or an uploaded file_id. Re-enable if Foundry adds data-URI support."
),
strict=False,
)
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_foundry_hosting_integration_tests_disabled
Expand Down
Loading