PregelLoop（_loop.py:158）是一次 invoke/stream 的"运行时上下文"：它持有当前 checkpoint、channels、待执行的 tasks、当前步号 step、上限 stop、状态 status 等。它有同步、异步两个子类（_loop.py:1469 SyncPregelLoop / _loop.py:1722 AsyncPregelLoop），共享同一份 tick/after_tick 逻辑。

先看"心跳"本身。在 Pregel.stream() 里（invoke 内部就是收集 stream 的最后一份），主循环长这样（main.py:2964-2988）：

# main.py:2960 —— 注释直接说明了 BSP：N 步的写入只在 N+1 步可见
# computation proceeds in steps, while there are channel updates.
# Channel updates from step N are only visible in step N+1
while loop.tick():                                  # 2964 备好本步任务；无任务→False→退出
    for task in loop.match_cached_writes():         # 2965 命中缓存的任务直接产出
        loop.output_writes(task.id, task.writes, cached=True)
    for _ in runner.tick(                           # 2967 执行器：并行跑本步任务（Day 22）
        [t for t in loop.tasks.values() if not t.writes],
        timeout=self.step_timeout,                  # 2969 单步墙钟超时
        get_waiter=get_waiter,
        schedule_task=loop.accept_push,             # 2971 支持运行中动态追加 Send 任务
    ):
        yield from _output(stream_mode, ...)        # 2974 每有任务完成就流式产出
    loop.after_tick()                               # 2984 合并写入 + 存档
    if durability_ == "sync":                       # 2987 sync 模式：等 checkpoint 落盘
        loop._put_checkpoint_fut.result()

进入 tick() 本体（_loop.py:599），前半段：

def tick(self) -> bool:                     # _loop.py:599
    """Execute a single iteration of the Pregel loop.
    Returns: True if more iterations are needed."""
    if self.step > self.stop:               # 607 超过步数上限
        self.status = "out_of_steps"
        return False

    self.tasks = prepare_next_tasks(        # 612 Plan：算出本步要跑哪些任务（Day 21）
        self.checkpoint, self.checkpoint_pending_writes,
        self.nodes, self.channels, self.managed, self.config,
        self.step, self.stop, for_execution=True,
        trigger_to_nodes=self.trigger_to_nodes,
        updated_channels=self.updated_channels,   # 626 上一步更新了哪些 channel
        retry_policy=self.retry_policy, cache_policy=self.cache_policy,
    )
    ...
    if not self.tasks:                      # 653 没有任务了 = 图跑完了
        self.status = "done"
        return False

tick() 后半段——在真正放行任务之前，还要处理"回放已有写入"和"静态断点"（_loop.py:661-681）：

    # 有上次遗留的 pending writes（比如从 checkpoint 恢复），先补上
    if not self.is_replaying and self.checkpoint_pending_writes:   # 662
        self._reapply_writes_to_succeeded_nodes(self.tasks)
        self._resume_error_handlers_if_applicable()

    # 执行前静态断点：interrupt_before 命中就抛 GraphInterrupt 暂停
    if self.interrupt_before and should_interrupt(                 # 667
        self.checkpoint, self.interrupt_before, self.tasks.values()
    ):
        self.status = "interrupt_before"
        raise GraphInterrupt()

    self._emit("tasks", map_debug_tasks, self.tasks.values())      # 674 debug 流产出（Day 26）

    for task in self.tasks.values():        # 677 已经有缓存写入的任务，直接产出
        if task.writes:
            self.output_writes(task.id, task.writes, cached=True)
    return True                             # 681 一切就绪，放行给执行器

执行器把本步任务都跑完后，主循环调 after_tick()（_loop.py:683）——这就是 Day 19 的 Update 阶段：

def after_tick(self) -> None:               # _loop.py:683
    writes = [w for t in self.tasks.values() for w in t.writes]   # 685 收齐本步所有写入
    ...
    self.updated_channels = apply_writes(   # 692 统一合并进 channel（Day 21）
        self.checkpoint, self.channels, self.tasks.values(),
        self.checkpointer_get_next_version, self.trigger_to_nodes,
    )
    # 如果 output 通道被更新了，产出一份 values 流
    if not self.updated_channels.isdisjoint(...):                 # 700
        self._emit("values", map_output_values, self.output_keys, writes, self.channels)
    self.checkpoint_pending_writes.clear()  # 714 清空本步 pending
    self.is_replaying = False               # 716 只有第一步会 replay
    self._put_checkpoint({"source": "loop"})# 718 存档：这一步的 checkpoint
    # 执行后静态断点
    if self.interrupt_after and should_interrupt(                 # 720
        self.checkpoint, self.interrupt_after, self.tasks.values()
    ):
        self.status = "interrupt_after"
        raise GraphInterrupt()
    self.config[CONF].pop(CONFIG_KEY_RESUMING, None)             # 726

tick 第一行 if self.step > self.stop 里的两个数是怎么来的？初始化时都是 0（_loop.py:301-302）：

# _loop.py:301
self.step = 0
self.stop = 0

而真正决定上限，是在从 checkpoint 建立起点时（SyncPregelLoop，_loop.py:1700-1701）：

# _loop.py:1700
self.step = self.checkpoint_metadata["step"] + 1          # 从上次存档的下一步开始
self.stop = self.step + self.config["recursion_limit"] + 1 # 上限 = 起点 + 递归上限 + 1

第一次 tick 之前，引擎要先把你的输入"喂进图"，这活由 _first()（_loop.py:848）干。它最重要的判断是"这是全新运行，还是从中断恢复？"（_loop.py:861-872）：

# _loop.py:861
input_is_command = isinstance(self.input, Command)
is_resuming = bool(self.checkpoint["channel_versions"]) and bool(
    configurable.get(
        CONFIG_KEY_RESUMING,
        self.input is None                  # invoke(None, config) = 恢复
        or input_is_command                 # 传 Command 也是对已有状态操作
        or (not self.is_nested and 同一个 run_id)  # stream 重连
    )
)

✋ 10 分钟动手

# 1. 读 tick 主体（判停 + 计划 + 断点）
sed -n '599,681p' libs/langgraph/langgraph/pregel/_loop.py

# 2. 读 after_tick（合并写入 + 存档）
sed -n '683,726p' libs/langgraph/langgraph/pregel/_loop.py

# 3. 看 step/stop 上限怎么定
sed -n '1700,1701p' libs/langgraph/langgraph/pregel/_loop.py

# 4. 亲手触发一次 GraphRecursionError（体会保险丝）
python -c "
from langgraph.graph import StateGraph, START
g = StateGraph(dict); g.add_node('loop', lambda s: {'n': s.get('n',0)+1})
g.add_edge(START, 'loop'); g.add_edge('loop', 'loop')   # 故意死循环
try: g.compile().invoke({'n':0}, {'recursion_limit':5})
except Exception as e: print(type(e).__name__, e)
"