`plaita.node`¶

节点定义与 NodeRegistry。包含 Node 基类、默认内置节点（Start / End / Switch / Assignment / Bool / SwitchLegacy / Loop / Map / Filter / Find / Reduce / InlineFlow / ReferenceFlow / Parallel / HTTP / EventNode）、默认注册表与插件发现。

CodeNode 不在默认注册表

0.5.0 起 CodeNode 需 register_code_node() 显式注册（默认 docker 沙箱）。见内置节点 - code 与迁移指南。

node ¶

plaita.node — Node definitions and the NodeRegistry.

The NodeRegistry class provides a scoped, dict-backed registry for node types. A module-level default registry is created at import time and populated with all built-in nodes plus any nodes discovered via importlib.metadata entry_points (group plaita.nodes).

Backward-compatible helpers (node_register, parse_node) emit DeprecationWarning; the nodes dict proxy is still silent.

NodeRegistry ¶

NodeRegistry(*, auto_discover: bool = False, parent: Optional[NodeRegistry] = None)

Scoped registry for Plaita node types.

参数：

名称	类型	描述	默认
`auto_discover`	`bool`	If True, scan `importlib.metadata` entry_points in the `plaita.nodes` group to load plugin nodes.	`False`
`parent`	`Optional[NodeRegistry]`	If supplied, the new registry is initialised with a copy of the parent's registered nodes.	`None`

源代码位于： plaita/node/__init__.py

def __init__(
    self,
    *,
    auto_discover: bool = False,
    parent: Optional[NodeRegistry] = None,
) -> None:
    self._nodes: Dict[str, Type[Node]] = {}
    self._discovered = False
    # #98：加载失败的插件现场（ep.name -> 异常）。发现不再"静默吞掉"——
    # 解析期可据此把「节点类型不存在」纠正为「插件在、但加载失败」。
    self._load_errors: Dict[str, BaseException] = {}
    if parent is not None:
        self._nodes.update(parent._nodes)
    self._register_builtins()
    # 默认不在构造期扫描 entry_points, 避免 import plaita.node 时耦合到
    # server 等插件依赖。需要插件发现时显式传 auto_discover=True,
    # 或调用 discover() / get_default_registry()。
    if auto_discover:
        self.discover()

register ¶

register(node_cls: Type[Node]) -> Type[Node]

Register a node class. Returns node_cls so it can be used as a decorator.

覆盖已注册的不同类时发 warning——历史上静默顶掉（包括劫持内置 start/end 等类型），排查时完全无线索。同类型重复注册（如 importlib 重复加载场景）不告警。

源代码位于： plaita/node/__init__.py

def register(self, node_cls: Type[Node]) -> Type[Node]:
    """Register a node class. Returns *node_cls* so it can be used as a decorator.

    覆盖已注册的**不同**类时发 warning——历史上静默顶掉（包括劫持内置
    ``start``/``end`` 等类型），排查时完全无线索。同类型重复注册（如
    importlib 重复加载场景）不告警。
    """
    existing = self._nodes.get(node_cls.node_type)
    if existing is not None and existing is not node_cls:
        _logger.warning(
            "NodeRegistry.register: node_type %r is being overridden: "
            "%s -> %s. If this is unintentional, pick a different node_type.",
            node_cls.node_type,
            f"{existing.__module__}.{existing.__qualname__}",
            f"{node_cls.__module__}.{node_cls.__qualname__}",
        )
    self._nodes[node_cls.node_type] = node_cls
    return node_cls

parse_node ¶

parse_node(node_dict) -> Node

Create a Node instance from a dict (or return an existing Node).

源代码位于： plaita/node/__init__.py

def parse_node(self, node_dict) -> Node:
    """Create a Node instance from a dict (or return an existing Node)."""
    if isinstance(node_dict, Node):
        return node_dict

    node_type = node_dict.get("type", None)
    if not node_type:
        raise RuntimeError(f"not specific node type: {node_dict}")
    node_cls = self._nodes.get(node_type)
    if not node_cls:
        import difflib

        available = sorted(self._nodes.keys())
        close = difflib.get_close_matches(node_type, available, n=3, cutoff=0.6)
        hint = ""
        if node_type == "code":
            hint = (
                " CodeNode was moved out of the default registry in 0.4.0; "
                "call register_code_node() at startup to opt in."
            )
        elif close:
            hint = f" Did you mean {close!r}?"
        # #98：该类型本应由插件提供、但插件加载失败——别把它归因为拼写
        # 错误，直接把真实失败原因（缺模块 / import 期异常）带出去。
        load_error = self._load_errors.get(node_type)
        if load_error is not None:
            hint += (
                f" The node plugin {node_type!r} is registered but failed to load: "
                f"{type(load_error).__name__}: {load_error}."
            )
        hint += f" Available node types: {available}"
        raise RuntimeError(f"unRecognized node type: {node_type}.{hint}")
    content = node_dict.copy()
    del content["type"]
    return node_cls.model_validate(content)

