plaita.core.executor — FlowExecution driver facade.
0.5.0 起, 执行策略层 (ExecutionMode / ExecutionStrategy Protocol / 三个
策略 NormalStrategy / GeneratorStrategy / DistributedStrategy / 模块级
helper _advance_one / _create_lazy_output / _create_end_output /
_subscribe_event) 已拆到 plaita.core.strategies。本模块只保留
FlowExecution facade——组合 ExecutionContext / NodeRunner /
CallbackManager / 策略, 提供公共运行入口 (run / arun / debug /
run_compatible / arun_compatible / run_distributed)。
SC-003 对类体是软预算（见 tests/e2e/test_success_criteria.py）：显式委托
属性允许膨胀，不必为过门再拆 mixin。
历史导入路径 from plaita.core.executor import ExecutionMode / NormalStrategy /
GeneratorStrategy / DistributedStrategy / ExecutionStrategy / RunOptions /
_subscribe_event 通过下方 re-export 保持不变, 不影响既有调用方与测试。
FlowExecution(parent: Optional[FlowExecution] = None, verbose: bool = False, mode: Optional[str] = None, callback_handlers: Optional[List[FlowCallback]] = None, callback_manager: Optional[BaseCallbackManager] = None, event_bus: Optional[EventBus] = None, registry=None)
Bases: ExpressNamespaceMixin, CompatRunMixin
Thin facade composing ExecutionContext, NodeRunner, and strategies.
Nodes receive this instance as the execution parameter. State and
context access is delegated to the underlying ExecutionContext via
explicit properties/methods — there is no __getattr__/__setattr__
catch-all, so execution.foo is either a real attribute, an explicit
delegate, or a loud AttributeError. No more phantom attributes that
silently land in context state.
源代码位于： plaita/core/executor.py
| def __init__(
self,
parent: Optional[FlowExecution] = None,
verbose: bool = False,
mode: Optional[str] = None,
callback_handlers: Optional[List[FlowCallback]] = None,
callback_manager: Optional[BaseCallbackManager] = None,
event_bus: Optional[EventBus] = None,
registry=None,
):
self.options = RunOptions(mode=_coerce_mode(mode), timeout=None)
self.parent = parent
self.verbose = verbose
# 可选自定义节点注册表。``None`` 时沿用 ``flow`` 自身已解析的节点
# (常规路径: ``Flow.model_validate`` 已用默认 registry 解析过)。传入
# 后会在执行前对含 dict 节点的 flow 兜底解析一次，支持"用一个隔离
# registry 跑某个 flow"的场景而不依赖进程级单例。
self._registry = registry
self._ctx = ExecutionContext(
parent=parent._ctx if parent else None, event_bus=event_bus,
**_inherited_express(parent),
)
# 非重入守卫: FlowExecution 的 $INPUT/$NODE/LAST_NODE 全是实例级状态,
# 两个线程同时对同一实例 run 会互相踩状态（静默结果串扰）。子流程走
# get_child_execution() 的新实例, 不受影响。
self._running = False
if callback_manager:
self.callback_manager = callback_manager
for handler in callback_handlers or []:
self.callback_manager.add_handler(handler)
else:
self.callback_manager = CallbackManager(callback_handlers or [])
if self.verbose:
self.callback_manager.add_handler(LoggerCallback())
self._runner = NodeRunner(self._ctx, node_execution=self)
self._strategies = {
ExecutionMode.NORMAL.value: NormalStrategy(),
ExecutionMode.GENERATOR.value: GeneratorStrategy(),
ExecutionMode.DISTRIBUTED.value: DistributedStrategy(),
}
|
property
执行级取消意图（粘滞；与每节点 clear 的 cancel_event 职责拆分）。
property
Typed, None-normalized view over context state.
execution.state.flow_id / .last_node_id / .last_branch.
Replaces the former execution.flow_id etc. bare-attribute
pass-throughs (break change — see MIGRATION.md).
请求取消本执行（幂等，多次调用无害）。
同时置位两个 Event，各司其职：
cancel_event：节点级协作取消信号——code 沙箱等待循环
（code._popen_wait_cancellable）观察到置位当场 killpg 整个
进程树；历史 sync 超时置位/下节点入口 clear 语义不变。cancel_requested：粘滞的执行级取消意图——引擎在节点边界
拒绝继续（抛 FlowCancelledException），不被任何节点入口 clear。
子执行与父执行共享同一对 Event 实例（ExecutionContext 父子共享
规则），对链上任一执行调用 cancel() 即传播到整条执行链。取消的
默认语义是步界取消（在途非沙箱节点跑完当前节点，§3.3）。
源代码位于： plaita/core/executor.py
| def cancel(self) -> None:
"""请求取消本执行（幂等，多次调用无害）。
同时置位两个 Event，各司其职：
- ``cancel_event``：节点级协作取消信号——code 沙箱等待循环
（``code._popen_wait_cancellable``）观察到置位当场 killpg 整个
进程树；历史 sync 超时置位/下节点入口 clear 语义不变。
- ``cancel_requested``：粘滞的执行级取消意图——引擎在节点边界
拒绝继续（抛 ``FlowCancelledException``），不被任何节点入口 clear。
子执行与父执行共享同一对 Event 实例（``ExecutionContext`` 父子共享
规则），对链上任一执行调用 ``cancel()`` 即传播到整条执行链。取消的
默认语义是步界取消（在途非沙箱节点跑完当前节点，§3.3）。
"""
self._ctx.cancel_event.set()
self._ctx.cancel_requested.set()
|
set_scratch(key: str, value: Any) -> None
plaita#164：把 value 声明为 key 的 scratch 绑定（不进 checkpoint）。
源代码位于： plaita/core/executor.py
| def set_scratch(self, key: str, value: Any) -> None:
"""plaita#164：把 *value* 声明为 *key* 的 scratch 绑定（不进 checkpoint）。"""
return self._ctx.set_scratch(key, value)
|
plaita#164：当前 scratch 绑定的键名。
源代码位于： plaita/core/executor.py
| def scratch_keys(self):
"""plaita#164：当前 scratch 绑定的键名。"""
return self._ctx.scratch_keys()
|
clear_scratch(key=None) -> None
plaita#164：丢掉全部（或指定键的）scratch 绑定。
源代码位于： plaita/core/executor.py
| def clear_scratch(self, key=None) -> None:
"""plaita#164：丢掉全部（或指定键的）scratch 绑定。"""
return self._ctx.clear_scratch(key)
|
run_distributed(flow, params: Optional[Dict] = None, *, saved_context: Optional[Dict] = None, resume_type: str = 'continue', resume_data: Optional[Any] = None, timeout: Optional[int] = None, execution_id: Optional[str] = None) -> Dict
Drive one distributed step, reusing this execution's context /
runner / callback manager so user callbacks persist across steps.
Prefer this over the FlowExecution.run classmethod when you need
to advance a distributed flow node-by-node without losing callbacks.
execution_id：预铸种子（首跑传，续跑不传）——worker 先落 running
行再执行时，行 id 与引擎 result.execution_id 保持一致（P0 可见性）。
「复用同一实例」指跨步骤顺序复用（每步调用返回后再调下一步）；
并发重叠调用同一实例仍然禁止——review-fix B4 起套用与
run_compatible 相同的 _begin_run 非重入守卫（每步调用期间
持有，返回即释放，顺序复用零额外成本）。
源代码位于： plaita/core/executor.py
| def run_distributed(
self,
flow,
params: Optional[Dict] = None,
*,
saved_context: Optional[Dict] = None,
resume_type: str = "continue",
resume_data: Optional[Any] = None,
timeout: Optional[int] = None,
execution_id: Optional[str] = None,
) -> Dict:
"""Drive one distributed step, **reusing this execution's** context /
runner / callback manager so user callbacks persist across steps.
Prefer this over the ``FlowExecution.run`` classmethod when you need
to advance a distributed flow node-by-node without losing callbacks.
``execution_id``：预铸种子（首跑传，续跑不传）——worker 先落 running
行再执行时，行 id 与引擎 result.execution_id 保持一致（P0 可见性）。
「复用同一实例」指跨步骤**顺序**复用（每步调用返回后再调下一步）；
并发重叠调用同一实例仍然禁止——review-fix B4 起套用与
``run_compatible`` 相同的 ``_begin_run`` 非重入守卫（每步调用期间
持有，返回即释放，顺序复用零额外成本）。
"""
self._begin_run()
try:
self._ensure_flow_resolved(flow)
timeout = _distributed_timeout_or_raise(
flow, self.timeout, timeout, self.callback_manager)
# 种子仅在调用方显式给出时透传——不带新 kwarg 地保持既有
# execute 签名兼容（DistributedStrategy 经 **options 消化）。
seed = {"execution_id": execution_id} if execution_id else {}
coro = self._strategies[ExecutionMode.DISTRIBUTED.value].execute(
flow, self._ctx, self._runner, self.callback_manager, params, timeout,
saved_context=saved_context, resume_type=resume_type, resume_data=resume_data,
**seed,
)
# 分布式每步独立 loop：步内同包 flow-scoped HTTP session（复用窗口=本步）
coro = _flow_session_scoped(coro)
# 历史上 run_distributed 把任何异常（含具体的 FlowExecutionException 子类）
# 归一化为 FLOW_ERROR / -500 作为分布式对外契约；此处保留该契约，
# 具体子类仅用于内部抛点与 normal 模式（_finish_normal 让其透传）。
try:
return _run_async_sync(coro)
except Exception as e:
_raise_distributed_error(e, flow, self.callback_manager)
finally:
self._running = False
|
run_compatible(flow, lazy, *args, **kwargs)
Sync execution. Returns the result, or a sync generator when lazy.
In lazy/generator mode on_flow_end is deferred until the returned
generator is actually consumed or closed — historically it fired
immediately with the unconsumed generator as result, which meant
the lifecycle end callback ran before any node executed.
源代码位于： plaita/core/executor_compat.py
| def run_compatible(self, flow, lazy, *args, **kwargs):
"""Sync execution. Returns the result, or a sync generator when lazy.
In lazy/generator mode ``on_flow_end`` is deferred until the returned
generator is actually consumed or closed — historically it fired
immediately with the unconsumed generator as ``result``, which meant
the lifecycle end callback ran before any node executed.
"""
self._begin_run()
try:
try:
return _drive_strategy(
self._prepare_strategy(flow, lazy, args, kwargs),
lazy=lazy, sync=True,
finish_coro=lambda coro: _finish_normal(coro, flow, self.callback_manager),
on_lazy_finally=lambda exc: (
_emit_flow_end_on_close(flow, exc, self.callback_manager), setattr(self, "_running", False),
),
)
except BaseException:
# review-fix B3: lazy 模式下 generator 交出前抛错（典型是
# _prepare_strategy 校验失败）时 on_lazy_finally 尚未注册，
# 不复位 _running 会永久毒化实例——下次 run 报 "already
# running" 伪装真实死因。交出之后由 on_lazy_finally 复位。
if lazy:
self._running = False
raise
finally:
if not lazy:
self._running = False
|
async
arun_compatible(flow, lazy, *args, **kwargs)
Async execution — canonical path.
源代码位于： plaita/core/executor_compat.py
| async def arun_compatible(self, flow, lazy, *args, **kwargs):
"""Async execution — canonical path."""
self._begin_run()
try:
try:
driven = _drive_strategy(
self._prepare_strategy(flow, lazy, args, kwargs),
lazy=lazy, sync=False,
finish_coro=lambda coro: _finish_normal(coro, flow, self.callback_manager),
on_lazy_finally=lambda exc: (
_emit_flow_end_on_close(flow, exc, self.callback_manager), setattr(self, "_running", False),
),
)
except BaseException:
# review-fix B3: 同 run_compatible——lazy 交出 generator 前
# 抛错必须复位 _running，否则实例永久毒化。
if lazy:
self._running = False
raise
if lazy:
return driven
return await driven
finally:
if not lazy:
self._running = False
|