plaita.core.runner — NodeRunner for single-node execution.
Handles timeout enforcement (thread-based for sync, asyncio.wait_for for async),
retry logic, error strategy dispatch (abort/continue/continue-with), and
cooperative cancellation via threading.Event.
InvalidTimeoutError(value, reason: str, source: Optional[str] = None)
Bases: ValueError
非法 timeout 的具名拒绝（plaita#139）。
reason 不含值/形状前缀，供上层（distributed 校验层）补上"来源"上下文
后重新抛出——避免出现 Invalid timeout X from S: Invalid timeout X: ...
这类重复前缀。
源代码位于： plaita/core/runner.py
| def __init__(self, value, reason: str, source: Optional[str] = None):
self.value = value
self.reason = reason
self.source = source
where = f" from {source}" if source else ""
super().__init__(f"Invalid timeout {value!r}{where}: {reason}; {_TIMEOUT_FORMS}")
|
Handles single-node execution with timeout, retry, and error handling.
源代码位于： plaita/core/runner.py
| def __init__(self, context: ExecutionContext, node_execution=None) -> None:
self.context = context
self.node_execution = node_execution
|
async
run_node(flow, node, *, max_timeout_ms: Optional[int] = None, callback_manager: Optional[CallbackManager] = None) -> Tuple[Any, Optional[str]]
Execute a node and return (result, branch).
Triggers callback_manager.on_node_start/on_node_end.
Updates context with last_node_id, last_branch, and node result.
源代码位于： plaita/core/runner.py
| async def run_node(
self,
flow,
node,
*,
max_timeout_ms: Optional[int] = None,
callback_manager: Optional[CallbackManager] = None,
) -> Tuple[Any, Optional[str]]:
"""Execute a node and return (result, branch).
Triggers callback_manager.on_node_start/on_node_end.
Updates context with last_node_id, last_branch, and node result.
"""
if callback_manager:
callback_manager.on_node_start(flow, node)
try:
result = await self._execute_with_retry(flow, node, max_timeout_ms)
except BaseException as exc:
# plaita#25：abort 策略（以及取消/超时）是**直接 raise**，
# 历史上 on_node_end 永不触发 ⇒ 失败节点在 node_timings、事件时间线
# 与 Langfuse span 三处同时失踪（obs.py 模块 docstring 自认），
# 「失败节点恰缺失」的观测盲区由此而来。补发一次 error 形态的
# on_node_end：错误形状与 resume 失败路径（strategies.py）逐字同形。
# 回调分发器逐 handler 吞异常，故这里不会遮蔽原始错误。
# `except BaseException` 是为了把取消（CancelledError/
# FlowCancelledException）也收进时间线——它们同样是终态事件。
if callback_manager:
callback_manager.on_node_end(
flow, node, None, _node_error_payload(exc), exception=exc,
# 收口批①（裁决 B）：这条是 **abort/取消/超时** 的补发路径 ⇒ 用
# **路径标记**告诉带 span 的观测端「标注错误但保持 span 打开」。
# 直接错误钩子不带此标记 ⇒ 仍按既有契约终结 span。
keep_span_open=True,
)
raise
self.context.last_node_id = node.id
logger.debug("result: %s", result)
logger.debug("set last node id: %s", node.id)
branch = result if node.branching else None
self.context.last_branch = branch
self.context.update_node_result(node, result)
if callback_manager:
callback_manager.on_node_end(flow, node, result)
logger.debug("branch: %s", branch)
return result, branch
|
is_retryable_error(exc: BaseException) -> bool
按异常类型/错误码判定是否值得重试（node-F2）。
NodeException：1001（请求生成 = 配置/参数确定性错误）不重试；
1002/1003（发送/处理 = 瞬态）重试；其余未知码维持历史行为（重试），
避免把未来新增的错误码静默降级成「一次都不重试」。- 非
NodeException 的 ValueError：参数/配置类确定性错误，不重试；
但节点输出校验钩子抛出的异常除外（#219 契约：校验异常仍走 retryTimes）。
- 其余异常（
RuntimeError、节点自身 TimeoutError 等）：保持历史行为。
A50-03：公开导出。worker 层的重投判据必须复用同一份分类，否则
NodeExecutionError 的 abort 包装会把 1001/ValueError 重新判成可重试，
core 的「确定性不重试」在跨层边界丢失（同一错误被 worker 重投 5 次）。
源代码位于： plaita/core/runner.py
| def is_retryable_error(exc: BaseException) -> bool:
"""按异常类型/错误码判定是否值得重试（node-F2）。
- ``NodeException``：1001（请求生成 = 配置/参数确定性错误）不重试；
1002/1003（发送/处理 = 瞬态）重试；其余未知码维持历史行为（重试），
避免把未来新增的错误码静默降级成「一次都不重试」。
- 非 ``NodeException`` 的 ``ValueError``：参数/配置类确定性错误，不重试；
但节点输出校验钩子抛出的异常除外（#219 契约：校验异常仍走 retryTimes）。
- 其余异常（``RuntimeError``、节点自身 ``TimeoutError`` 等）：保持历史行为。
A50-03：公开导出。worker 层的重投判据必须复用同一份分类，否则
``NodeExecutionError`` 的 abort 包装会把 1001/ValueError 重新判成可重试，
core 的「确定性不重试」在跨层边界丢失（同一错误被 worker 重投 5 次）。
"""
if isinstance(exc, NodeException):
code = getattr(exc, "code", None)
if code in _RETRYABLE_NODE_CODES:
return True
if code in _NON_RETRYABLE_NODE_CODES:
return False
return True
if isinstance(exc, ValueError):
return is_output_validation_retryable(exc)
return True
|