copy ¶

copy() -> NodeRegistry

Return an independent shallow copy of this registry.

源代码位于： plaita/node/__init__.py

def copy(self) -> NodeRegistry:
    """Return an independent shallow copy of this registry."""
    new = NodeRegistry.__new__(NodeRegistry)
    new._nodes = dict(self._nodes)
    # #98：发现状态与失败现场随副本走——否则副本上 discover()/parse_node()
    # 会在缺失属性上抛 AttributeError。
    new._discovered = self._discovered
    new._load_errors = dict(self._load_errors)
    return new

discover ¶

discover() -> 'NodeRegistry'

Lazily discover node plugins via plaita.nodes entry points.

Idempotent: only scans on the first successful call for this registry instance. A scan that raises (entry_points() itself failing) leaves the registry undiscovered so the next call retries —

98: a transient environment failure must not freeze the node set for¶

the lifetime of the process.

源代码位于： plaita/node/__init__.py

def discover(self) -> "NodeRegistry":
    """Lazily discover node plugins via ``plaita.nodes`` entry points.

    Idempotent: only scans on the first **successful** call for this
    registry instance. A scan that raises (``entry_points()`` itself
    failing) leaves the registry undiscovered so the next call retries —
    #98: a transient environment failure must not freeze the node set for
    the lifetime of the process.
    """
    if self._discovered:
        return self
    try:
        self._discover_entry_points()
    except Exception:
        self._discovered = False
        raise
    self._discovered = True
    return self

init_default_registry ¶

init_default_registry(*extra_nodes: Type[Node], auto_discover: bool = True) -> NodeRegistry

显式 (重新) 初始化进程级默认 NodeRegistry。

历史上默认 registry 是模块级 _default_registry = NodeRegistry() 在 import 期静默创建, 插件发现 (plaita.nodes entry points) 在首次 get_default_registry() 时隐式触发——"隐式可变单例 + import 期副作用"是典型反模式: 测试间状态污染、插件加载顺序敏感、复现困难。本入口把它变成 显式启动步骤:

清空并重建默认 registry (重注册内置节点);
auto_discover=True 时扫描 plaita.nodes entry points 装载插件节点;
把 extra_nodes (节点类) 逐个注册进去;
标记为"已显式初始化", 之后 get_default_registry() 不再发隐式初始化提示。

建议在应用启动脚本顶部调用一次::

from plaita.node import init_default_registry, register_code_node
init_default_registry()              # 显式装载内置 + 插件节点
register_code_node(default_backend="docker")  # 按需 opt-in CodeNode

参数：

名称	类型	描述	默认
`*extra_nodes`	`Type[Node]`	额外要注册的 `Node` 子类 (例如自定义节点)。	`()`
`auto_discover`	`bool`	是否扫描 entry points 装载插件节点, 默认 True。	`True`

返回：

类型	描述
`NodeRegistry`	初始化后的默认 `NodeRegistry` (即 `get_default_registry()` 将返回者)。

源代码位于： plaita/node/__init__.py

