`plaita.core.errors`¶

规范异常类型与错误处理策略。

基类 / 业务错误：FlowExecutionException、FlowResultError、NodeException、FlowErrorType、ResumeType。

细分异常（0.5.x）：NodeNotFoundError、FlowStartMissingError、NodeExecutionError、NodeTimeoutError、FlowTimeoutError、FlowErrorException、ErrorResultException、ResumeError。

Worker 侧（plaita.server.flow_worker，非 core）：ResumeProtocolError——挂起守卫类 ResumeError（如 continue 打在 pending 挂起节点上）不终态化、执行保持原状时上抛（#33）。

错误策略：ErrorStrategy、ErrorHandler、RecoverableErrorHandler。

errors ¶

plaita.core.errors — Canonical location for Plaita error classes.

Migrated from plaita/errors.py (which is now a compatibility shim).

FlowResultError ¶

FlowResultError(code=None, message=None)

Bases: RuntimeError

流程执行结果异常，用于在End节点中抛出

源代码位于： plaita/core/errors.py

def __init__(self, code=None, message=None):
    super().__init__(f"FlowResultError: code={code}, message={message}")
    self.code = code
    self.message = message

NodeException ¶

NodeException(code, message)

Bases: RuntimeError

节点执行异常,在节点的Execution方法中抛出

源代码位于： plaita/core/errors.py

def __init__(self, code, message):
    super().__init__(f"NodeException: code={code}, message={message}")
    self.code = code
    self.message = message

FlowErrorType ¶

Bases: Enum

流程异常类型

ResumeType ¶

Bases: Enum

分布式恢复语义类型, 取代裸字符串 magic string。

CONTINUE 为默认 (带 saved_context 直接续跑); RETRY 专对 error 态 execution 放行 (worker 终态短路例外), 策略层与 continue 同一条步进路径—— 失败节点在 checkpoint 里无条目, 从 last_node_id 后继步进即恰好重跑失败节点 (G1); 其余三种仅对挂起的 EventNode 生效 (由 DistributedStrategy._handle_resume 处理)。

coerce `classmethod` ¶

coerce(value) -> ResumeType

接受 enum 或字符串 (兼容外部旧调用方), 无法识别时抛 ResumeError。

源代码位于： plaita/core/errors.py

@classmethod
def coerce(cls, value) -> "ResumeType":
    """接受 enum 或字符串 (兼容外部旧调用方), 无法识别时抛 ``ResumeError``。"""
    if isinstance(value, cls):
        return value
    if isinstance(value, str):
        try:
            return cls(value)
        except ValueError:
            raise ResumeError(f"Unsupported resume type: {value!r}")
    raise ResumeError(f"Unsupported resume type: {value!r}")

FlowExecutionException ¶

FlowExecutionException(code: int = -500, message: str = '', error_type=None, node=None, source_line: Optional[int] = None)

Bases: RuntimeError

流程执行异常基类。

保留 (code, message, error_type, node) 位置签名以兼容历史调用方。新代码应优先抛出具体子类（NodeNotFoundError / NodeExecutionError / FlowErrorException 等），它们自带默认 code / error_type，调用方只需提供 message 与 node，不必再往调用点塞魔法数字。

源代码位于： plaita/core/errors.py

def __init__(self, code: int = -500, message: str = "", error_type=None, node=None,
             source_line: Optional[int] = None):
    self.code = code
    self.message = message
    self.error_type = error_type if error_type is not None else type(self).error_type
    self.node = node
    # 运行期错误回标的源码行号（仅 @flow 前端编译期写入 IR 时可用）。
    # 调用方可显式传入；未传时由 runner 从 node.source_line 兜底回填。
    self.source_line = source_line
    super().__init__(message)

NodeNotFoundError ¶

NodeNotFoundError(node_id=None, message=None, node=None)

Bases: FlowExecutionException

节点找不到（含流程缺少 start 节点）。

源代码位于： plaita/core/errors.py

def __init__(self, node_id=None, message=None, node=None):
    if message is None:
        message = f"Node with id '{node_id}' not found" if node_id else "Node not found"
    super().__init__(self.code, message, self.error_type, node)

FlowStartMissingError ¶

FlowStartMissingError(message='Flow has no start node', node=None)

Bases: NodeNotFoundError

流程缺少起始节点。共享 NODE_NOT_FOUND 语义。

源代码位于： plaita/core/errors.py

def __init__(self, message="Flow has no start node", node=None):
    super().__init__(message=message, node=node)

NodeExecutionError ¶

NodeExecutionError(message: str, node=None, code: int = DEFAULT_NODE_ABORT_CODE)

Bases: FlowExecutionException

节点执行出错（abort 策略）。

code 默认取 DEFAULT_NODE_ABORT_CODE；当调用方配了 error_handler 时由 runner 传入该 handler 的 error_code。

源代码位于： plaita/core/errors.py

def __init__(self, message: str, node=None, code: int = DEFAULT_NODE_ABORT_CODE):
    super().__init__(code, message, self.error_type, node)

