asyncio: eager_start Task created from a thread not running the target loop silently corrupts that loop's current-task tracking (3.12, 3.13; guarded in 3.14)

[smitrob]

Bug report

Bug description

On 3.12 and 3.13, constructing asyncio.Task(coro, loop=loop, eager_start=True) (or calling asyncio.eager_task_factory(loop, coro)) from a thread that is not running loop executes the eager start in the calling thread: task_eager_start in Modules/_asynciomodule.c gates only on loop.is_running() — which is true from any thread — then runs swap_current_task(loop, task) → synchronous first step → swap back, all in the caller's thread, mutating the target loop's entry in the interpreter-global current_tasks dict.

If the eager first chunk releases the GIL (any syscall — e.g. the sock.send() an HTTP client's body writer performs), the loop's own thread runs while the foreign task sits in its current-task slot. Consequences observed:

• Every Task.task_wakeup callback that runs in the window fails with RuntimeError: Cannot enter into task <victim> while another task <the eager task> is being executed, reported through the loop exception handler.
• The victim task is then permanently hung: its wakeup callback raised and is never retried, and its awaited future is already done, so nothing ever reschedules it. This is silent — no exception ever reaches the awaiting code.
• Typically also RuntimeError: Leaving task ... does not match the current task, after which the loop's bookkeeping can stay poisoned and the whole loop wedges.

I'm aware Tasks are documented as not thread-safe, so on its own this could be read as API misuse. Two things make it worth a guard on 3.13 in my view:

1. Third-party libraries construct such tasks on the caller's behalf. aiohttp ≥3.10 starts its request-body writer with Task(write_bytes, loop=self.loop, eager_start=True) where self.loop is the session's loop ('Optimization for Python 3.12' in client_reqrep.py), and langgraph calls asyncio.eager_task_factory(loop, ...) directly. An application that drives a session from the wrong loop never touches Task() itself, and before eager start this misuse failed loudly (cross-loop future errors) rather than corrupting the other loop's state and deadlocking its tasks. We hit this in production: a worker wedged mid-run with no exception, no crash, nothing to alarm on.
2. 3.14 already fixes it, loudly and structurally. Since the current task moved to per-thread state (Store current task on the loop in asyncio  #128415), enter_task/leave_task/swap_current_task all check ts->asyncio_running_loop != loop and raise RuntimeError: loop <...> is not the running loop in the calling thread; the target loop is unharmed (verified with the repro below on 3.14.7).

Reproducer

Self-contained, stdlib only. On 3.12.13 and 3.13.14 it prints a stream of Cannot enter into task ... errors within seconds, then the beat counter stops advancing (main-loop tasks permanently hung); on 3.14.7 the foreign thread raises loop ... is not the running loop and the main loop is unaffected. Removing the time.sleep(0.0005) (the GIL release) from the eager task's first chunk makes the corruption unobservable, confirming the window mechanism.

"""Repro: Task(coro, loop=..., eager_start=True) from a thread not running
`loop` corrupts that loop's current-task bookkeeping.

Run on CPython 3.12+ (reproduces within seconds on 3.12.13). Expected output:
repeated "RuntimeError: Cannot enter into task <victim> while another task
<eager_victim()> is being executed" raised from Task.task_wakeup callbacks on
the main loop, usually one "Leaving task ... does not match the current task",
and then `beat` stops advancing: the wakeup callbacks that raised are never
retried, so those tasks hang forever and the loop is permanently wedged.

Removing the GIL-releasing call (time.sleep) from eager_victim's first chunk
makes the corruption window unobservable and the errors vanish.
"""

import asyncio
import threading
import time

errors: list[str] = []
stop = threading.Event()
beat = {"n": 0}


async def eager_victim():
    # First chunk performs a GIL-releasing call (stand-in for the sock.send()
    # an HTTP client's body writer does), then suspends.
    time.sleep(0.0005)
    await asyncio.sleep(0.001)


def foreign_thread(loop):
    while not stop.is_set():
        # The problematic call: eager_start only checks loop.is_running(),
        # so the swap/step/swap of `loop`'s current task runs on THIS thread.
        asyncio.Task(eager_victim(), loop=loop, eager_start=True)
        time.sleep(0)


async def wakeup_heavy():
    loop = asyncio.get_running_loop()
    while not stop.is_set():
        fut = loop.create_future()
        loop.call_soon(fut.set_result, None)
        await fut
        beat["n"] += 1


async def cpu_burner():
    import json

    blob = {"k%d" % i: "x" * 100 for i in range(500)}
    while not stop.is_set():
        for _ in range(50):
            json.loads(json.dumps(blob))
        await asyncio.sleep(0)


def watchdog():
    last = -1
    while not stop.is_set():
        time.sleep(5)
        print(f"beat={beat['n']} advanced={beat['n'] != last} errors={len(errors)}", flush=True)
        last = beat["n"]


async def main():
    loop = asyncio.get_running_loop()

    def handler(loop, ctx):
        msg = str(ctx.get("exception"))
        if "Cannot enter" in msg or "Leaving task" in msg:
            errors.append(msg)
            print("HIT:", msg[:200], flush=True)

    loop.set_exception_handler(handler)
    for _ in range(2):
        threading.Thread(target=foreign_thread, args=(loop,), daemon=True).start()
    threading.Thread(target=watchdog, daemon=True).start()
    tasks = [asyncio.ensure_future(wakeup_heavy()) for _ in range(8)]
    tasks += [asyncio.ensure_future(cpu_burner()) for _ in range(2)]
    deadline = time.monotonic() + 30
    while time.monotonic() < deadline and len(errors) < 5:
        await asyncio.sleep(0.2)
    stop.set()
    print(f"done: errors={len(errors)}")


asyncio.run(main())

Sample 3.12.13 output:

HIT: Cannot enter into task <Task pending name='Task-4' coro=<wakeup_heavy() ...>> while another task <Task pending name='Task-3' coro=<eager_victim() ...>> is being executed
HIT: Leaving task <Task pending name='Task-1' coro=<main() ...>> does not match the current task ...
beat=0 advanced=False errors=12   # forever — loop wedged

Suggested fix

Backport the thread-affinity guard to 3.13 (still in bugfix): the minimal form is having task_eager_start (or swap_current_task) verify the calling thread's running loop is task->task_loop — falling back to call_soon scheduling, or raising as 3.14 does. I realize 3.12 is security-only; noting it here for completeness since it is affected.

Your environment

• CPython versions tested: 3.12.13 (affected), 3.13.14 (affected), 3.14.7 (guarded)
• Operating system: Linux (Docker python:3.12-slim/3.13-slim/3.14-slim, aarch64); originally observed on ECS Fargate
• Extension modules involved: none — pure stdlib repro (originally surfaced via aiohttp 3.14.3)