def init_default_registry(*extra_nodes: Type[Node], auto_discover: bool = True) -> NodeRegistry:
    """显式 (重新) 初始化进程级默认 ``NodeRegistry``。

    历史上默认 registry 是模块级 ``_default_registry = NodeRegistry()`` 在 import
    期静默创建, 插件发现 (``plaita.nodes`` entry points) 在首次
    ``get_default_registry()`` 时隐式触发——"隐式可变单例 + import 期副作用"是
    典型反模式: 测试间状态污染、插件加载顺序敏感、复现困难。本入口把它变成
    **显式启动步骤**:

    - 清空并重建默认 registry (重注册内置节点);
    - ``auto_discover=True`` 时扫描 ``plaita.nodes`` entry points 装载插件节点;
    - 把 ``extra_nodes`` (节点类) 逐个注册进去;
    - 标记为"已显式初始化", 之后 ``get_default_registry()`` 不再发隐式初始化提示。

    建议在应用启动脚本顶部调用一次::

        from plaita.node import init_default_registry, register_code_node
        init_default_registry()              # 显式装载内置 + 插件节点
        register_code_node(default_backend="docker")  # 按需 opt-in CodeNode

    Args:
        *extra_nodes: 额外要注册的 ``Node`` 子类 (例如自定义节点)。
        auto_discover: 是否扫描 entry points 装载插件节点, 默认 True。

    Returns:
        初始化后的默认 ``NodeRegistry`` (即 ``get_default_registry()`` 将返回者)。
    """
    global _default_registry_explicitly_initialized
    reg = _default_registry
    reg._nodes.clear()
    reg._discovered = False
    # #98：显式重建即重试——上一轮的失败现场不应残留成误导性的解析提示。
    reg._load_errors.clear()
    reg._register_builtins()
    if auto_discover:
        reg.discover()
    for node_cls in extra_nodes:
        reg.register(node_cls)
    _default_registry_explicitly_initialized = True
    _logger.debug(
        "init_default_registry: %d node types registered (%s)",
        len(reg._nodes), sorted(reg._nodes.keys()),
    )
    return reg

get_default_registry ¶

get_default_registry() -> NodeRegistry

Return the process-wide default NodeRegistry.

Plugin discovery via plaita.nodes entry points happens lazily on the first call, not at import time, so importing plaita.node does not pull in optional server/plugin dependencies.

若未先调 init_default_registry, 首次调用走"隐式 discover"并 debug 提示——建议在启动脚本显式调 init_default_registry() 一次, 把进程级单例状态变成显式、可复现的启动步骤。

源代码位于： plaita/node/__init__.py

def get_default_registry() -> NodeRegistry:
    """Return the process-wide default ``NodeRegistry``.

    Plugin discovery via ``plaita.nodes`` entry points happens lazily on the
    first call, not at import time, so importing ``plaita.node`` does not pull
    in optional server/plugin dependencies.

    若未先调 ``init_default_registry``, 首次调用走"隐式 discover"并 ``debug``
    提示——建议在启动脚本显式调 ``init_default_registry()`` 一次, 把进程级
    单例状态变成显式、可复现的启动步骤。
    """
    global _default_registry_explicitly_initialized
    if not _default_registry_explicitly_initialized and not _default_registry._discovered:
        _logger.debug(
            "get_default_registry() called without explicit init_default_registry(); "
            "doing implicit plugin discovery. Call init_default_registry() at startup "
            "for an explicit, reproducible node set."
        )
    _default_registry.discover()
    return _default_registry

operator_configured_sandbox_allowed_backends ¶

operator_configured_sandbox_allowed_backends() -> tuple

运营者显式配置的沙箱后端白名单（不含生效默认后端）。

PLAITA_SANDBOX_ALLOWED_BACKENDS 未设置时返回 :data:DEFAULT_SANDBOX_ALLOWED_BACKENDS。与 :func:resolve_sandbox_allowed_backends 的区别：后者会把生效默认后端并入结果（"默认后端必须自身可用"），因此无法回答"运营者是否显式放行了某后端"。启动期降级闸门必须用本函数判断（A56-01：Docker 不可用时不得因探测失败扩集）。

源代码位于： plaita/node/__init__.py

def operator_configured_sandbox_allowed_backends() -> tuple:
    """运营者**显式配置**的沙箱后端白名单（不含生效默认后端）。

    ``PLAITA_SANDBOX_ALLOWED_BACKENDS`` 未设置时返回
    :data:`DEFAULT_SANDBOX_ALLOWED_BACKENDS`。与
    :func:`resolve_sandbox_allowed_backends` 的区别：后者会把生效默认后端并入
    结果（"默认后端必须自身可用"），因此无法回答"运营者是否显式放行了某后端"。
    启动期降级闸门必须用本函数判断（A56-01：Docker 不可用时不得因探测失败扩集）。
    """
    from . import code as _code_module

    raw = os.environ.get(SANDBOX_ALLOWED_BACKENDS_ENV, "").strip()
    if not raw:
        return tuple(DEFAULT_SANDBOX_ALLOWED_BACKENDS)
    configured = [b.strip() for b in re.split(r"[,;:\s]+", raw) if b.strip()]
    invalid = [b for b in configured if b not in _code_module._PYTHON_BACKENDS]
    if invalid:
        raise ValueError(
            f"{SANDBOX_ALLOWED_BACKENDS_ENV}: unknown sandbox backend(s) "
            f"{invalid}; supported: {sorted(_code_module._PYTHON_BACKENDS)}"
        )
    return tuple(sorted(set(configured)))

