[BUG] kickoff_for_each_async / akickoff_for_each wipe replay data on every run

Description

Crew.kickoff_for_each_async and Crew.akickoff_for_each both delegate to the shared helper run_for_each_async in lib/crewai/src/crewai/crews/utils.py. After gathering results from all input copies, that helper unconditionally resets the shared replay store:

https://github.com/crewAIInc/crewAI/blob/main/lib/crewai/src/crewai/crews/utils.py#L500-L514

    results = await asyncio.gather(*async_tasks)

    total_usage_metrics = UsageMetrics()
    for crew_copy in crew_copies:
        if crew_copy.usage_metrics:
            total_usage_metrics.add_usage_metrics(crew_copy.usage_metrics)
    crew.usage_metrics = total_usage_metrics

    crew._task_output_handler.reset()
    return list(results)

Each crew copy already resets and repopulates the shared TaskOutputStorageHandler (backed by a single SQLite file) as part of its own kickoff_async/akickoff run. The crew._task_output_handler.reset() call above then deletes whatever the copies just persisted, immediately before returning.

This is the async/native-async counterpart of #6650 (which covers the synchronous kickoff_for_each); the fix for #6650 only touches Crew.kickoff_for_each in crew.py and does not touch run_for_each_async, so this path is still broken.

Steps to Reproduce

import asyncio
from unittest.mock import patch
from crewai import Agent, Crew, Task, Process
from crewai.tasks.task_output import TaskOutput

agent = Agent(role="Researcher", goal="Research a topic.", backstory="You are a careful researcher.")
task = Task(description="Research {topic}.", expected_output="A concise research note.", agent=agent)
crew = Crew(agents=[agent], tasks=[task], process=Process.sequential)

first_output = TaskOutput(description="Research first.", raw="first result", agent="Researcher")
latest_output = TaskOutput(description="Research latest.", raw="latest result", agent="Researcher")

async def main():
    with patch.object(Task, "execute_sync", side_effect=[first_output, latest_output]):
        results = await crew.kickoff_for_each_async(inputs=[{"topic": "first"}, {"topic": "latest"}])
    print("stored:", crew._task_output_handler.load())

asyncio.run(main())

Observed on unmodified main:

stored: []

Expected behavior

After kickoff_for_each_async (or akickoff_for_each) completes, crew._task_output_handler.load() should return the persisted output of at least one of the runs, so Crew.replay(task_id) has something to replay. Instead it is always empty.

Note: because the input copies for the async variants execute concurrently on separate threads against one shared SQLite-backed store, which copy's output "wins" is not deterministic the way it is for the sequential kickoff_for_each (where the last input in the list is guaranteed to be the one left standing). That ordering guarantee is a separate, harder problem; this issue is specifically about the unconditional reset silently discarding every run's output.

Possible Solution

Remove the trailing crew._task_output_handler.reset() call in run_for_each_async (lib/crewai/src/crewai/crews/utils.py), mirroring the fix applied to the synchronous path for #6650.

Environment

• crewai: main branch (commit f15844b2)
• Python 3.13

Additional context

This issue was found and this fix was prepared with AI assistance (Claude Code), per this repo's llm-generated labeling requirement in .github/CONTRIBUTING.md.

Start in lib/crewai/src/crewai/crews/utils.py at run_for_each_async, called by Crew.kickoff_for_each_async and Crew.akickoff_for_each, and inspect the trailing task-output reset after the gathered results are processed. Run the supplied reproduction with two inputs; done means the shared handler retains persisted output after completion so Crew.replay(task_id) has data.