Fix ParallelExecutor crash when Parallel called from within module forward()

[LukeDevs]

## Summary

Fixes a bug where dspy.Parallel silently returns None for all tasks when called from within a module's forward() method while MLflow autologging is enabled.

**## Problem

When using mlflow.dspy.autolog(), calling dspy.Parallel from inside a module's forward() method causes all results to silently return None.

The root cause is twofold:

1. Incorrect ContextVar access in parallelizer.py (line 95)

thread_local_overrides is a ContextVar, which requires .get() and .set() methods for access. The existing code incorrectly attempts to access an .overrides attribute directly:

thread_local_overrides.overrides["usage_tracker"] = copy.deepcopy(parent_overrides["usage_tracker"])

This raises AttributeError: '_contextvars.ContextVar' object has no attribute 'overrides'.

The bug only manifests when usage_tracker is present in the parent context, which occurs when MLflow autologging is enabled.

2. Silent exception handling (line ~155)

Worker exceptions are caught and silently discarded:

try:
    index, outcome = f.result()
except Exception:
    pass

This masked the underlying AttributeError.

**## Solution

1. Restructured ContextVar access to build the complete overrides dictionary before setting:

original = thread_local_overrides.get()
new_overrides = {**original, **parent_overrides.copy()}
if parent_overrides.get("usage_tracker"):
    new_overrides["usage_tracker"] = copy.deepcopy(parent_overrides["usage_tracker"])
token = thread_local_overrides.set(new_overrides)

2. Added error logging for worker failures:

except Exception as e:
    logger.error(f"Worker failed: {e}")
    if self.provide_traceback:
        traceback.print_exc()

**## Reproduction

import dspy
import mlflow

mlflow.dspy.autolog()  # Enables usage_tracker, triggering the bug

class ProcessItem(dspy.Module):
    def __init__(self):
        self.predictor = dspy.ChainOfThought("item -> result")

    def forward(self, item):
        return self.predictor(item=item)

class GetItems(dspy.Module):
    def __init__(self):
        self.predictor = dspy.ChainOfThought("objective -> task_items: list[str]")

    def forward(self, objective):
        return self.predictor(objective=objective)

class Pipeline(dspy.Module):
    def __init__(self):
        self.get_items = GetItems()
        self.process_item = ProcessItem()

    def forward(self, objective):
        result = self.get_items(objective=objective)

        parallel = dspy.Parallel(num_threads=4, provide_traceback=True)
        results = parallel([
            (self.process_item, {"item": item})
            for item in result.task_items
        ])
        return dspy.Prediction(results=results)  # Returns [None, None, None, ...] before fix

lm = dspy.LM("azure/gpt-4")
dspy.configure(lm=lm)

pipeline = Pipeline()
results = pipeline("do something")
print(results)

[TomeHirata]

This is not really helpful since we catch exceptions by wrapping the original function in _wrap_function

[LukeDevs]

@TomeHirata Understood. I've reverted back to pass.

[TomeHirata]

Looks reasonable, can we keep the comment? And also can you add a test?

[LukeDevs]

@TomeHirata Added the comment back in directly above the deep copy.

[LukeDevs]

@TomeHirata. Implemented a unit test as requested. Tried to use the existing tests as reference.