NodeTimeoutError ¶

NodeTimeoutError(message: str, node=None, error_type=NODE_ERROR)

Bases: FlowExecutionException

节点超时。error_type 随超时来源切换：节点自身超时为 NODE_ERROR，流程级超时强加给节点时为 FLOW_ERROR。

源代码位于： plaita/core/errors.py

def __init__(self, message: str, node=None, error_type=FlowErrorType.NODE_ERROR):
    super().__init__(self.code, message, error_type, node)

FlowTimeoutError ¶

FlowTimeoutError(message: str = 'Flow execution timeout', node=None)

Bases: FlowExecutionException

流程整体超时。

源代码位于： plaita/core/errors.py

def __init__(self, message: str = "Flow execution timeout", node=None):
    super().__init__(self.code, message, self.error_type, node)

ExecutionDeadlineError ¶

ExecutionDeadlineError(message: str = 'Flow execution timeout', node=None)

Bases: FlowTimeoutError

执行级总预算（deadline）耗尽（plaita#267）。

与 :class:FlowTimeoutError 同 code=-1 / FLOW_ERROR 归属，既有消费方（normal 模式错误契约、worker 的「超时不重试」判据）零回归；单独成类是为了让执行总预算到期与节点自身超时可判别：

worker _is_retryable_node_failure 按 FlowTimeoutError 家族判定「不重试」，执行终态化 error 可被收尸（不会再重投一次）；
观测端可按类型名区分「执行 deadline 到期」与「节点 timeoutHandler 超时」。

源代码位于： plaita/core/errors.py

def __init__(self, message: str = "Flow execution timeout", node=None):
    super().__init__(self.code, message, self.error_type, node)

FlowErrorException ¶

FlowErrorException(message: str, node=None)

Bases: FlowExecutionException

流程执行通用错误（兜底包装）。

源代码位于： plaita/core/errors.py

def __init__(self, message: str, node=None):
    super().__init__(self.code, message, self.error_type, node)

FlowCancelledException ¶

FlowCancelledException(message: str = 'Flow execution cancelled', node=None)

Bases: FlowExecutionException

执行已被取消（2026-10 波次③）。

由 FlowExecution.cancel() 置位 cancel_requested 后，引擎在节点入口抛出本异常拒绝继续执行。职责边界：

不经节点 errorHandler 分发（不重试、不被 continue 策略吞掉）—— 取消是执行级意图，不是节点错误；在 runner 重试循环之前抛出，沿 FlowExecutionException 同族既有语义冒泡（normal 模式原样透传， distributed 模式按既有契约归一化为 FlowErrorException/-500）。
code=-4 为取消专属错误码（与 -500 通用/-520 节点 abort/-1 超时区分），供上层 worker 识别取消终态。

源代码位于： plaita/core/errors.py

def __init__(self, message: str = "Flow execution cancelled", node=None):
    super().__init__(self.code, message, self.error_type, node)

ErrorResultException ¶

ErrorResultException(code: int, message: str, node=None)

Bases: FlowExecutionException

End 节点返回错误结果。code/message 来自 FlowResultError。

源代码位于： plaita/core/errors.py

def __init__(self, code: int, message: str, node=None):
    super().__init__(code, message, self.error_type, node)

ResumeError ¶

ResumeError(message: str, node=None)

Bases: FlowExecutionException

分布式恢复阶段的不合法状态（无挂起节点 / 节点非 EventNode / 状态不对 / resume_type 不支持 / 恢复执行抛错）。

源代码位于： plaita/core/errors.py

def __init__(self, message: str, node=None):
    super().__init__(self.code, message, self.error_type, node)

ResumeGuardError ¶

ResumeGuardError(message: str, node=None)

Bases: ResumeError

恢复守卫：恢复请求与执行状态不匹配（策略层在分发到节点 resume() 之前就拒绝），执行应保持原状而不是被终态化。

与基类的分工（worker 侧 _chain_has_resume_guard_error 靠本类型区分）：

本类（守卫）：continue/retry 试图绕过 pending 挂起节点、resume 打在无挂起节点 / 非挂起节点的 checkpoint 上。这类请求不携带推进语义，且几乎都来自 at-least-once 重投与竞速（重复投递的 event resume 打在已推进的 checkpoint 上同族）——执行保持 suspended 等真正的决议路径（event/cancel/timeout），终态化成 error 会把可恢复的挂起执行永久打封（连 event 都被终态短路拒绝，plaita#33）。
其余 ResumeError（含 strategies._handle_resume 把 current_node.resume() 抛出的异常包成的那个）：是执行自身失败，由 worker 现状路径终态化 error（可观测、可人工 retry），不得静默保持原状。

源代码位于： plaita/core/errors.py

def __init__(self, message: str, node=None):
    super().__init__(message, node)

StateSnapshotError ¶

StateSnapshotError(code: int = -500, message: str = '', error_type=None, node=None, source_line: Optional[int] = None)

Bases: FlowExecutionException

状态快照出口无法安全物化（plaita#107）。

