From 73bdfaad56f4beacf23de36f4474b73043c7d238 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Sat, 11 Jul 2026 19:20:03 -0300 Subject: [PATCH] fix: correct execution-boundary hook ordering and input aliasing Reworks the crew and flow boundary seams flagged in review. `OUTPUT` and `EXECUTION_END` now run before the completion event (`CrewKickoffCompletedEvent` and `FlowFinishedEvent`) so a `HookAborted` no longer leaves a spurious completed signal and a returned payload replacement is honored on the emitted and returned result. Boundary contexts alias `inputs` to the same object as `payload` instead of a fresh dict from `or`, so in-place edits survive read-back. Flows re-publish the resolved inputs into `flow_inputs` baggage after the `INPUT` hook so trigger-payload injection observes hook rewrites, and a resumed flow now dispatches `OUTPUT`/`EXECUTION_END` on its completion path. --- lib/crewai/src/crewai/crew.py | 42 ++++++++------- lib/crewai/src/crewai/crews/utils.py | 13 ++++- .../src/crewai/flow/runtime/__init__.py | 52 ++++++++++++++++--- 3 files changed, 80 insertions(+), 27 deletions(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index 2179458c8..db49b3a1e 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -1906,6 +1906,30 @@ class Crew(FlowTrackable, BaseModel): final_string_output = final_task_output.raw self._finish_execution(final_string_output) self.token_usage = self.calculate_usage_metrics() + + from crewai.hooks.contexts import ExecutionEndContext, OutputContext + from crewai.hooks.dispatch import InterceptionPoint, dispatch + + crew_output = CrewOutput( + raw=final_task_output.raw, + pydantic=final_task_output.pydantic, + json_dict=final_task_output.json_dict, + tasks_output=task_outputs, + token_usage=self.token_usage, + ) + + # OUTPUT/EXECUTION_END run before the kickoff-completed event (mirroring + # the flow OUTPUT-before-FlowFinishedEvent ordering) so a HookAborted + # prevents a spurious completed signal and any payload replacement is + # honored on the returned output. + output_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output) + dispatch(InterceptionPoint.OUTPUT, output_ctx) + crew_output = output_ctx.payload + + end_ctx = ExecutionEndContext(crew=self, output=crew_output, payload=crew_output) + dispatch(InterceptionPoint.EXECUTION_END, end_ctx) + crew_output = end_ctx.payload + # Ensure background memory saves finish (and emit their # completed/failed events) before the kickoff-completed event below # triggers listener teardown/finalization. @@ -1924,24 +1948,6 @@ class Crew(FlowTrackable, BaseModel): # Finalization is handled by trace listener (always initialized) # The batch manager checks contextvar to determine if tracing is enabled - from crewai.hooks.contexts import ExecutionEndContext, OutputContext - from crewai.hooks.dispatch import InterceptionPoint, dispatch - - crew_output = CrewOutput( - raw=final_task_output.raw, - pydantic=final_task_output.pydantic, - json_dict=final_task_output.json_dict, - tasks_output=task_outputs, - token_usage=self.token_usage, - ) - - output_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output) - dispatch(InterceptionPoint.OUTPUT, output_ctx) - crew_output = output_ctx.payload - - end_ctx = ExecutionEndContext(crew=self, output=crew_output, payload=crew_output) - dispatch(InterceptionPoint.EXECUTION_END, end_ctx) - return crew_output def _process_async_tasks( diff --git a/lib/crewai/src/crewai/crews/utils.py b/lib/crewai/src/crewai/crews/utils.py index c096a8fd2..bfac589ec 100644 --- a/lib/crewai/src/crewai/crews/utils.py +++ b/lib/crewai/src/crewai/crews/utils.py @@ -289,8 +289,13 @@ def prepare_kickoff( ) normalized = dict(inputs) + # ``inputs`` aliases the same object as ``payload`` (not a fresh ``{}`` from + # ``or``) so in-place edits to either survive read-back, per the context + # contract. ``None`` inputs are preserved rather than coerced to ``{}``. start_ctx = ExecutionStartContext( - crew=crew, inputs=normalized or {}, payload=normalized + crew=crew, + inputs=normalized if normalized is not None else {}, + payload=normalized, ) dispatch(InterceptionPoint.EXECUTION_START, start_ctx) normalized = start_ctx.payload @@ -300,7 +305,11 @@ def prepare_kickoff( normalized = {} normalized = before_callback(normalized) - input_ctx = InputContext(crew=crew, inputs=normalized or {}, payload=normalized) + input_ctx = InputContext( + crew=crew, + inputs=normalized if normalized is not None else {}, + payload=normalized, + ) dispatch(InterceptionPoint.INPUT, input_ctx) normalized = input_ctx.payload diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 201ec4c04..812d83268 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -1476,6 +1476,22 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): else (resumed_method_output if emit else result) ) + # A resumed flow completes here rather than in kickoff_async, so the + # OUTPUT/EXECUTION_END seams must fire on this path too (before + # FlowFinishedEvent) to expose the final result to policy hooks. + from crewai.hooks.contexts import ExecutionEndContext, OutputContext + from crewai.hooks.dispatch import InterceptionPoint, dispatch + + output_ctx = OutputContext(flow=self, output=final_result, payload=final_result) + dispatch(InterceptionPoint.OUTPUT, output_ctx) + final_result = output_ctx.payload + + end_ctx = ExecutionEndContext( + flow=self, output=final_result, payload=final_result + ) + dispatch(InterceptionPoint.EXECUTION_END, end_ctx) + final_result = end_ctx.payload + if self._event_futures: await asyncio.gather( *[ @@ -2037,6 +2053,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): flow_name_token = None flow_defer_trace_finalization_token = None request_id_token = None + # Re-published after the INPUT hook so trigger-payload injection reads + # the hook-rewritten inputs rather than the pre-hook baggage above. + flow_inputs_token = None if current_flow_id.get() is None: flow_id_token = current_flow_id.set(self.flow_id) flow_name_token = current_flow_name.set( @@ -2070,16 +2089,29 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): ) from crewai.hooks.dispatch import InterceptionPoint, dispatch + # ``inputs`` aliases the same object as ``payload`` (not a fresh + # ``{}`` from ``or``) so in-place edits survive read-back. start_ctx = ExecutionStartContext( - flow=self, inputs=inputs or {}, payload=inputs + flow=self, + inputs=inputs if inputs is not None else {}, + payload=inputs, ) dispatch(InterceptionPoint.EXECUTION_START, start_ctx) inputs = start_ctx.payload - input_ctx = InputContext(flow=self, inputs=inputs or {}, payload=inputs) + input_ctx = InputContext( + flow=self, + inputs=inputs if inputs is not None else {}, + payload=inputs, + ) dispatch(InterceptionPoint.INPUT, input_ctx) inputs = input_ctx.payload + # Publish the resolved inputs so trigger-payload injection and other + # baggage readers observe hook rewrites (the baggage set before the + # hooks carried the pre-hook inputs). + flow_inputs_token = attach(baggage.set_baggage("flow_inputs", inputs or {})) + # Reset flow state for fresh execution unless restoring from persistence is_restoring = ( inputs and "id" in inputs and self.persistence is not None @@ -2321,6 +2353,15 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): dispatch(InterceptionPoint.OUTPUT, output_ctx) final_output = output_ctx.payload + # EXECUTION_END runs before FlowFinishedEvent so a HookAborted + # prevents a spurious finished signal and payload replacement is + # honored on the emitted result and the returned value. + end_ctx = ExecutionEndContext( + flow=self, output=final_output, payload=final_output + ) + dispatch(InterceptionPoint.EXECUTION_END, end_ctx) + final_output = end_ctx.payload + if self._event_futures: await asyncio.gather( *[asyncio.wrap_future(f) for f in self._event_futures] @@ -2371,11 +2412,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): else: trace_listener.batch_manager.finalize_batch() - end_ctx = ExecutionEndContext( - flow=self, output=final_output, payload=final_output - ) - dispatch(InterceptionPoint.EXECUTION_END, end_ctx) - return final_output finally: # Safety net for the exception path; the success path already @@ -2399,6 +2435,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): current_flow_name.reset(flow_name_token) if flow_id_token is not None: current_flow_id.reset(flow_id_token) + if flow_inputs_token is not None: + detach(flow_inputs_token) detach(flow_token) crewai_event_bus._exit_runtime_scope(runtime_scope)