Python: [Bug]: Hosted WorkflowAgent checkpoint is not restored after session compute recreation

[statefb]

Description

A Python WorkflowAgent exposed through ResponsesHostServer resumes correctly from a separate Responses request while its Microsoft Foundry Hosted session compute remains active.

After the same session is stopped and reaches idle, a continuation request recreates compute but does not restore the pending workflow checkpoint. The workflow starts from its initial state instead.

Expected behavior

A continuation request using the same agent_session_id, the initial previous_response_id, and the matching function_call_output should restore the pending request and complete the workflow after compute recreation.

Actual behavior

The continuation succeeds while compute remains active.

After compute recreation, it fails in the initial executor because the continuation contains only function_call_output. This indicates that execution restarted from the workflow entry point instead of the pending approval request.

Steps to reproduce

1. Deploy the code below as a Foundry Hosted Agent using the Responses protocol.

2. Create a hosted session and send a non-empty user message.

3. Capture the response ID and request_info call ID.

4. Stop the session compute:

   POST .../endpoint/sessions/{session-id}:stop?api-version=v1

5. Poll the session resource until status becomes idle:

   GET .../endpoint/sessions/{session-id}?api-version=v1

6. Send the continuation request below with the same session ID, response ID, and call ID.

7. Observe compute recreation followed by response.failed from StartExecutor instead of workflow completion.

Code Sample

from agent_framework import (
    Executor,
    Message,
    WorkflowBuilder,
    WorkflowContext,
    handler,
    response_handler,
)
from agent_framework_foundry_hosting import ResponsesHostServer


class StartExecutor(Executor):
    def __init__(self) -> None:
        super().__init__(id="start")

    @handler
    async def handle(
        self,
        messages: list[Message],
        ctx: WorkflowContext[str, str],
    ) -> None:
        if not messages or not messages[-1].text.strip():
            raise ValueError(
                "The approval spike requires a non-empty user message."
            )
        await ctx.send_message(messages[-1].text.strip())


class ApprovalExecutor(Executor):
    def __init__(self) -> None:
        super().__init__(id="approval")

    @handler
    async def request(
        self,
        value: str,
        ctx: WorkflowContext[str, str],
    ) -> None:
        await ctx.request_info(
            request_data=f"Approve: {value}",
            response_type=str,
        )

    @response_handler
    async def respond(
        self,
        original_request: str,
        response: str,
        ctx: WorkflowContext[str, str],
    ) -> None:
        await ctx.yield_output(
            f"Completed with decision: {response}"
        )


start = StartExecutor()
approval = ApprovalExecutor()

builder = WorkflowBuilder(
    start_executor=start,
    output_from=[approval],
)
builder.add_edge(start, approval)

workflow_agent = builder.build().as_agent(
    name="checkpoint-repro"
)

ResponsesHostServer(workflow_agent).run()

Error Messages / Stack Traces

Package Versions

agent-framework-core==1.11.0, agent-framework-foundry==1.10.1, agent-framework-foundry-hosting==1.0.0a260709, agent-framework-openai==1.10.1, azure-ai-agentserver-core==2.0.0b7, azure-ai-agentserver-responses==1.0.0b8, azure-ai-projects==2.3.0

Python Version

Python 3.13

Additional Context

The installed implementation rejects workflow-provided checkpoint storage and manages it in the host:

agent-framework/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

