昨天 _call_with_options 从 config 里取回调 pregel/_call.py:288：

# pregel/_call.py:288
config = get_config()
impl = config[CONF][CONFIG_KEY_CALL]         # ← 这个 impl 是谁放的？
fut = impl(func, (args, kwargs), retry_policy=retry_policy,
           cache_policy=cache_policy, callbacks=config["callbacks"], timeout=timeout)
return fut

答案在执行器 runner 启动每个任务时。它把 _call 用 partial 绑好一堆上下文，塞进 CONFIG_KEY_CALL pregel/_runner.py:211：

# pregel/_runner.py:211
run_with_retry(
    t,
    retry_policy,
    configurable={
        CONFIG_KEY_CALL: partial(
            _call,                         # ← 就是它！
            weakref.ref(t),                # 当前正在跑的任务（弱引用）
            retry_policy=retry_policy,
            futures=weakref.ref(futures),  # 收集子任务 Future 的容器
            schedule_task=schedule_task,   # "向引擎追加任务"的钩子
            submit=self.submit,            # 线程池提交器
        ),
    },
)

_call 是接缝的核心 pregel/_runner.py:700。它做三件事：造 Call、请引擎 schedule_task、拿到（或提交）任务的 Future：

# pregel/_runner.py:720
scratchpad = task().config[CONF][CONFIG_KEY_SCRATCHPAD]
if next_task := schedule_task(              # ① 请引擎追加一个任务
    task(),
    scratchpad.call_counter(),              # 这是本任务内第几次 task 调用（计数器）
    Call(                                   # ② 把这次调用打包成 Call
        func, input,
        retry_policy=retry_policy, cache_policy=cache_policy,
        callbacks=callbacks, timeout=timeout,
    ),
):
    ...
    elif next_task.writes:                  # ③a 若任务已经跑过（有写）→ 直接取缓存结果
        fut = concurrent.futures.Future()
        ret = next((v for c, v in next_task.writes if c == RETURN), MISSING)
        if ret is not MISSING:
            fut.set_result(ret)
        ...
    else:                                   # ③b 否则真正提交去线程池跑
        fut = submit()(run_with_retry, next_task, retry_policy,
                       configurable={CONFIG_KEY_CALL: partial(_call, ...)},
                       __next_tick__=True)
        futures()[fut] = next_task

Call 是个极简的数据容器 pregel/_algo.py:120，装着"跑这个 task 需要的一切"：

# pregel/_algo.py:120
class Call:
    __slots__ = ("func", "input", "retry_policy", "cache_policy",
                 "callbacks", "timeout")
    func: Callable                                   # 你的原函数
    input: tuple[tuple[Any, ...], dict[str, Any]]    # (args, kwargs)
    retry_policy: Sequence[RetryPolicy] | None       # 来自 @task 的重试策略
    cache_policy: CachePolicy | None                 # 来自 @task 的缓存策略
    callbacks: Callbacks
    timeout: TimeoutPolicy | None

func 本身不能直接当节点跑——引擎节点吃的是 Runnable。所以 get_runnable_for_task 把它裹一层 pregel/_call.py:200：

# pregel/_call.py:200
def get_runnable_for_task(func) -> Runnable:
    ...
    run = RunnableCallable(func, ..., explode_args=True, trace=False, recurse=False)
    seq = RunnableSeq(
        run,                                          # ① 跑你的函数
        ChannelWrite([ChannelWriteEntry(RETURN)]),    # ② 把返回值写进 RETURN 通道
        name=name,
        trace_inputs=functools.partial(_explode_args_trace_inputs, inspect.signature(func)),
    )
    ...
    return CACHE.setdefault(key, seq)                 # 缓存，同一 func 只包一次

L03 看到 task 跑完会把返回值写进 RETURN 通道。L02 看到 _call 从 next_task.writes 里挑出 c == RETURN 的写，填进 Future pregel/_runner.py:748：

# pregel/_runner.py:748
ret = next((v for c, v in next_task.writes if c == RETURN), MISSING)
if ret is not MISSING:
    fut.set_result(ret)                       # ← 结果回到你手里的 Future
elif exc := next((v for c, v in next_task.writes if c == ERROR), None):
    fut.set_exception(exc if isinstance(exc, BaseException) else Exception(exc))
else:
    fut.set_result(None)

schedule_task 最终会调到 prepare_push_task_functional pregel/_algo.py:801，把 Call 变成一个真正的可执行任务。核心是命名和坐标：

# pregel/_algo.py:826
call = task_path[-1]                              # 取出附在路径末尾的 Call
proc_ = get_runnable_for_task(call.func)          # L03 那个包装
name = proc_.name
...
checkpoint_ns = f"{parent_ns}{NS_SEP}{name}" if parent_ns else name   # 命名空间嵌套
task_id = task_id_func(checkpoint_id_bytes, checkpoint_ns, str(step),
                       name, PUSH, task_path_str(task_path[1]), str(task_path[2]))
task_checkpoint_ns = f"{checkpoint_ns}:{task_id}"

然后组装成 PregelExecutableTask pregel/_algo.py:893：

# pregel/_algo.py:893
return PregelExecutableTask(
    name,
    call.input,                                   # (args, kwargs)
    proc_,                                         # 要跑的 Runnable
    writes,
    patch_config(..., configurable={
        CONFIG_KEY_TASK_ID: task_id,
        CONFIG_KEY_SEND: writes.extend,           # task 写结果的出口
        CONFIG_KEY_CHECKPOINT_NS: task_checkpoint_ns,  # 独立命名空间
        CONFIG_KEY_RUNTIME: runtime,
        ...
    }),
    triggers,                                     # PUSH_TRIGGER
    call.retry_policy or retry_policy,            # @task 的重试策略在此生效（D52）
    cache_key,                                    # @task 的缓存 key 在此生成（D51）
    task_id,
    in_progress_task_path,
    timeout=call.timeout,
)

回看 L02 那个分支——这正是 D47 docstring 说的"恢复后 compose_essay 不会重跑" pregel/_runner.py:745：

# pregel/_runner.py:745
elif next_task.writes:
    # if it already ran, return the result
    fut = concurrent.futures.Future()
    ret = next((v for c, v in next_task.writes if c == RETURN), MISSING)
    if ret is not MISSING:
        fut.set_result(ret)          # 直接返回上次的结果，不重新执行 func

✋ 10 分钟动手

# 1. 顺着接缝读三段
sed -n '700,786p'  libs/langgraph/langgraph/pregel/_runner.py   # _call
sed -n '200,242p'  libs/langgraph/langgraph/pregel/_call.py     # get_runnable_for_task
sed -n '801,933p'  libs/langgraph/langgraph/pregel/_algo.py     # prepare_push_task_functional

# 2. 观察 task 不重跑：中断再恢复
python - <<'PY'
import time
from langgraph.func import entrypoint, task
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
calls = {"n": 0}
@task
def compose(topic: str) -> str:
    calls["n"] += 1
    return f"essay about {topic}"
@entrypoint(checkpointer=InMemorySaver())
def wf(topic: str):
    essay = compose(topic).result()
    review = interrupt({"essay": essay})
    return {"essay": essay, "review": review}
cfg = {"configurable": {"thread_id": "t"}}
list(wf.stream("cats", cfg))                 # 触发 interrupt，compose 跑 1 次
list(wf.stream(Command(resume="good"), cfg)) # 恢复
print("compose 执行次数:", calls["n"])       # 仍是 1，没重跑
PY