Concurrent task execution fails with ContextVar errors when using asyncio.gather()

[zzstoatzz]

Problem

Users cannot run multiple independent tasks concurrently using asyncio.gather() due to ContextVar token errors. The current implementation uses ContextVars for tracking actors, which don't work across asyncio context boundaries.

Current Behavior

import asyncio
import marvin
from marvin import Task

# This works but runs sequentially (slow)
task_1 = Task("Say 'one'", result_type=str)
task_2 = Task("Say 'two'", result_type=str) 
task_3 = Task("Say 'three'", result_type=str)

results = marvin.run_tasks([task_1, task_2, task_3])  # ✅ Works, 3.0s

# This should run concurrently but fails with ContextVar errors
await asyncio.gather(
    task_1.run_async(),
    task_2.run_async(), 
    task_3.run_async()
)  # ❌ Fails with context errors

Error Details

ValueError: <Token var=<ContextVar name='current_actor' default=None at 0x10428d3a0> at 0x1107b2040> was created in a different Context

Full stack trace:

  File "/Users/nate/github.com/prefecthq/marvin/src/marvin/engine/orchestrator.py", line 192, in run_once
    with actor:
         ^^^^^
  File "/Users/nate/github.com/prefecthq/marvin/src/marvin/agents/actor.py", line 79, in __exit__
    _current_actor.reset(self._tokens.pop())
ValueError: <Token var=<ContextVar name='current_actor' default=None at 0x10428d3a0> at 0x1107b2040> was created in a different Context

Root Cause

The issue stems from Marvin's use of ContextVars in src/marvin/agents/actor.py:

_current_actor: ContextVar["Actor | None"] = ContextVar(
    "current_actor",
    default=None,
)

class Actor:
    def __enter__(self):
        token = _current_actor.set(self)
        self._tokens.append(token)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        _current_actor.reset(self._tokens.pop())  # ❌ Fails here

When asyncio.gather() runs multiple coroutines concurrently, each runs in its own async context. The ContextVar token created in __enter__ exists in one context, but __exit__ tries to reset it from a different context, causing the error.

Performance Impact

Current timings:

• run_tasks([task1, task2, task3]): 3.0 seconds (sequential)
• asyncio.gather(...): FAILS (would be ~1.0 second if working)

Users lose ~66% performance by being forced to use sequential execution.

Attempted Solutions

1. Remove ContextVars entirely

• Problem: Breaks existing functionality that depends on get_current_actor()
• Used by: CLI tools (src/marvin/tools/interactive/cli.py), tests, public API

2. Try/except around ContextVar operations

• Problem: Masks real errors, doesn't solve the fundamental issue

3. Use contextvars.copy_context()

• Problem: Still doesn't solve cross-context token reset

Potential Solutions

Option 1: Make ContextVar usage optional

def __exit__(self, exc_type, exc_val, exc_tb):
    if self._tokens:
        try:
            _current_actor.reset(self._tokens.pop())
        except ValueError:
            # Context was created in different async context - ignore
            pass

Option 2: Pass actor explicitly instead of global state

• Refactor tools to receive actor as parameter
• Remove ContextVar dependency entirely
• Breaking change but cleaner architecture

Option 3: Use thread-local storage

• Replace ContextVar with threading.local()
• Problem: Doesn't work well with async code

Option 4: Detect concurrent execution and skip context tracking

def __enter__(self):
    try:
        token = _current_actor.set(self)
        self._tokens.append(token)
    except RuntimeError:
        # Running in concurrent context - skip tracking
        pass
    return self

Expected Behavior

Users should be able to run independent tasks concurrently:

# Should work and complete in ~1.0 second
await asyncio.gather(
    task_1.run_async(),
    task_2.run_async(), 
    task_3.run_async()
)

Workaround

Currently, users must use sequential execution:

# Works but slow (3x longer)
results = marvin.run_tasks([task_1, task_2, task_3])

Related Issues

This issue was originally reported in Discord by Matthew Dangerfield, where multiple independent tasks caused "Multiple EndTurn tools detected" warnings and infinite loops. The sequential fix (PR #1229) solved the warnings but didn't enable concurrent execution.

Environment

• Python: 3.12.8
• Marvin: main branch
• OS: macOS (but affects all platforms)

Reproduction

import asyncio
import time
import marvin
from marvin import Task

async def reproduce_issue():
    task_1 = Task("Say 'one'", result_type=str)
    task_2 = Task("Say 'two'", result_type=str) 
    task_3 = Task("Say 'three'", result_type=str)
    
    # This will fail with ContextVar errors
    await asyncio.gather(
        task_1.run_async(),
        task_2.run_async(), 
        task_3.run_async()
    )

asyncio.run(reproduce_issue())