[anmolxlight]

Fixes #74575.

Problem

LocalExecutor gave all workers one SimpleQueue as their input. In CPython, SimpleQueue.get() holds a shared read lock while it blocks on the pipe. If the idle worker holding that lock is SIGKILLed (for example by the OOM killer), nothing releases the lock. Every other worker then waits on it forever, and the executor keeps accepting work it can never run. Tasks stay queued while the scheduler reports healthy.

Fix

Give each worker its own input queue and dispatch each workload to a specific worker:

• _dispatch picks an idle worker, spawns one under parallelism, or queues behind a busy one.
• A dead worker is reaped together with its own queue, so it cannot block the others.
• Dispatched-but-not-started keys are tracked separately (_dispatched), so a worker that dies before START still leaves the key for the scheduler's stuck-in-queued revoke, as before.
• Terminal results are honored even for keys no longer in running, so a worker is not left excluded.

Tests

• New regression test test_workload_runs_after_worker_holding_read_lock_is_killed: kills all idle workers, then requires a later task to succeed. On main it fails with the symptom from the issue (task never leaves the queue, 30s timeout). With this fix it passes (~3.6s).
• Rewrote test_local_executor_check_workers.py for the new reap contract.
• test_local_executor.py and test_local_executor_check_workers.py: 65 passed, 1 skipped.
• test_base_executor.py has 3 test_debug_dump* errors that also occur on unmodified main; unrelated.
• ruff check, ruff format --check, and mypy on local_executor.py are clean.

Notes

This does not change behavior when workers stay alive. The per-worker queue is also what makes a killed worker harmless to the rest of the pool.

[o-nikolas]

Thanks, a few things before this can go in:

1. This now conflicts with #70545 on main. _pick_worker already drains results before each put, so its extra _read_results() line can go, but its two tests use activity_queue/_unread_messages and need porting.
2. Please drop 74588.bugfix.rst. Core only takes significant newsfragments now (see contributing-docs/18_contribution_workflow.rst).
3. When can the _pick_worker fallback (next(iter(self.workers))) actually happen? If it's real, the workload waits behind a possibly long task, and if that worker dies it sits until task_queued_timeout. If it's only defensive, a warning log would help. Also please drop the ponytail: prefix.
4. In fork mode, replacement workers are now forked without gc.freeze; the old _check_workers refill used _spawn_workers_with_gc_freeze. Worth keeping?
5. Could you add unit tests for the _pick_worker branches (idle reuse, spawn under parallelism, fallback) and the new late-result branch in _read_results? test_executor_lazy_worker_spawning also lost its spawn-on-dispatch check.
   Not for this PR: result_queue is still shared, so a worker killed mid-put holds its write lock the same way. Small window, but maybe worth a follow-up issue.

---

Drafted-by: Kiro (claude-opus-5.5); reviewed by @o-nikolas before posting