diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index a7c47c2893..adea4bda20 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -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, diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py index d984ec644c..103a7273ba 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -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.