执行器夹在 Day 20 的 tick 和 after_tick 中间，职责被官方注释说得很清楚（_runner.py:136-138）：

• 并发执行一组 Pregel 任务；
• 把它们的写入 commit 回去（攒到 checkpoint 的 pending_writes）；
• 有输出可吐时把控制权 yield 给外层（流式）；
• 必要时 中断/取消其他任务（一个失败，全部叫停）。

看它的 __init__（_runner.py:140-169）。最关键是两个参数都用 weakref.ref 包着：

# _runner.py:135
class PregelRunner:
    """Responsible for executing a set of Pregel tasks concurrently, committing
    their writes, yielding control to caller when there is output to emit, and
    interrupting other tasks if appropriate."""
    def __init__(self, *,
        submit: weakref.ref[Submit],                                   # 143 拿到线程池的提交入口
        put_writes: weakref.ref[Callable[[str, Sequence[...]], None]], # 144 把写入交回 loop
        use_astream: bool = False,
        node_finished: Callable[[str], None] | None = None,
        node_error_handler_map: Mapping[str, str] | None = None,       # 147 节点→错误处理节点
        ...
    ) -> None:
        self.submit = submit
        self.put_writes = put_writes
        self._handled_exception_ids: set[int] = set()                  # 169 已被错误处理器接管的异常

tick()（_runner.py:176）是执行器的主入口。开头先建一个特殊的 FuturesDict，然后 yield 一次：

# _runner.py:176
def tick(self, tasks, *, reraise=True, timeout=None, retry_policy=None,
         get_waiter=None, schedule_task) -> Iterator[None]:
    tasks = tuple(tasks)
    futures = FuturesDict(                          # 190 {future: task} 的特殊字典
        callback=weakref.WeakMethod(self.commit),   # 191 每个 future 完成自动回调 commit
        event=threading.Event(),
        should_stop=partial(_should_stop_others, ...),  # 193 判断"要不要停其他任务"
        future_type=concurrent.futures.Future,
    )
    yield                                           # 199 先把控制权还给调用方
    if len(tasks) == 0:
        return                                      # 201 没任务，直接结束

如果这一步只有一个任务、没超时、没 waiter，开线程池纯属浪费。tick 有一条快路径（_runner.py:203-221）：

# _runner.py:203
elif len(tasks) == 1 and timeout is None and get_waiter is None:
    t = tasks[0]
    try:
        run_with_retry(                             # 207 直接在当前线程跑（含重试，Day 25）
            t, retry_policy,
            configurable={CONFIG_KEY_CALL: partial(_call, weakref.ref(t), ...)},  # 210
        )
        self.commit(t, None)                        # 221 成功：commit（无异常）
    except Exception as exc:
        self.commit(t, exc)                         # 223 失败：也 commit（带异常，存错误）
        ...错误处理器路由 / reraise...

多任务时走通用路径：全部提交线程池，然后循环"等最先完成的一个、处理它、再等下一个"（_runner.py:259-289）：

# _runner.py:259
for t in tasks:
    fut = self.submit()(                            # 260 提交到线程池，拿到 future
        run_with_retry, t, retry_policy,
        configurable={CONFIG_KEY_CALL: partial(_call, weakref.ref(t), ...)},
        __reraise_on_exit__=reraise,
    )
    futures[fut] = t                                # 276 记账：这个 future 对应哪个 task
# 每有一个任务完成就吐一次输出
end_time = timeout + time.monotonic() if timeout else None   # 280 墙钟截止时刻
while len(futures) > (1 if get_waiter is not None else 0):    # 282
    done, inflight = concurrent.futures.wait(
        futures, return_when=concurrent.futures.FIRST_COMPLETED,   # 285 等"第一个完成"
        timeout=(max(0, end_time - time.monotonic()) if end_time else None),
    )
    if not done:
        break                                       # 289 一个都没完成=超时了，跳出

每轮收割后判断要不要停其他任务（_runner.py:330-335 + _runner.py:616）：

# _runner.py:330
if _should_stop_others(done_for_stop, handled_exception_ids=self._handled_exception_ids):
    break                                           # 333 有任务真失败 → 停掉其余
yield                                               # 335 否则让出控制权，流式吐输出