Lines 434 to 447 in 978cfcd

	self._is_workflow_agent = False
	self._checkpoint_storage_path = None
	if isinstance(agent, WorkflowAgent):
	if agent.workflow._runner_context.has_checkpointing(): # pyright: ignore[reportPrivateUsage]
	raise RuntimeError(
	"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._is_workflow_agent = True

It uses /.checkpoints and creates per-user/per-context FileCheckpointStorage instances:

agent-framework/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

Lines 253 to 287 in 978cfcd

	def _checkpoint_storage_for_context(root: str, context_id: str, *, user_id: str | None = None) -> FileCheckpointStorage:
	"""Build a ``FileCheckpointStorage`` for ``context_id`` rooted under ``root``.
	When the platform supplies a per-user partition key (``user_id``, from the
	``x-agent-user-id`` header on container protocol v2), the per-conversation
	checkpoint directory is nested under it: ``<root>/<user_id>/<context_id>``.
	This isolates each tenant's workflow state so one user can never restore or
	observe another user's checkpoint, even with a guessed or forged
	``context_id``. An absent (``None``) or empty ``user_id`` -- local
	development or protocol v1 -- falls back to the unscoped
	``<root>/<context_id>`` layout.
	Both ``context_id`` and ``user_id`` are validated as single safe path
	segments, and each resolved directory is verified to stay under its parent
	before any directory is created on disk (CWE-22).
	"""
	_validate_path_segment(context_id, kind="context id")
	base_path = Path(root).resolve()
	if user_id:
	_validate_path_segment(user_id, kind="user id")
	user_path = (base_path / user_id).resolve()
	if not user_path.is_relative_to(base_path):
	raise RuntimeError(f"Invalid user id: {user_id!r}")
	base_path = user_path
	storage_path = (base_path / context_id).resolve()
	if not storage_path.is_relative_to(base_path):
	raise RuntimeError(f"Invalid context id: {context_id!r}")
	return FileCheckpointStorage(
	storage_path,
	# Keep this provider-specific allowlist narrow. Hosted workflow
	# checkpoints can persist Azure's role enum inside Message objects.
	allowed_checkpoint_types=[_AZURE_RESPONSES_MESSAGE_ROLE_TYPE],
	)

agent-framework/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

Lines 386 to 447 in 978cfcd

	# region ResponsesHostServer
	class ResponsesHostServer(ResponsesAgentServerHost):
	"""A responses server host for an agent."""
	# TODO(@taochen): Allow a different checkpoint storage that stores checkpoints externally
	CHECKPOINT_STORAGE_PATH = "/.checkpoints"
	FUNCTION_APPROVAL_STORAGE_PATH = "/.function_approvals/approval_requests.json"
	def __init__(
	self,
	agent: SupportsAgentRun,
	*,
	prefix: str = "",
	options: ResponsesServerOptions | None = None,
	store: ResponseProviderProtocol | None = None,
	**kwargs: Any,
	) -> None:
	"""Initialize a ResponsesHostServer.
	Args:
	agent: The agent to handle responses for.
	prefix: The URL prefix for the server.
	options: Optional server options.
	store: Optional response store.
	**kwargs: Additional keyword arguments.
	Note:
	1. The agent must not have a history provider with `load_messages=True`,
	because history is managed by the hosting infrastructure.
	2. The agent must not have any context providers that maintain context
	in memory, because the hosting environment may get deactivated between
	requests, and any in-memory context would be lost.
	"""
	super().__init__(prefix=prefix, options=options, store=store, **kwargs)
	for provider in getattr(agent, "context_providers", []):
	if isinstance(provider, HistoryProvider) and provider.load_messages:
	raise RuntimeError(
	"There shouldn't be a history provider with `load_messages=True` already present. "
	"History is managed by the hosting infrastructure."
	)
	provider = cast(ContextProvider, provider)
	logger.warning(
	"Context provider %s is present. If it maintains context in memory, "
	"the context may be lost between requests. Use with caution.",
	provider.source_id,
	)
	self._is_workflow_agent = False
	self._checkpoint_storage_path = None
	if isinstance(agent, WorkflowAgent):
	if agent.workflow._runner_context.has_checkpointing(): # pyright: ignore[reportPrivateUsage]
	raise RuntimeError(
	"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._is_workflow_agent = True

Checkpoint restoration uses the inbound previous_response_id or conversation context:

agent-framework/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

Lines 691 to 780 in 978cfcd

	# Determine the latest checkpoint (if any) so we can resume the
	# workflow's prior state for this turn. The directory is keyed by
	# the inbound context id (conversation_id when set, otherwise
	# previous_response_id). Multi-turn declarative workflows need the
	# workflow's internal state (e.g. Conversation.messages,
	# intermediate Local.* variables) to survive across user turns;
	# the only place that state lives is the workflow checkpoint, so
	# on every turn we restore the latest checkpoint and feed the new
	# input back into the start executor as a continuation rather than
	# a fresh run.
	latest_checkpoint_id: str | None = None
	restore_storage: FileCheckpointStorage | None = None
	if context_id is not None:
	restore_storage = _checkpoint_storage_for_context(
	self._checkpoint_storage_path, context_id, user_id=user_id
	)
	latest_checkpoint = await restore_storage.get_latest(workflow_name=self._agent.workflow.name)
	if latest_checkpoint is not None:
	latest_checkpoint_id = latest_checkpoint.checkpoint_id
	# Storage that will receive checkpoints written during this turn.
	# When the caller chains with previous_response_id, the next turn
	# will reference the current response_id as its previous_response_id,
	# so new checkpoints must land under the current response_id (or the
	# conversation_id when set). When conversation_id is set, this
	# matches restore_storage; when only previous_response_id was
	# supplied, restore_storage points at the *prior* response's
	# directory and write_storage points at the *current* response's.
	write_context_id = context.conversation_id or context.response_id
	write_storage = _checkpoint_storage_for_context(
	self._checkpoint_storage_path, write_context_id, user_id=user_id
	)
	# Multi-turn pattern: when we have a prior checkpoint, restore it
	# first (drive the workflow back to idle with prior state intact),
	# then make a separate call that delivers the new user input. This
	# depends on Workflow.run preserving shared state across calls. The
	# restore-only call may yield events from any pending in-flight
	# work in the checkpoint; we consume those internally here so they
	# don't surface to the response stream as duplicates.
	#
	# If the restored checkpoint had pending request_info events, the
	# restore-only call replays them through
	# ``WorkflowAgent._convert_workflow_event_to_agent_response_updates``
	# and populates ``self._agent.pending_requests``. That is the correct
	# state: those requests are genuinely outstanding, and the next
	# ``run(input_messages, ...)`` call may contain ``function_call_output``
	# items (carried as FunctionResult/FunctionApprovalResponse content)
	# that fulfill them via :meth:`WorkflowAgent._process_pending_requests`.
	if latest_checkpoint_id is not None:
	if is_streaming_request:
	async for _ in self._agent.run(
	stream=True,
	checkpoint_id=latest_checkpoint_id,
	checkpoint_storage=restore_storage,
	):
	pass
	else:
	await self._agent.run(
	stream=False,
	checkpoint_id=latest_checkpoint_id,
	checkpoint_storage=restore_storage,
	)
	if not is_streaming_request:
	# Run the agent in non-streaming mode with the new user input.
	response = await self._agent.run(
	input_messages,
	stream=False,
	checkpoint_storage=write_storage,
	)
	async for item in _to_outputs_for_messages(
	response_event_stream,
	response.messages,
	approval_storage=approval_storage,
	):
	yield item
	await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name)
	yield response_event_stream.emit_completed()
	return
	tracker = _OutputItemTracker(response_event_stream)
	# Run the workflow agent in streaming mode with the new user input.
	async for update in self._agent.run(
	input_messages,
	stream=True,
	checkpoint_storage=write_storage,

A related .NET fix moved its hosted default from /.checkpoints to $HOME/.checkpoints:

#6714

The .NET failure mode was not identical, but the fix documents $HOME as the durable hosted location, noted on Learn Documentation.

[github-actions]

🤖 Automated triage reproduction notes (agent-authored — trust but verify)

Agent analysis

Repro: ResponsesHostServer.init() in python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py (lines 442-446) sets _checkpoint_storage_path to the raw class constant "/.checkpoints" when config.is_hosted is True, instead of re-rooting under Path.home(). Minimal repro: construct a ResponsesHostServer with a mock WorkflowAgent in hosted mode (FOUNDRY_HOSTING_ENVIRONMENT=true) and assert _checkpoint_storage_path starts with str(Path.home(). The fix should introduce a _resolve_storage_path helper that uses Path.home() for hosted and Path.cwd() for local, as done in unmerged commit 596538f.

• Failing test: python/packages/foundry_hosting/tests/test_checkpoint_path_bug_7137.py
• Files examined: python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py, python/packages/foundry_hosting/tests/test_responses.py, python/packages/foundry_hosting/pyproject.toml
• Tests run: python/packages/foundry_hosting/tests/test_checkpoint_path_bug_7137.py::TestCheckpointStoragePathBug7137::test_hosted_checkpoint_path_is_under_home_not_root, python/packages/foundry_hosting/tests/test_checkpoint_path_bug_7137.py::TestCheckpointStoragePathBug7137::test_checkpoint_storage_path_constant_is_not_home_aware, python/packages/foundry_hosting/tests/test_checkpoint_path_bug_7137.py::TestCheckpointStoragePathBug7137::test_init_sets_hosted_checkpoint_path_to_root
• Reported version: 1.0.0a260709
• Current version: 1.0.0a260709

[impeachmentright]

Durable WorkflowAgent checkpoints are the resume gate for paid peers: quote -> approve -> settle fails if session compute recreation drops the checkpoint.

Live A2A contractor with gated tools: https://a2a.elonsusk.com/.well-known/agent-card.json

If an external checkpoint-restore probe would help, happy to assist. Otherwise ignore.