resolve_sandbox_allowed_backends ¶

resolve_sandbox_allowed_backends(default_backend: Optional[str] = None, component: str = 'runtime') -> tuple

解析部署入口生效的沙箱后端白名单（plaita#22）。

次序：显式 PLAITA_SANDBOX_ALLOWED_BACKENDS（逗号/冒号/空白分隔的后端名） → 未设则 DEFAULT_SANDBOX_ALLOWED_BACKENDS；再并入生效的默认后端（default_backend 或模块默认）——运营者选定的默认后端必须自身可用，否则不带 sandbox_backend 的流程会被自己的白名单拦下。

结果交给 register_code_node(allowed_backends=...)：硬拦载体是模块级 code._ALLOWED_SANDBOX_BACKENDS，CodeNode 解析期据此拒绝流程 JSON 里的逐节点降级。unsafe 进白名单打 CRITICAL（宿主任意代码执行）；白名单未显式配置时打 WARNING 说明当前档位。

源代码位于： plaita/node/__init__.py

def resolve_sandbox_allowed_backends(default_backend: Optional[str] = None,
                                     component: str = "runtime") -> tuple:
    """解析部署入口生效的沙箱后端白名单（plaita#22）。

    次序：显式 ``PLAITA_SANDBOX_ALLOWED_BACKENDS``（逗号/冒号/空白分隔的后端名）
    → 未设则 ``DEFAULT_SANDBOX_ALLOWED_BACKENDS``；再并入生效的默认后端
    （``default_backend`` 或模块默认）——运营者选定的默认后端必须自身可用，否则
    不带 ``sandbox_backend`` 的流程会被自己的白名单拦下。

    结果交给 ``register_code_node(allowed_backends=...)``：硬拦载体是模块级
    ``code._ALLOWED_SANDBOX_BACKENDS``，``CodeNode`` 解析期据此拒绝流程 JSON 里的
    逐节点降级。``unsafe`` 进白名单打 CRITICAL（宿主任意代码执行）；白名单未显式
    配置时打 WARNING 说明当前档位。
    """
    from . import code as _code_module

    raw = os.environ.get(SANDBOX_ALLOWED_BACKENDS_ENV, "").strip()
    configured = list(operator_configured_sandbox_allowed_backends())
    effective = default_backend or _code_module._DEFAULT_SANDBOX_BACKEND
    allowed = tuple(sorted(set(configured) | {effective}))

    if "unsafe" in allowed:
        _logger.critical(
            "[%s] 沙箱后端白名单包含 'unsafe'：流程作者可逐节点声明 "
            "sandbox_backend='unsafe' 在 %s 进程内 raw exec 任意代码（可读宿主"
            "凭据）。仅限完全信任流程作者的部署。", component, component,
        )
    elif not raw:
        _logger.warning(
            "[%s] %s 未配置，沙箱后端白名单取默认 %s ∪ 生效默认后端 %r = %s；"
            "流程 JSON 里声明的更弱后端（如 'unsafe'）将被解析期拒绝。",
            component, SANDBOX_ALLOWED_BACKENDS_ENV,
            list(DEFAULT_SANDBOX_ALLOWED_BACKENDS), effective, list(allowed),
        )
    else:
        _logger.info("[%s] 沙箱后端白名单: %s", component, list(allowed))
    return allowed

resolve_sandbox_allowed_languages ¶

resolve_sandbox_allowed_languages(component: str = 'runtime') -> tuple

解析部署入口生效的语言白名单（plaita#29）。

js 的历史实现（PyExecJS）完全绕开 sandbox_backend 档位——无隔离 / 无超时 / 无取消，且后端白名单只作用于 Python：流程作者写一行 language: "js" 即拿到宿主任意代码执行 + 读 worker 全量 env。故语言默认fail-closed：只放行 python，放行 js 必须是运营者的显式决定。

次序：显式 :data:SANDBOX_ALLOWED_LANGUAGES_ENV（逗号/冒号/空白分隔）→ 未设则 code.DEFAULT_SANDBOX_ALLOWED_LANGUAGES；python（CodeNode 的默认语言）始终保留——与后端解析同理，默认值必须自身可用，否则不带 language 的流程会被自己的白名单拦下。

