microsoft / microsoft/agent-framework

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

Open
#7,677 0 comments 0 reactions 1 assignee Claimed by @TaoChenOSU View on GitHub
python workflows
Dominant language
Python
Stars
13.6k
Forks
2.3k
Avg merge
2d 45m
Merged PRs (30d)
358

Description

## 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`:

```python
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=, 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`

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.