# _runner.py:616
def _should_stop_others(done, *, handled_exception_ids=None) -> bool:
    """Check if any task failed, if so, cancel all other tasks.
    GraphInterrupts are not considered failures."""
    for fut in done:
        if fut.cancelled():
            continue
        elif exc := fut.exception():                # 626 这个 future 抛异常了？
            if (id(exc) not in (handled_exception_ids or set())
                and not isinstance(exc, GraphBubbleUp)):   # 629 中断/续跑信号不算失败
                return True
    return False

而每个任务完成后（无论成败）由 commit 归档写入（_runner.py:574）：

# _runner.py:574
def commit(self, task, exception) -> None:
    if isinstance(exception, asyncio.CancelledError):
        task.writes.append((ERROR, exception)); self.put_writes()(task.id, task.writes)
    elif exception:
        if isinstance(exception, GraphInterrupt):   # 585 是中断 → 存 INTERRUPT 写入
            ... self.put_writes()(task.id, [(INTERRUPT, exception.args[0]), *resumes])
        elif isinstance(exception, GraphBubbleUp):  # 592 冒泡信号 → 交给上层处理
            pass
        else:                                       # 595 真错误 → 存 ERROR 写入
            task.writes.append((ERROR, exception)); self.put_writes()(task.id, task.writes)
    else:                                           # 604 成功
        if not task.writes:
            task.writes.append((NO_WRITES, None))   # 611 没写任何东西 → 补一个占位标记
        self.put_writes()(task.id, task.writes)     # 613 把写入交回 loop

上面的 self.submit() 从哪来？来自 BackgroundExecutor（_executor.py:40）——一个包着线程池的上下文管理器：

# _executor.py:54
def submit(self, fn, *args,
    __cancel_on_exit__=False,     # 59 退出时若还没启动，可取消
    __reraise_on_exit__=True,     # 60 退出时把任务里的异常重新抛出
    __next_tick__=False, **kwargs,
) -> concurrent.futures.Future[T]:
    ctx = copy_context()                            # 64 拷贝上下文（ContextVar 隔离）
    task = self.executor.submit(ctx.run, fn, *args, **kwargs)   # 71 丢进线程池
    self.tasks[task] = (__cancel_on_exit__, __reraise_on_exit__)
    task.add_done_callback(self.done)               # 74 完成即从 tasks 里摘除
    return task

# _executor.py:93 —— 退出上下文时（一步跑完）
def __exit__(self, exc_type, exc_value, traceback):
    tasks = self.tasks.copy()
    for task, (cancel, _) in tasks.items():
        if cancel:
            task.cancel()                           # 104 取消标了 cancel_on_exit 的
    if pending := {t for t in tasks if not t.done()}:
        concurrent.futures.wait(pending)            # 107 等所有未完成的跑完
    self.stack.__exit__(exc_type, exc_value, traceback)   # 109 关线程池
    if exc_type is None:                            # 111 没有正在传播的异常时
        for task, (_, reraise) in tasks.items():    # 113 才把任务内的异常重新抛出

✋ 10 分钟动手

# 1. 读 PregelRunner.tick 主体（快路径 + 并行循环）
sed -n '176,290p' libs/langgraph/langgraph/pregel/_runner.py

# 2. 读 commit 三条分支
sed -n '574,613p' libs/langgraph/langgraph/pregel/_runner.py

# 3. 读线程池 BackgroundExecutor
sed -n '40,120p' libs/langgraph/langgraph/pregel/_executor.py

# 4. 亲眼看并行：两个节点同一步跑，观察它们的时间戳交错
python -c "
import time
from langgraph.graph import StateGraph, START, END
from typing import Annotated; import operator
def make(name):
    def f(s):
        print(name,'start',time.time()); time.sleep(0.5); print(name,'end',time.time())
        return {'log': [name]}
    return f
class S(dict): pass
g = StateGraph(dict)
g.add_node('a', make('A')); g.add_node('b', make('B'))
g.add_edge(START,'a'); g.add_edge(START,'b'); g.add_edge('a',END); g.add_edge('b',END)
g.compile().invoke({})   # A、B 同一超步并行，start 几乎同时
"