结果交给 register_code_node(allowed_languages=...)：硬拦载体是模块级 code._ALLOWED_LANGUAGES，CodeNode 解析期据此拒绝 js 节点。

源代码位于： plaita/node/__init__.py

def resolve_sandbox_allowed_languages(component: str = "runtime") -> tuple:
    """解析部署入口生效的 **语言** 白名单（plaita#29）。

    ``js`` 的历史实现（PyExecJS）完全绕开 ``sandbox_backend`` 档位——无隔离 / 无
    超时 / 无取消，且后端白名单只作用于 Python：流程作者写一行 ``language: "js"``
    即拿到宿主任意代码执行 + 读 worker 全量 env。故语言默认**fail-closed**：只放行
    ``python``，放行 js 必须是运营者的显式决定。

    次序：显式 :data:`SANDBOX_ALLOWED_LANGUAGES_ENV`（逗号/冒号/空白分隔）→ 未设则
    ``code.DEFAULT_SANDBOX_ALLOWED_LANGUAGES``；``python``（CodeNode 的默认语言）
    始终保留——与后端解析同理，默认值必须自身可用，否则不带 ``language`` 的流程
    会被自己的白名单拦下。

    结果交给 ``register_code_node(allowed_languages=...)``：硬拦载体是模块级
    ``code._ALLOWED_LANGUAGES``，``CodeNode`` 解析期据此拒绝 js 节点。
    """
    from . import code as _code_module

    raw = os.environ.get(SANDBOX_ALLOWED_LANGUAGES_ENV, "").strip()
    if raw:
        configured = [lang.strip() for lang in re.split(r"[,;:\s]+", raw) if lang.strip()]
        invalid = [lang for lang in configured
                   if lang not in _code_module._BUILTIN_LANGUAGES]
        if invalid:
            raise ValueError(
                f"{SANDBOX_ALLOWED_LANGUAGES_ENV}: unknown language(s) {invalid}; "
                f"supported: {sorted(_code_module._BUILTIN_LANGUAGES)}"
            )
    else:
        configured = list(_code_module.DEFAULT_SANDBOX_ALLOWED_LANGUAGES)

    allowed = tuple(sorted(set(configured) | {_code_module.LANGUAGE_PYTHON}))
    if _code_module.LANGUAGE_JS in allowed:
        _logger.warning(
            "[%s] 语言白名单放行 'js'（来源 %s）：js 节点将按运营者配置的 "
            "sandbox_backend 档位执行（默认 docker 需拉取 node 镜像）。", component,
            SANDBOX_ALLOWED_LANGUAGES_ENV if raw else "默认值",
        )
    else:
        _logger.info("[%s] 语言白名单: %s（未放行 'js'）", component, list(allowed))
    return allowed

register_code_node ¶

register_code_node(registry: Optional[NodeRegistry] = None, default_backend: Optional[str] = None, allowed_backends: Optional[list] = None, allowed_languages: Optional[list] = None) -> None

Register :class:CodeNode for use in flows.

CodeNode executes arbitrary user-supplied code (Python exec and/or PyExecJS JS). It is therefore excluded from the default registry and must be enabled explicitly.

参数：

名称	类型	描述	默认
`registry`	`Optional[NodeRegistry]`	The :class:`NodeRegistry` to register into. If None, the process-wide default registry is used.	`None`
`default_backend`	`Optional[str]`	Python 沙箱后端默认值, 写入 `code._DEFAULT_SANDBOX_BACKEND`。未指定时沿用模块默认 (0.5.0 起 `"docker"`)。	`None`
`allowed_backends`	`Optional[list]`	沙箱后端白名单（2026-09 安全评审 P0）。设置后，流程 JSON 里声明的 `sandbox_backend` 不在白名单内即解析期硬失败——没有它，流程作者可把运营者选定的默认后端逐节点覆盖为 `"unsafe"`（宿主任意代码执行）。多租户部署务必显式设置，且不要发放 `unsafe`/`subprocess`:: `register_code_node(default_backend="docker", allowed_backends=("docker",))` 生产部署入口应经 :func:`resolve_sandbox_allowed_backends` 取值（读 `PLAITA_SANDBOX_ALLOWED_BACKENDS`，默认 `(docker,)`），而不是自己拼元组——见 `plaita/server/flow_worker.py` 与 console `main.py`。安全 gate: 若生效后端 (`default_backend` 或模块默认) 为 `"docker"` 但当前环境 docker daemon 不可用, 拒绝注册并抛 `RuntimeError`, 指明三种降级路径——装 Docker / `default_backend="subprocess"` / `default_backend="unsafe"`。不允许静默降级到 `"restricted"` (其 AST 沙箱有已知绕过向量, 不该作为"对用户透明"的兜底)。	`None`
`allowed_languages`	`Optional[list]`	语言白名单（2026-10 安全评审 P1，plaita#29）。默认 `("python",)`——`js` 的历史实现（PyExecJS）完全绕开档位体系（无隔离 / 无超时 / 无取消），流程作者写一行 `language: "js"` 即可执行宿主任意代码并读 worker 全量 env。放行 js 是运营者的显式决定，且放行后 js 仍按 `sandbox_backend` 走档位表（`restricted` 没有 js 实现，解析期拒绝）:: `register_code_node(allowed_backends=("docker",), allowed_languages=("python", "js"))` 生产部署入口应经 :func:`resolve_sandbox_allowed_languages` 取值（读 `PLAITA_SANDBOX_ALLOWED_LANGUAGES`，默认 `(python,)`）。	`None`

