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

@@ -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(

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
 

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)