整个 _node.py 只有 95 行，前 80 行几乎全是"节点长相"的类型声明。看开头几个 Protocol：

class _Node(Protocol[NodeInputT_contra]):
    def __call__(self, state: NodeInputT_contra) -> Any: ...

class _NodeWithConfig(Protocol[NodeInputT_contra]):
    def __call__(self, state: NodeInputT_contra, config: RunnableConfig) -> Any: ...

class _NodeWithWriter(Protocol[NodeInputT_contra]):
    def __call__(self, state: NodeInputT_contra, *, writer: StreamWriter) -> Any: ...

class _NodeWithStore(Protocol[NodeInputT_contra]):
    def __call__(self, state: NodeInputT_contra, *, store: BaseStore) -> Any: ...

后面还有 config+writer、config+store、config+writer+store、以及带 runtime 的组合，最后用一个 TypeAlias 把它们全"或"起来：

StateNode: TypeAlias = (
    _Node[NodeInputT]
    | _NodeWithConfig[NodeInputT]
    | _NodeWithWriter[NodeInputT]
    | _NodeWithStore[NodeInputT]
    | _NodeWithWriterStore[NodeInputT]
    | _NodeWithConfigWriter[NodeInputT]
    | _NodeWithConfigStore[NodeInputT]
    | _NodeWithConfigWriterStore[NodeInputT]
    | _NodeWithRuntime[NodeInputT, ContextT]
    | Runnable[NodeInputT, Any]
)

_node.py 真正装数据的只有末尾这个 dataclass——昨天 add_node 打包出来的就是它：

@dataclass(slots=True)
class StateNodeSpec(Generic[NodeInputT, ContextT]):
    runnable: StateNode[NodeInputT, ContextT]
    metadata: dict[str, Any] | None
    input_schema: type[NodeInputT]
    retry_policy: RetryPolicy | Sequence[RetryPolicy] | None
    cache_policy: CachePolicy | None
    is_error_handler: bool = False
    error_handler_node: str | None = None
    ends: tuple[str, ...] | dict[str, str] | None = EMPTY_SEQ
    defer: bool = False
    timeout: TimeoutPolicy | None = None

引擎到底认哪些"特殊参数名"？答案是一张硬编码表 KWARGS_CONFIG_KEYS，每一项 = (参数名, 允许的类型, 从哪取值, 默认值)：

KWARGS_CONFIG_KEYS = (
    ("config",  (RunnableConfig, "RunnableConfig", Optional[RunnableConfig], ...),
                "N/A", inspect.Parameter.empty),
    ("writer",  (StreamWriter, "StreamWriter", inspect.Parameter.empty),
                "stream_writer", lambda _: None),
    ("store",   (BaseStore, "BaseStore", inspect.Parameter.empty),
                "store", inspect.Parameter.empty),
    ("store",   (Optional[BaseStore], "Optional[BaseStore]"),
                "store", None),                  # ← store 可选版，默认 None
    ("previous",(ANY_TYPE,), "previous", inspect.Parameter.empty),
    ("runtime", (ANY_TYPE,), "N/A", inspect.Parameter.empty),
    ("error",   (NodeError, "NodeError"), "N/A", None),
)

包装函数的 RunnableCallable 在初始化时就读一遍你的签名，决定"这个函数接受哪些特殊参数"：

        self.func_accepts: dict[str, tuple[str, Any]] = {}
        params = inspect.signature(func or afunc).parameters      # 读你的参数表

        for kw, typ, runtime_key, default in KWARGS_CONFIG_KEYS:
            p = params.get(kw)
            if p is None or p.kind not in VALID_KINDS:             # 没这个参数 → 跳过
                continue
            if typ != (ANY_TYPE,) and p.annotation not in typ:     # 类型不匹配 → 跳过
                if kw == "config" and p.annotation != inspect.Parameter.empty:
                    warnings.warn("The 'config' parameter should be typed as ...")
                continue
            self.func_accepts[kw] = (runtime_key, default)          # ✓ 记下"要注入这个"

运行时（每次调用节点）再按这张 func_accepts 把值填进 kwargs：

        runtime = config.get(CONF, {}).get(CONFIG_KEY_RUNTIME)
        for kw, (runtime_key, default) in self.func_accepts.items():
            if kw in kwargs:                     # 调用方已给 → 不覆盖
                continue
            kw_value = MISSING
            if kw == "config":       kw_value = config
            elif kw == "error":      kw_value = config...get(CONFIG_KEY_NODE_ERROR, MISSING)
            elif runtime:
                if kw == "runtime":  kw_value = runtime
                else:                kw_value = getattr(runtime, runtime_key)   # 从 Runtime 取
            if kw_value is MISSING:
                if default is inspect.Parameter.empty:
                    raise ValueError(f"Missing required config key '{runtime_key}' ...")
                kw_value = default
            kwargs[kw] = kw_value

回顾 D03：普通边进 edges，汇合边进 waiting_edges。取用时有个属性把两者拍平成统一的 (起, 终)：

    @property
    def _all_edges(self) -> set[tuple[str, str]]:
        return self.edges | {
            (start, end) for starts, end in self.waiting_edges for start in starts
        }

编译时 attach_edge 把"边"翻译成"往通道写"。看普通边（单起点）：

    def attach_edge(self, starts, end):
        if isinstance(starts, str):
            if end != END:
                self.nodes[starts].writers.append(
                    ChannelWrite(
                        (ChannelWriteEntry(_CHANNEL_BRANCH_TO.format(end), None),)
                    )
                )

再看节点自己被 attach 时（state.py:1518-1521），它的触发器 triggers 正是"通往我的那个分支通道"：

            self.nodes[key] = PregelNode(
                triggers=[branch_channel],       # 被 branch:to:key 通道触发就跑
                channels=("__root__" if is_single_input else input_channels),
                ...
                bound=node.runnable,             # 你的函数在这
            )

✋ 10 分钟动手

# 1. 通读 _node.py：9 种节点协议 + StateNodeSpec
sed -n '16,96p' libs/langgraph/langgraph/graph/_node.py

# 2. 看可注入参数清单
sed -n '147,208p' libs/langgraph/langgraph/_internal/_runnable.py

# 3. 看签名反射如何决定 func_accepts
sed -n '317,343p' libs/langgraph/langgraph/_internal/_runnable.py

# 4. 亲手验证"按需注入"：给函数加 config 参数看是否被塞
python3 -c "
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class S(TypedDict): x:int
def n(state, config): return {'x': state['x']+1}   # 要 config
g=StateGraph(S); g.add_node('A', n); g.add_edge(START,'A'); g.add_edge('A',END)
print(g.compile().invoke({'x':1}))   # config 会被自动注入
"

# 5. 看边如何编译成通道写/订阅
sed -n '1537,1561p' libs/langgraph/langgraph/graph/state.py