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)

# 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 (gh-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. ```python """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)