Python: Clarify and fix message/checkpoint_id/responses interaction semantics in Workflow.run() / WorkflowAgent.run() #7677

Summary

Workflow.run() and WorkflowAgent.run() accept three independent-ish inputs — message, checkpoint_id, and responses — whose combined semantics are only partially specified and, in at least one combination, appear to silently drop workflow output. This surfaced while hardening ResponsesHostServer (packages/foundry_hosting) crash-recovery for resilient_background=True workflow agents, and warrants a dedicated investigation + spec before the workaround currently in agent_framework_foundry_hosting/_responses.py becomes the de facto contract.

Current validation rules (Workflow._validate_run_params, packages/core/agent_framework/_workflows/_workflow.py)

• message + responses: mutually exclusive (raises ValueError)
• message + checkpoint_id: mutually exclusive (raises ValueError)
• checkpoint_id + responses: allowed — "restore then send" (_restore_and_send_responses)
• At least one of the three must be provided

So at the Workflow.run() level, message can never be combined with checkpoint_id in a single call.

Where the ambiguity/bug shows up: WorkflowAgent.run() / WorkflowAgent._run_core()

packages/core/agent_framework/_workflows/_agent.py (_run_core, ~line 420) simulates a message + checkpoint_id combination by issuing two sequential separate calls to the underlying Workflow:

if checkpoint_id is not None:
    ...
    if streaming:
        async for _ in self.workflow.run(stream=True, checkpoint_id=checkpoint_id, checkpoint_storage=checkpoint_storage):
            pass          # <-- events from this phase are unconditionally discarded
    else:
        _ = await self.workflow.run(checkpoint_id=checkpoint_id, checkpoint_storage=checkpoint_storage)
    if not input_messages:
        logger.info("No input messages provided; the workflow has been restored to the checkpoint state.")
        return            # <-- returns without yielding anything at all

final_state = self._workflow.status
...
elif final_state == WorkflowRunState.IDLE:
    async for event in self.workflow.run(message=input_messages, stream=True, ...):
        yield event

This assumes checkpoint restoration is purely "replay old history to get back to idle," so any events it produces are safe to discard, and that genuinely new work only ever happens in the follow-up message=... call. That assumption breaks down for self-driving workflows resumed mid-flight (e.g. a workflow with an executor self-edge queued to continue automatically, restored from a checkpoint taken while it was still actively running): the discard-only restore call can run the workflow all the way to completion internally — genuinely new output, not replayed history — and every event from that is thrown away. If input_messages is empty (the natural case for crash recovery, where there's no new user turn to deliver), the caller receives nothing, even though the workflow fully executed.

Repro context

Sample: samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow — a workflow with a countdown executor that self-edges (add_edge(countdown, countdown)) to decrement a counter once per second via ctx.yield_output(). Using ResponsesHostServer(..., options=ResponsesServerOptions(resilient_background=True)) and crashing the process mid-countdown, then restarting and calling WorkflowAgent.run(stream=True, checkpoint_id=<last checkpoint>, checkpoint_storage=...) (no messages, per the resilience contract's recovery flow) resulted in the workflow silently completing the entire remaining countdown with zero forwarded events.

Current workaround (not proposed as the fix — flagging as evidence + stopgap)

agent_framework_foundry_hosting/_responses.py (ResponsesHostServer._resume_workflow_from_checkpoint) now bypasses WorkflowAgent.run() for this case entirely, driving agent.workflow.run(stream=True, checkpoint_id=..., checkpoint_storage=...) directly and converting each raw WorkflowEvent via WorkflowAgent._convert_workflow_event_to_agent_response_updates (a protected method, accessed cross-module with pyright: ignore[reportPrivateUsage]). This isn't a great long-term answer: it duplicates/depends on WorkflowAgent internals and doesn't address the underlying ambiguity for other callers of WorkflowAgent.run().

Questions to resolve

1. What is the intended contract for checkpoint_id alone (no message, no responses) at the WorkflowAgent level?
   • "Restore to idle only, discard everything, caller must follow up with message=... or responses=..."? If so, should WorkflowAgent.run() guarantee the workflow is actually idle-and-waiting after the restore call, rather than potentially fully completing?
   • Or should message-less checkpoint resume forward every event produced while draining the checkpoint's own queued work (the behavior a resumed background/self-driving workflow needs)?
2. What is the correct behavior at WorkflowAgent when message is supplied together with checkpoint_id? Workflow.run() itself rejects this combination, but WorkflowAgent.run() effectively performs it via two sequential calls. Is that composition itself well-defined for all workflow states the restore can leave behind (idle, idle-with-pending-requests, or — as found here — freshly completed)? What should happen if the restore call already drives the workflow to a terminal state and a message is still supplied?
3. Should Workflow.status distinguish "idle, restored, ready for new input" from "idle, just completed via its own internal queued work during this very restore call"? Right now both look like WorkflowRunState.IDLE to the caller.
4. Is accessing WorkflowAgent._convert_workflow_event_to_agent_response_updates from outside the class (as the current workaround does) an acceptable pattern for hosting layers that need raw event conversion, or should this be promoted to a supported/public API?

Suggested next steps

• Write out the full intended state-transition table for (message, checkpoint_id, responses) combinations at both the Workflow and WorkflowAgent layers, including what happens to events produced during a restore-only phase.
• Decide whether WorkflowAgent.run(checkpoint_id=..., messages=None) should forward restore-phase events (opt-in or by default) instead of discarding them.
• Add test coverage for "checkpoint being restored has its own queued in-flight work that completes the workflow during the restore phase" (currently untested — this is exactly the resilient-background crash-recovery scenario).
• Revisit the _convert_workflow_event_to_agent_response_updates protected-access workaround in agent_framework_foundry_hosting once the above is settled.

References

• packages/core/agent_framework/_workflows/_agent.py — WorkflowAgent._run_core, _run_stream_impl, _convert_workflow_event_to_agent_response_updates
• packages/core/agent_framework/_workflows/_workflow.py — Workflow.run, _validate_run_params, _run_core
• packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py — ResponsesHostServer._resume_workflow_from_checkpoint (workaround)
• Sample: samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow