核心开关是 is_replaying，在 loop 初始化时按"有没有指定 checkpoint_id"决定（_loop.py:315）：

# _loop.py:173
    is_replaying: bool
# _loop.py:315
        self.is_replaying = CONFIG_KEY_CHECKPOINT_ID in config[CONF]

它只在第一个 tick 为真，之后立刻关掉（_loop.py:715-716，在 after_tick 里）：

# _loop.py:715
        # only replay (re-execute) done tasks on the first tick
        self.is_replaying = False

tick 开始时，如果不是重放且有遗留写，就把它们复原到任务上（_loop.py:661-664）：

# _loop.py:661
        # if there are pending writes from a previous loop, apply them
        if not self.is_replaying and self.checkpoint_pending_writes:
            self._reapply_writes_to_succeeded_nodes(self.tasks)
            self._resume_error_handlers_if_applicable()

复原逻辑 _loop.py:736-749：

# _loop.py:736
    def _reapply_writes_to_succeeded_nodes(self, tasks) -> None:
        """Restore successful channel writes from checkpoint to in-memory tasks.
        Skips control signals (ERROR, ERROR_SOURCE_NODE, INTERRUPT, RESUME)
        so that failed/interrupted tasks remain with empty writes and will be
        re-executed (or routed to error handlers) by the runner."""
        for tid, k, v in self.checkpoint_pending_writes:
            if k in (ERROR, ERROR_SOURCE_NODE, INTERRUPT, RESUME):
                continue                                    # ① 控制信号跳过
            if task := tasks.get(tid):
                task.writes.append((k, v))                  # ② 普通写复原到任务

上一讲那句 if k in (ERROR, ERROR_SOURCE_NODE, INTERRUPT, RESUME): continue 是幂等的灵魂。函数注释（_loop.py:739-743）解释了动机：

幂等的另一半在写入端。put_writes（_loop.py:415-437）存写时会清掉同一任务的旧写再存新的：

# _loop.py:415
    def put_writes(self, task_id: str, writes: WritesT) -> None:
        if not writes:
            return
        # deduplicate writes to special channels, last write wins
        if all(w[0] in WRITES_IDX_MAP for w in writes):
            writes = list({w[0]: w for w in writes}.values())   # ① 特殊通道去重
        if task_id == NULL_TASK_ID:
            ...  # null 任务的写是累积的（如多个 RESUME）
        else:
            # remove existing writes for this task
            self.checkpoint_pending_writes = [
                w for w in self.checkpoint_pending_writes if w[0] != task_id  # ② 先删旧
            ]
            writes_to_save = writes                                          # ③ 再存新

D44 已见过 ReplayState（_replay.py:14），这里从"幂等"角度再理解它。子图重放的难点：同一个逻辑子图在循环里会跑多次，重放时该给它哪一份存档？

# _replay.py:34
    def _is_first_visit(self, checkpoint_ns: str) -> bool:
        # "sub_node:task_id" -> "sub_node"  剥掉 task_id 后缀
        stable_ns = (checkpoint_ns.rsplit(NS_END, 1)[0]
                     if NS_END in checkpoint_ns else checkpoint_ns)
        if stable_ns in self._visited_ns:
            return False                          # 已访问过 → 不是第一次
        self._visited_ns.add(stable_ns)
        return True                               # 第一次 → 记下