fix: set tags before context manager closes (#8911)

Author: mathislucka

haystack/core/pipeline/async_pipeline.py

@@ -212,29 +212,29 @@ async def _run_component_async(component_name: str, component_inputs: Dict[str,
                         loop = asyncio.get_running_loop()
                         outputs = await loop.run_in_executor(None, lambda: instance.run(**component_inputs))
 
-                component_visits[component_name] += 1
-
-                if not isinstance(outputs, dict):
-                    raise PipelineRuntimeError(
-                        f"Component '{component_name}' returned an invalid output type. "
-                        f"Expected a dict, but got {type(outputs).__name__} instead. "
+                    component_visits[component_name] += 1
+
+                    if not isinstance(outputs, dict):
+                        raise PipelineRuntimeError(
+                            f"Component '{component_name}' returned an invalid output type. "
+                            f"Expected a dict, but got {type(outputs).__name__} instead. "
+                        )
+
+                    span.set_tag("haystack.component.visits", component_visits[component_name])
+                    span.set_content_tag("haystack.component.outputs", deepcopy(outputs))
+
+                    # Distribute outputs to downstream inputs; also prune outputs based on `include_outputs_from`
+                    pruned = self._write_component_outputs(
+                        component_name=component_name,
+                        component_outputs=outputs,
+                        inputs=inputs_state,
+                        receivers=cached_receivers[component_name],
+                        include_outputs_from=include_outputs_from,
                     )
+                    if pruned:
+                        pipeline_outputs[component_name] = pruned
 
-                span.set_tag("haystack.component.visits", component_visits[component_name])
-                span.set_content_tag("haystack.component.outputs", deepcopy(outputs))
-
-                # Distribute outputs to downstream inputs; also prune outputs based on `include_outputs_from`
-                pruned = self._write_component_outputs(
-                    component_name=component_name,
-                    component_outputs=outputs,
-                    inputs=inputs_state,
-                    receivers=cached_receivers[component_name],
-                    include_outputs_from=include_outputs_from,
-                )
-                if pruned:
-                    pipeline_outputs[component_name] = pruned
-
-                return pruned
+                    return pruned
 
             async def _run_highest_in_isolation(component_name: str) -> AsyncIterator[Dict[str, Any]]:
                 """