CheckpointState.to_checkpoint_dict / update_node_result 的出口走 plaita.core.state._snapshot_value：它现在是显式栈的迭代物化，Python 递归深度与数据深度解耦，容器环按占位符安全截断。唯一仍会抛出的情形是数据深度超过防御性上限——此时抛本异常而不是让 RecursionError 穿透到分布式归一化层（后者会把真实原因抹成误导性的 FlowErrorException("maximum recursion depth exceeded")）。消息携带超限深度与容器路径，normal 模式原样透传、distributed 归一化后仍保留该文本。

源代码位于： plaita/core/errors.py

def __init__(self, code: int = -500, message: str = "", error_type=None, node=None,
             source_line: Optional[int] = None):
    self.code = code
    self.message = message
    self.error_type = error_type if error_type is not None else type(self).error_type
    self.node = node
    # 运行期错误回标的源码行号（仅 @flow 前端编译期写入 IR 时可用）。
    # 调用方可显式传入；未传时由 runner 从 node.source_line 兜底回填。
    self.source_line = source_line
    super().__init__(message)

ErrorStrategy ¶

Bases: Enum

错误处理策略 - abort: 结束流程，返回指定code及message - continue: 继续执行,但本节点无返回值 - continue-with: 继续执行,并返回指定的值

ErrorHandler ¶

Bases: BaseModel

错误处理器

strategy 字段是 ErrorStrategy enum (2026-07 重构前是 Optional[str]，比较时全程依赖 _strategy_eq 这个 wart 同时识别 enum 与字符串)。Pydantic 会自动处理 enum ↔ "abort"/"continue"/"continue-with" 字符串的双向序列化， JSON 输入仍写字符串即可。历史 continue_with (下划线) 别名在 validator 里归一化为规范连字符。

handle ¶

handle()

处理超时异常 - abort: 抛出TimeoutError异常 - continue: 返回None - continue-with: 返回默认值

源代码位于： plaita/core/errors.py

def handle(self):
    """
    处理超时异常
    - abort: 抛出TimeoutError异常
    - continue: 返回None
    - continue-with: 返回默认值
    """
    if self.strategy == ErrorStrategy.ABORT:
        message = self.error_message or "Node execution timed out"
        raise TimeoutError(message)
    elif self.strategy == ErrorStrategy.CONTINUE:
        return None
    elif self.strategy == ErrorStrategy.CONTINUE_WITH:
        return self.default_value

ExecutionLeaseError ¶

Bases: RuntimeError

执行租约失守（resume 抢占失败 / fence 世代失配）。

定义下沉到 core（plaita#96）：storage（fenced CAS 写拒绝旧世代）与 server（execution_lease / flow_worker）两侧必须共用同一类对象—— 定义留在 server 曾迫使 storage 相对导入 ..server，构成被分层禁止的反向依赖。server 侧 execution_lease.py 复用本定义。

mark_output_validation_retryable ¶

mark_output_validation_retryable(exc: BaseException) -> None

给输出校验钩子抛出的异常打上可重试标记。

打标而不是换异常类型：用户日志与既有测试都按原异常类型/文案断言（ValueError: …）。BaseException 实例始终有 __dict__（即使子类声明 __slots__），setattr 不会失败。

源代码位于： plaita/core/errors.py

def mark_output_validation_retryable(exc: BaseException) -> None:
    """给输出校验钩子抛出的异常打上可重试标记。

    打标而不是换异常类型：用户日志与既有测试都按原异常类型/文案断言
    （``ValueError: …``）。``BaseException`` 实例始终有 ``__dict__``（即使
    子类声明 ``__slots__``），``setattr`` 不会失败。
    """
    setattr(exc, _OUTPUT_VALIDATION_RETRY_ATTR, True)

is_output_validation_retryable ¶

is_output_validation_retryable(exc: BaseException) -> bool

该异常是否来自节点输出校验钩子（见 :func:mark_output_validation_retryable）。

源代码位于： plaita/core/errors.py

def is_output_validation_retryable(exc: BaseException) -> bool:
    """该异常是否来自节点输出校验钩子（见 :func:`mark_output_validation_retryable`）。"""
    return bool(getattr(exc, _OUTPUT_VALIDATION_RETRY_ATTR, False))

plaita.core.errors¶

errors ¶

FlowResultError ¶

NodeException ¶

FlowErrorType ¶

ResumeType ¶

coerce classmethod ¶

FlowExecutionException ¶

NodeNotFoundError ¶

FlowStartMissingError ¶

NodeExecutionError ¶

NodeTimeoutError ¶

FlowTimeoutError ¶

ExecutionDeadlineError ¶

FlowErrorException ¶

FlowCancelledException ¶

ErrorResultException ¶

ResumeError ¶

ResumeGuardError ¶

StateSnapshotError ¶

ErrorStrategy ¶

ErrorHandler ¶

handle ¶

ExecutionLeaseError ¶

mark_output_validation_retryable ¶

is_output_validation_retryable ¶

`plaita.core.errors`¶

coerce `classmethod` ¶