Example::

from plaita.node import register_code_node
register_code_node()                      # 默认 docker, 需 daemon
register_code_node(default_backend="subprocess")  # 半信任, 无需 docker

.. warning:: Only call this if you trust all flow definitions that will be executed in this process. A CodeNode in a flow JSON allows any code the flow author chooses to run; "unsafe" 后端连文件系统/网络访问都不限制（python exec 与 js PyExecJS 都没有超时/取消），js 语言还须 allowed_languages 显式放行。

``"subprocess"`` **不是租户凭据隔离边界**（plaita#256 首版传播租户与
凭据路径，A42-06 收窄）：子进程与 worker 同用户、文件系统不受限，
且所有租户共用同一把解密密钥，租户代码可以改写租户上下文或直接读取
default 凭据文件。因此非 default 租户在 spawn 前即被拒绝
（``TenantContextPropagationError``，用户代码不执行）；default 租户
语义不变。多租户 / 租户级 code 节点必须用 ``"docker"``（容器内不挂载
宿主凭据文件）。

源代码位于： plaita/node/__init__.py

def register_code_node(registry: Optional[NodeRegistry] = None,
                       default_backend: Optional[str] = None,
                       allowed_backends: Optional[list] = None,
                       allowed_languages: Optional[list] = None) -> None:
    """Register :class:`CodeNode` for use in flows.

    ``CodeNode`` executes **arbitrary user-supplied code** (Python ``exec``
    and/or PyExecJS JS).  It is therefore excluded from the default registry
    and must be enabled explicitly.

    Args:
        registry: The :class:`NodeRegistry` to register into.  If *None*, the
            process-wide default registry is used.
        default_backend: Python 沙箱后端默认值, 写入 ``code._DEFAULT_SANDBOX_BACKEND``。
            未指定时沿用模块默认 (0.5.0 起 ``"docker"``)。
        allowed_backends: 沙箱后端白名单（2026-09 安全评审 P0）。设置后，流程 JSON
            里声明的 ``sandbox_backend`` 不在白名单内即**解析期硬失败**——没有它，
            流程作者可把运营者选定的默认后端逐节点覆盖为 ``"unsafe"``（宿主任意
            代码执行）。多租户部署务必显式设置，且不要发放 ``unsafe``/``subprocess``::

                register_code_node(default_backend="docker",
                                   allowed_backends=("docker",))

            生产部署入口应经 :func:`resolve_sandbox_allowed_backends` 取值（读
            ``PLAITA_SANDBOX_ALLOWED_BACKENDS``，默认 ``(docker,)``），而不是自己
            拼元组——见 ``plaita/server/flow_worker.py`` 与 console ``main.py``。

            **安全 gate**: 若生效后端 (``default_backend`` 或模块默认) 为 ``"docker"``
            但当前环境 docker daemon 不可用, **拒绝注册**并抛 ``RuntimeError``, 指明
            三种降级路径——装 Docker / ``default_backend="subprocess"`` /
            ``default_backend="unsafe"``。不允许静默降级到 ``"restricted"`` (其 AST
            沙箱有已知绕过向量, 不该作为"对用户透明"的兜底)。
        allowed_languages: 语言白名单（2026-10 安全评审 P1，plaita#29）。默认
            ``("python",)``——``js`` 的历史实现（PyExecJS）完全绕开档位体系（无隔离 /
            无超时 / 无取消），流程作者写一行 ``language: "js"`` 即可执行宿主任意
            代码并读 worker 全量 env。放行 js 是运营者的显式决定，且放行后 js 仍按
            ``sandbox_backend`` 走档位表（``restricted`` 没有 js 实现，解析期拒绝）::

                register_code_node(allowed_backends=("docker",),
                                   allowed_languages=("python", "js"))

            生产部署入口应经 :func:`resolve_sandbox_allowed_languages` 取值（读
            ``PLAITA_SANDBOX_ALLOWED_LANGUAGES``，默认 ``(python,)``）。

    Example::

        from plaita.node import register_code_node
        register_code_node()                      # 默认 docker, 需 daemon
        register_code_node(default_backend="subprocess")  # 半信任, 无需 docker

    .. warning::
        Only call this if you trust all flow definitions that will be executed
        in this process. A ``CodeNode`` in a flow JSON allows any code the flow
        author chooses to run; ``"unsafe"`` 后端连文件系统/网络访问都不限制
        （python ``exec`` 与 js PyExecJS 都没有超时/取消），js 语言还须
        ``allowed_languages`` 显式放行。

        ``"subprocess"`` **不是租户凭据隔离边界**（plaita#256 首版传播租户与
        凭据路径，A42-06 收窄）：子进程与 worker 同用户、文件系统不受限，
        且所有租户共用同一把解密密钥，租户代码可以改写租户上下文或直接读取
        default 凭据文件。因此非 default 租户在 spawn 前即被拒绝
        （``TenantContextPropagationError``，用户代码不执行）；default 租户
        语义不变。多租户 / 租户级 code 节点必须用 ``"docker"``（容器内不挂载
        宿主凭据文件）。
    """
    from . import code as _code_module
    effective = default_backend if default_backend is not None else _code_module._DEFAULT_SANDBOX_BACKEND
    if effective == "docker" and not _code_module._docker_available():
        raise RuntimeError(
            "register_code_node: default sandbox_backend is 'docker' but the "
            "Docker daemon is not available. Either install/start Docker, or "
            "explicitly choose a weaker backend: "
            "register_code_node(default_backend='subprocess') (process-level, "
            "no FS/network isolation) or "
            "register_code_node(default_backend='unsafe') (no sandbox, only "
            "for fully-trusted flow authors). 'restricted' is no longer the "
            "default because RestrictedPython has known AST bypass vectors."
        )
    if default_backend is not None:
        _code_module._DEFAULT_SANDBOX_BACKEND = default_backend
    if allowed_backends is not None:
        invalid = [b for b in allowed_backends if b not in _code_module._PYTHON_BACKENDS]
        if invalid:
            raise ValueError(
                f"register_code_node: unknown sandbox backend(s) {invalid}; "
                f"supported: {sorted(_code_module._PYTHON_BACKENDS)}"
            )
        _code_module._ALLOWED_SANDBOX_BACKENDS = frozenset(allowed_backends)
    if allowed_languages is not None:
        invalid = [lang for lang in allowed_languages
                   if lang not in _code_module._BUILTIN_LANGUAGES]
        if invalid:
            raise ValueError(
                f"register_code_node: unknown language(s) {invalid}; "
                f"supported: {sorted(_code_module._BUILTIN_LANGUAGES)}"
            )
        _code_module._ALLOWED_LANGUAGES = frozenset(allowed_languages)
    target = registry if registry is not None else get_default_registry()
    target.register(CodeNode)

