plaita.core.context — ExecutionContext for flow runtime state management.
Extracted from FlowExecution: state storage, variable scoping ($INPUT,
$NODE, $GLOBAL, $PARENT, $ENV), parent-child chain, env var filtering,
and expression evaluation delegation.
BranchCancelEvent(parent: Optional[Event] = None)
Bases: Event
节点级协作取消信号：分支独立置位，父级置位单向可见（plaita#206）。
历史实现让子上下文直接复用父的 threading.Event（parent.cancel_event
同一实例），于是 Parallel 的每个分支共享一个节点级信号：A 分支的 sync
节点超时置位后，B 分支正在跑的 code 子进程等待循环
（code._popen_wait_cancellable）看到置位即 killpg 整棵进程树——兄弟
分支被误杀。
本类给每个子上下文一个自己的 threading.Event，并把父级 Event 作为
「上游」只读查询：
set() / clear() 只作用于本分支（A 的超时不会传播给 B）；is_set() = 本分支置位 或 父级置位——父执行的 Parallel 节点超时、
cancel() 仍能向下传播（既有语义不丢）；wait() 轮询两级（父级置位不会 notify 本 Event，故不能用裸
super().wait()）。
执行级取消意图 cancel_requested 仍是父子共享的同一实例（粘滞、节点入口
只读不 clear），不受本类影响。
源代码位于： plaita/core/context.py
| def __init__(self, parent: Optional[threading.Event] = None) -> None:
super().__init__()
self._parent_cancel_event = parent
|
ExecutionContext(parent: Optional[ExecutionContext] = None, *, express_prefix: str = '$', express_input_name: str = 'INPUT', express_parent_name: str = 'PARENT', express_node_name: str = 'NODE', express_global_name: str = 'GLOBAL', express_environment_variable: str = 'ENV', event_bus=None, evaluator: Optional[ExpressionEvaluator] = None, expose_env: Optional[List[str]] = None)
Bases: _SystemStateAccessors
Runtime state facade: CheckpointState + parent/event_bus/cancel/evaluator.
In-memory-only pieces never enter a checkpoint; state R/W goes through
CheckpointState so the magic-string key schema lives in one place.
源代码位于： plaita/core/context.py
| def __init__(
self,
parent: Optional[ExecutionContext] = None,
*,
express_prefix: str = "$",
express_input_name: str = "INPUT",
express_parent_name: str = "PARENT",
express_node_name: str = "NODE",
express_global_name: str = "GLOBAL",
express_environment_variable: str = "ENV",
event_bus=None,
evaluator: Optional[ExpressionEvaluator] = None,
expose_env: Optional[List[str]] = None,
) -> None:
self.parent = parent
self.event_bus = event_bus
# plaita#237：租户是执行身份的一部分，在**构造期**钉在对象上——进程池
# 的 pickler 跑在队列线程，那里 current_tenant() 是 default；只在
# __getstate__ 读 ContextVar 会把租户丢掉。子上下文继承父执行身份。
parent_tenant = getattr(parent, "_tenant_id", None) if parent is not None else None
self._tenant_id = parent_tenant or current_tenant()
# plaita#239：来源标记同样属于执行身份，构造期钉住；子上下文继承。
parent_source = getattr(parent, "_tenant_source", None) if parent is not None else None
self._tenant_source = parent_source or current_tenant_source()
self.expose_env = list(expose_env) if expose_env else []
# plaita#141：**构造期**继承的 `$ENV` allowlist（仅子上下文可能非空）。
# #106 起子执行继承父级 express_* 命名空间，但 `expose_env` 此前只落在上
# 一行：`setup_flow` 会按「被执行 flow 自己声明的值」整体覆盖它，父级
# allowlist 在子执行里不可观测（子 `$ENV` 恒空，读单个变量靠 `$PARENT`
# 回落兜底才不显形）。`setup_flow` 因此需要一份**不被 flow 覆盖**的继承
# 快照——根上下文（无 parent）恒为空，避免同一实例跨 run 把上一次 flow
# 的 allowlist 泄漏给下一次（历史上 `clean()` 只负责快照、不负责清值）。
self._inherited_expose_env = (
list(expose_env) if (parent is not None and expose_env) else []
)
# 节点级取消信号（plaita#206）：子上下文持有**自己**的 Event，父级
# Event 单向只读——分支超时不再误杀兄弟分支，父级取消仍向下传播。
# 跨进程由 __getstate__/__setstate__ 重建（进程模式本就不跨进程传播）。
self.cancel_event = (
BranchCancelEvent(parent.cancel_event) if parent is not None
else threading.Event()
)
# 执行级取消意图（2026-10 波次③）：与 cancel_event 职责拆分——
# cancel_event 是"当前节点"级信号（sync 超时置位、下节点入口 clear），
# cancel_requested 是粘滞的"整个执行"级取消意图（cancel() 置位、
# 节点入口只读不 clear）。父子共享规则与 cancel_event **不同**：
# 取消意图必须沿整条执行链共享（plaita#206 边界：父级取消走这里）。
self.cancel_requested = (
parent.cancel_requested if parent is not None else threading.Event()
)
self.express_prefix = express_prefix
self.express_input_name = express_input_name
self.express_parent_name = express_parent_name
self.express_node_name = express_node_name
self.express_global_name = express_global_name
self.express_environment_variable = express_environment_variable
# plaita#72：子上下文在**构造这一刻**钉住父状态快照。历史实现把 live 父
# 状态留到 ``setup_flow`` 才包一层 ``_LazyRootSnapshot``，而真正的物化又
# 推迟到子侧首读——只有「父在子执行存续期内被阻塞」时才等价于快照。
# ``Parallel`` 的 fire-and-forget 后台分支不阻塞父：父在分支提交后继续
# 推进并写 ``$NODE``，分支稍后的首读会读进父的"未来态"。构造期快照把
# 时点前移到「子执行创建」，配合 ``concurrent._execute_background_branches``
# 在 submit 前创建子执行，即得分支创建时冻结。
self._parent_snapshot = (
_LazyRootSnapshot(parent.context) if parent is not None else None
)
self._state = CheckpointState(**self._state_kwargs())
self._state[f"{express_prefix}EXECUTION_ID"] = uuid.uuid4().hex
self._evaluator = evaluator or ExpressionEvaluator()
|
set_scratch(key: str, value: Any) -> None
把 value 声明为 key 的 scratch 绑定（plaita#164）。
"这个对象只归本次执行所有、可以被就地改写"。在 #69 的
默认拒绝就地改写下恢复便利写法::
ctx.set_scratch("$V", {"a": 1}) # 显式声明
parser.evaluate("$F.set($V,'k','v')", ctx.context) # 就地写，允许
边界（详见 :meth:CheckpointState.set_scratch）：只登记被声明的那个
对象（不登记它的图）、只对下一次读取它的那次求值有效、不进
checkpoint、不得占用 checkpoint 键。
源代码位于： plaita/core/context.py
| def set_scratch(self, key: str, value: Any) -> None:
"""把 *value* 声明为 *key* 的 **scratch** 绑定（plaita#164）。
"这个对象只归本次执行所有、可以被就地改写"。在 ``#69`` 的
**默认拒绝就地改写**下恢复便利写法::
ctx.set_scratch("$V", {"a": 1}) # 显式声明
parser.evaluate("$F.set($V,'k','v')", ctx.context) # 就地写，允许
边界（详见 :meth:`CheckpointState.set_scratch`）：只登记**被声明的那个
对象**（不登记它的图）、**只对下一次读取它的那次求值有效**、**不进
checkpoint**、不得占用 checkpoint 键。
"""
self._state.set_scratch(key, value)
|
scratch_keys() -> List[str]
当前 scratch 绑定的键名（这些键不在 checkpoint 里）。
源代码位于： plaita/core/context.py
| def scratch_keys(self) -> List[str]:
"""当前 scratch 绑定的键名（这些键不在 checkpoint 里）。"""
return self._state.scratch_keys()
|
clear_scratch(key: Optional[str] = None) -> None
丢掉全部 scratch 绑定（或只丢 key）；checkpoint 键不受影响。
源代码位于： plaita/core/context.py
| def clear_scratch(self, key: Optional[str] = None) -> None:
"""丢掉全部 scratch 绑定（或只丢 *key*）；checkpoint 键不受影响。"""
self._state.clear_scratch(key)
|
clean(execution_id: Optional[str] = None) -> None
Reset state for a fresh run and re-sync cancel_event/cancel_requested.
expose_env is not cleared here — setup_flow overwrites it from
flow.expose_env. Root gets a new Event; child re-syncs to parent.
execution_id 可由调用方预铸（worker 先落 running 行再执行的可见性
修复），缺省仍就地铸造。
源代码位于： plaita/core/context.py
| def clean(self, execution_id: Optional[str] = None) -> None:
"""Reset state for a fresh run and re-sync cancel_event/cancel_requested.
``expose_env`` is not cleared here — ``setup_flow`` overwrites it from
``flow.expose_env``. Root gets a new Event; child re-syncs to parent.
``execution_id`` 可由调用方预铸（worker 先落 running 行再执行的可见性
修复），缺省仍就地铸造。
"""
self._state = CheckpointState.fresh(
**self._state_kwargs(),
execution_id=execution_id or uuid.uuid4().hex,
env=_safe_environment(self.expose_env),
)
self.cancel_event = (
BranchCancelEvent(self.parent.cancel_event) if self.parent is not None
else threading.Event()
)
self.cancel_requested = (
self.parent.cancel_requested if self.parent is not None
else threading.Event()
)
# 根上下文 fresh run 重新钉住当前租户；子上下文继承父执行身份
# （plaita#237，与 __init__ 同口径）。来源标记同理（plaita#239）。
self._tenant_id = (
getattr(self.parent, "_tenant_id", None) if self.parent is not None
else current_tenant()
) or current_tenant()
self._tenant_source = (
getattr(self.parent, "_tenant_source", None) if self.parent is not None
else current_tenant_source()
) or current_tenant_source()
|
setup_flow(flow, args: tuple, kwargs: dict) -> None
Populate $INPUT/$PARENT/$GLOBAL/$FLOW_ID/$ENV and sync expose_env.
源代码位于： plaita/core/context.py
| def setup_flow(self, flow, args: tuple, kwargs: dict) -> None:
"""Populate $INPUT/$PARENT/$GLOBAL/$FLOW_ID/$ENV and sync expose_env."""
flow_allowlist = list(getattr(flow, "expose_env", None) or [])
if not flow_allowlist:
# plaita#141：被执行 flow 未声明 `expose_env` 时，子执行继承父级
# allowlist（构造期快照，见 __init__）——与 `ExecutionContext.child()`
# 一直以来的行为对齐，也让 #106「子执行继承父 express 命名空间」对
# `$ENV` 真正可观测（此前子 `$ENV` 恒空，读变量靠 $PARENT 回落兜底）。
# 根上下文（无 parent）该快照恒为空 ⇒ 默认 $ENV 用户零行为变更。
flow_allowlist = list(self._inherited_expose_env)
# review-fix B3（伴生）: 先验证新 allowlist 再提交——否则非法 allowlist
# （如 ["*"]）在此处抛错时 expose_env 已被污染，同一实例下次 run 的
# clean() 立刻用脏值再次爆炸，把「prepare 抛错」放大成「实例报废」。
env = _safe_environment(flow_allowlist)
self.expose_env = flow_allowlist
# review-fix B2: 必须是**物化边界**的私有快照——浅 copy 的话 run 内节点写
# $GLOBAL 嵌套值（如 $GLOBAL.cfg.timeout = 30）会穿透进 Flow 定义本体，
# 同一 Flow 实例的下一次 run 继承脏值（跨 run 状态泄漏）。每次 run 仅此
# 一次拷贝，非热路径。plaita#69 r7：这里不再用 ``copy.deepcopy``（它会调用
# 宿主类的 ``__deepcopy__``，是活对象混进 freshness 作用域的通道），改走
# ``_snapshot_value`` —— 与 checkpoint / ``$PARENT`` / 子流程输入**同一个**
# 物化边界：容器按结构重建为精确类型（JSON 不可表达的非容器宿主对象按引用
# 共享，它们本就进不了 checkpoint；就地改写由 #69 门 default-deny 拒绝）。
global_context = _snapshot_value(flow.global_context) if flow.global_context else {}
global_context.update({"flow_id": flow.flow_id})
if self.parent is None:
# run 身份（plaita-nodes#9）：report 等「run 级结果通道」的缺省
# token 必须跨 run 唯一，否则不同 run 写同一个 default 文件。只在
# 根上下文写入；子流程（map/while/child）不复制，经
# get_global_variable 的 parent 回落共享同一个 run_token。
# flow.global_context 显式给出的值优先（setdefault）。
global_context.setdefault("run_token", self.execution_id)
# $PARENT is a snapshot (checkpoint-safe; not a live state)。
# C1-2（2026-10 评审修复包 C1）：历史上 ``dict(parent.context)`` 是浅
# 拷贝——``$PARENT.NODE`` 与父 live ``$NODE`` 是同一对象，双向写穿
# （子流程写 $PARENT 污染父运行态；父继续 update_node_result 让子侧
# "冻结"快照漂移）。
# 惰性按根键快照（_LazyRootSnapshot）：子流程 setup 在热路径上
# （while/for 每轮一次），全量深快照会重演 loop.py ML1 修掉的
# 「每轮全量 deepcopy」回归；从未读取的根键零成本。落盘/恢复经
# to_checkpoint_dict / from_checkpoint_dict 完全物化，惰性对象不出进程。
# plaita#72：快照在**子上下文构造时**就取好（``_parent_snapshot``），
# 这里只消费——setup 时点晚于子执行创建，后台分支（父不阻塞）会在
# 两者之间读到父的"未来态"。
self._state.setup_flow(
input_value=_coerce_input_value(args, kwargs),
parent_context=self._parent_snapshot if self.parent else {},
global_context=global_context,
flow_id=flow.flow_id,
env=env,
)
|