agent/agent-framework
1
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Address Tao's review on PR 5531

Browse Source 
- Rename Workflow._run_workflow_with_tracing parameter
  is_fresh_message_run -> is_continuation (default False, inverted).
  Fresh-message turns reset per-run accounting; continuations
  (checkpoint restores, responses replays) preserve it.
- Simplify the in-flight-messages guard: _validate_run_params already
  enforces that 'message' is mutually exclusive with 'checkpoint_id'
  and 'responses', so the additional checks were dead code.
- foundry_hosting _responses: move the restore-only pre-pass above
  emit_created/emit_in_progress; restore is preparation, not run
  progress. Drop the skip-restore gate (state preservation requires
  unconditional restore) and instead clear agent.pending_requests
  after the restore-only call. Collapse over-conditioned check.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
alliscode
2026-04-28 14:02:53 -07:00
Unverified
parent d34a4834eb
commit 31a0011063
2 changed files with 44 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/core/agent_framework/_workflows/_workflow.py
+23 -24
View File
@@ -299,7 +299,7 @@ class Workflow(DictConvertible):
async def _run_workflow_with_tracing(
self,
initial_executor_fn: Callable[[], Awaitable[None]] | None = None,
is_fresh_message_run: bool = True,
is_continuation: bool = False,
streaming: bool = False,
function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None,
client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None,
@@ -311,12 +311,13 @@ class Workflow(DictConvertible):
Args:
initial_executor_fn: Optional function to execute initial executor.
is_fresh_message_run: True when this run is a fresh new turn delivered
via the start executor (i.e. ``message`` is provided without a
``checkpoint_id`` or ``responses``). Resets per-run accounting
(iteration counter and run kwargs) without touching the shared
workflow state. False for checkpoint restores and responses-only
runs, which are continuations of prior work.
is_continuation: True when this run is a continuation of prior
work (a checkpoint restore or a responses-only replay) rather
than a fresh new turn delivered via the start executor with
``message=...``. Continuations preserve per-run accounting
(iteration counter and run kwargs) from the prior turn;
fresh-message runs reset them. Shared workflow state is
preserved in both cases.
streaming: Whether to enable streaming mode for agents.
function_invocation_kwargs: Optional kwargs to store in State for function
invocations in subagents.
@@ -358,7 +359,7 @@ class Workflow(DictConvertible):
# the same instance and have prior turns' context survive.
# Iteration counting and per-run kwargs ARE per-run though,
# so they're reset here.
if is_fresh_message_run:
if not is_continuation:
self._runner.reset_iteration_count()
# Store run kwargs in State so executors can access them.
@@ -381,7 +382,7 @@ class Workflow(DictConvertible):
client_kwargs, "client_kwargs"
)
self._state.set(WORKFLOW_RUN_KWARGS_KEY, combined_kwargs)
elif is_fresh_message_run:
elif not is_continuation:
self._state.set(WORKFLOW_RUN_KWARGS_KEY, {})
self._state.commit() # Commit immediately so kwargs are available
@@ -601,20 +602,18 @@ class Workflow(DictConvertible):
if checkpoint_storage is not None:
self._runner.context.set_runtime_checkpoint_storage(checkpoint_storage)
# Async validation: a fresh-message run (no checkpoint, no responses)
# is only allowed when the runner context has fully drained from any
# prior run. If it still has in-flight executor messages, the prior
# run didn't complete - the caller must either resume from a
# checkpoint or wait for the prior run to drain. (Pending request_info
# events are intentionally NOT blocked here: a follow-up run with
# message=... is the normal way to deliver a response to those
# pending requests, e.g. via WorkflowAgent._process_pending_requests.)
if (
message is not None
and checkpoint_id is None
and responses is None
and await self._runner.context.has_messages()
):
# Async validation: a fresh-message run is only allowed when the
# runner context has fully drained from any prior run. If it still
# has in-flight executor messages, the prior run didn't complete -
# the caller must either resume from a checkpoint or wait for the
# prior run to drain. (Pending request_info events are intentionally
# NOT blocked here: a follow-up run with message=... is the normal
# way to deliver a response to those pending requests, e.g. via
# WorkflowAgent._process_pending_requests.)
# NOTE: _validate_run_params already enforces that ``message`` is
# mutually exclusive with both ``checkpoint_id`` and ``responses``,
# so we don't need to re-check those here.
if message is not None and await self._runner.context.has_messages():
raise RuntimeError(
"Cannot start a new run with 'message' while in-flight executor "
"messages remain from a prior run. Resume from a checkpoint "
@@ -629,7 +628,7 @@ class Workflow(DictConvertible):
async for event in self._run_workflow_with_tracing(
initial_executor_fn=initial_executor_fn,
is_fresh_message_run=(message is not None and checkpoint_id is None and responses is None),
is_continuation=(message is None),
streaming=streaming,
function_invocation_kwargs=function_invocation_kwargs,
client_kwargs=client_kwargs,
python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+21 -29
View File
@@ -284,33 +284,12 @@ class ResponsesHostServer(ResponsesAgentServerHost):
# a fresh run.
latest_checkpoint_id: str | None = None
restore_storage: FileCheckpointStorage | None = None
latest_checkpoint = None
if context_id is not None:
restore_storage = FileCheckpointStorage(os.path.join(self._checkpoint_storage_path, context_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
# If the latest checkpoint represents a workflow that was idle with
# pending request_info events (human-in-the-loop interrupts), the
# restore-only pre-pass below would replay those events through
# ``WorkflowAgent._convert_workflow_event_to_agent_response_updates``,
# populating ``self._agent.pending_requests``. The subsequent
# ``run(input_messages, ...)`` call would then route through
# :meth:`WorkflowAgent._process_pending_requests`, which expects
# function-response content and rejects plain text input. The host
# currently does not support resuming workflows with outstanding
# request_info via plain-text user turns, so in that scenario we
# skip the restore-only pre-pass and start a fresh turn. State
# accumulated by purely state-preserving workflows (no request_info)
# is unaffected.
skip_restore_due_to_pending_requests = bool(
latest_checkpoint is not None and latest_checkpoint.pending_request_info_events
)
# Now run the agent with the latest input
response_event_stream = ResponseEventStream(response_id=context.response_id, model=request.model)
# 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,
@@ -322,9 +301,6 @@ class ResponsesHostServer(ResponsesAgentServerHost):
write_context_id = context.conversation_id or context.response_id
write_storage = FileCheckpointStorage(os.path.join(self._checkpoint_storage_path, write_context_id))
yield response_event_stream.emit_created()
yield response_event_stream.emit_in_progress()
# 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
@@ -332,11 +308,20 @@ class ResponsesHostServer(ResponsesAgentServerHost):
# 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 (
latest_checkpoint_id is not None
and restore_storage is not None
and not skip_restore_due_to_pending_requests
):
#
# 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 would route
# the subsequent ``run(input_messages, ...)`` call through
# :meth:`WorkflowAgent._process_pending_requests`, which expects
# function-response content and rejects plain text. The host's
# contract for plain-text user input is "treat as new turn", not
# "respond to outstanding request_info", so we clear
# ``pending_requests`` after the restore-only pre-pass. State
# (Conversation.*, Inputs.*, Local.*) lives in the workflow
# checkpoint and is unaffected by this clear.
if latest_checkpoint_id is not None:
if is_streaming_request:
async for _ in self._agent.run(
stream=True,
@@ -350,6 +335,13 @@ class ResponsesHostServer(ResponsesAgentServerHost):
checkpoint_id=latest_checkpoint_id,
checkpoint_storage=restore_storage,
)
self._agent.pending_requests.clear()
# Now run the agent with the latest input
response_event_stream = ResponseEventStream(response_id=context.response_id, model=request.model)
yield response_event_stream.emit_created()
yield response_event_stream.emit_in_progress()
if not is_streaming_request:
# Run the agent in non-streaming mode with the new user input.