node_register ¶

node_register(node_cls: Type[Node]) -> None

Register a node class on the default registry.

.. deprecated:: Use get_default_registry().register(node_cls) instead.

源代码位于： plaita/node/__init__.py

def node_register(node_cls: Type[Node]) -> None:
    """Register a node class on the default registry.

    .. deprecated::
        Use ``get_default_registry().register(node_cls)`` instead.
    """
    warnings.warn(
        "node_register() is deprecated. "
        "Use get_default_registry().register(node_cls) instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    _default_registry.register(node_cls)

parse_node ¶

parse_node(node_dict) -> Node

Parse a node dict using the default registry.

.. deprecated:: Use get_default_registry().parse_node(node_dict) instead.

源代码位于： plaita/node/__init__.py

def parse_node(node_dict) -> Node:
    """Parse a node dict using the default registry.

    .. deprecated::
        Use ``get_default_registry().parse_node(node_dict)`` instead.
    """
    warnings.warn(
        "Module-level parse_node() is deprecated. "
        "Use get_default_registry().parse_node(node_dict) instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return _default_registry.parse_node(node_dict)

basic ¶

Node ¶

Bases: BaseModel

validate ¶

validate() -> None

构造期校验节点配置。由 FlowBuilder.validate 调用，子类可覆写（如 Switch / HTTP）以在流程构建阶段就拦下非法配置。

源代码位于： plaita/node/basic.py

def validate(self) -> None:
    """构造期校验节点配置。由 ``FlowBuilder.validate`` 调用，子类可覆写
    （如 ``Switch`` / ``HTTP``）以在流程构建阶段就拦下非法配置。"""
    return None

execute ¶

execute(execution: 'NodeExecutionContext') -> Any

执行节点并返回输出。

execution 满足 NodeExecutionContext（evaluate / state / child）。运行时传入的是 FlowExecution facade，它实现该 Protocol。

源代码位于： plaita/node/basic.py

def execute(self, execution: "NodeExecutionContext") -> Any:
    """执行节点并返回输出。

    ``execution`` 满足 ``NodeExecutionContext``（evaluate / state / child）。
    运行时传入的是 ``FlowExecution`` facade，它实现该 Protocol。
    """
    raise NotImplementedError()

resume ¶

resume(execution: 'NodeExecutionContext', resume_type, resume_data=None) -> Any

恢复一个此前挂起的节点 (仅 is_suspending 节点需要实现)。

内核 DistributedStrategy._handle_resume 通过本方法多态分发 resume, 避免直接 isinstance 具体挂起节点类型 (如 EventNode), 让 core 层不反向依赖 node 插件层。基类默认抛错, 挂起型节点覆写之。

源代码位于： plaita/node/basic.py

def resume(self, execution: "NodeExecutionContext", resume_type, resume_data=None) -> Any:
    """恢复一个此前挂起的节点 (仅 ``is_suspending`` 节点需要实现)。

    内核 ``DistributedStrategy._handle_resume`` 通过本方法多态分发 resume,
    避免直接 ``isinstance`` 具体挂起节点类型 (如 EventNode), 让 core 层
    不反向依赖 node 插件层。基类默认抛错, 挂起型节点覆写之。
    """
    raise NotImplementedError(
        f"Node {self.id} ({type(self).__name__}) is not a suspending node and "
        f"cannot be resumed; is_suspending={self.is_suspending}"
    )

warn_unknown_keys ¶

warn_unknown_keys(cls, values)

Schema 卫生：原始 dict 里的未知键告警（不报错）。

pydantic 对未声明字段默认 extra="ignore"——JSON 流程里拼错的字段名（如 "conditon"）会被静默吞掉，流程带着错误配置"成功"运行。0.5.0 的一贯策略是把沉默变可见：逐键对照声明字段+alias+LEGACY_KEYS，未知即 logger.warning（带节点 id 与合法键提示），不改变解析结果。

源代码位于： plaita/node/basic.py

def warn_unknown_keys(cls, values):
    """Schema 卫生：原始 dict 里的未知键告警（不报错）。

    pydantic 对未声明字段默认 ``extra="ignore"``——JSON 流程里拼错的字段名
    （如 ``"conditon"``）会被静默吞掉，流程带着错误配置"成功"运行。0.5.0 的
    一贯策略是把沉默变可见：逐键对照 声明字段+alias+LEGACY_KEYS，未知即
    ``logger.warning``（带节点 id 与合法键提示），不改变解析结果。
    """
    if not isinstance(values, dict):
        return values
    known = _known_input_keys(cls)
    unknown = sorted(k for k in values if isinstance(k, str) and k not in known)
    if unknown:
        _logger.warning(
            "%s %r: unknown config keys %s will be IGNORED (possible typo?). "
            "Known keys: %s",
            cls.__name__, values.get("id", "?"), unknown, sorted(known),
        )
    return values

plaita.node¶

node ¶

NodeRegistry ¶

register ¶

parse_node ¶

copy ¶

discover ¶

98: a transient environment failure must not freeze the node set for¶

init_default_registry ¶

get_default_registry ¶

operator_configured_sandbox_allowed_backends ¶

resolve_sandbox_allowed_backends ¶

resolve_sandbox_allowed_languages ¶

register_code_node ¶

node_register ¶

parse_node ¶

basic ¶

Node ¶

validate ¶

execute ¶

resume ¶

warn_unknown_keys ¶

`plaita.node`¶