plaita.storage.base

状态存储抽象：ExecutionState、ExecutionStorage、FlowStorage。生产路径后端为 memory / redis（同步契约）。sqlalchemy 模块仍存在但其方法为 async，与本 ABC 及 FlowWorker 不兼容，公开 factory/CLI 已拒绝用作 execution/flow 存储。

base

StateSerializationError

StateSerializationError(execution_id: str, cause: BaseException)

Bases: RuntimeError

执行状态不可序列化（写侧永久失败）——与「Redis 瞬断」严格区分。

为什么必须有这个类型（plaita#80）：RedisExecutionStorage.save_execution_state 原先用一个 except Exception -> return False 把两类性质完全不同的故障 混成同一返回值：

• 不可序列化：state.context 里出现了非 JSON 值。重试一万次也不会变好；
• Redis 瞬断：该退避后重投。

调用方（FlowWorker）把 False 一律当瞬态处理（不 ack、等重投），于是前者 表现为「每次落盘都失败 → 反复从 checkpoint 重跑同一节点 → 副作用重放 → 执行 永远拿不到终态，直到投递耗尽进 DLQ」。同一个 state 在 memory 后端恒 True （它不做 JSON 序列化），所以这个缺陷只在 Redis 后端显形、且会被任何 memory 后端写的单测掩盖。

与 ExecutionStateLoadError 对称（读侧/写侧各一支）。刻意不继承 ValueError：FlowWorker 把 ValueError 当「畸形消息」poison ack， 继承它会让本应终态化的永久失败被当成消息格式错误。

源代码位于： plaita/storage/base.py

def __init__(self, execution_id: str, cause: BaseException):
    super().__init__(
        f"执行状态不可序列化（永久失败，重试无用）：execution_id={execution_id}；"
        f"底层 {type(cause).__name__}: {cause}"
    )
    self.execution_id = execution_id
    self.cause = cause

ExecutionState

Bases: BaseModel

流程执行状态模型，用于表示流程执行的完整状态信息

属性

execution_id: 执行ID，唯一标识一次流程执行 flow_id: 流程ID，标识所属流程 flow_name: 流程名称，标识流程名称 flow_version: 流程版本号，标识流程定义版本 flow_hash: 启动时实际加载执行的 Flow 定义指纹（sha256，可选）。 resume 时与当前定义指纹比对，不一致则拒绝续跑（运行中发布新 版本会导致挂起执行走错分支）；None（老状态/老 worker 写入） 跳过校验，零回归。 context: 执行上下文，存储流程执行过程中的所有状态数据 status: 当前状态，包括 running(运行中)、suspended(挂起)、completed(完成)、error(错误) start_time: 流程开始时间，ISO格式字符串 last_update_time: 最后更新时间，ISO格式字符串 end_time: 流程结束时间，ISO格式字符串 error: 错误信息，当status为error时存储错误详情 invoker: 调用者，标识发起流程的实体

ExecutionStorage

Bases: ABC

状态存储基类，定义了所有状态存储必须实现的接口

save_execution_state abstractmethod

save_execution_state(execution_id: str, state: ExecutionState) -> bool

保存流程执行状态

参数：

名称	类型	描述	默认
execution_id	str	执行ID	必需
state	ExecutionState	执行状态数据	必需

返回：

名称	类型	描述
bool	bool	是否保存成功

源代码位于： plaita/storage/base.py

@abstractmethod
def save_execution_state(self, execution_id: str, state: ExecutionState) -> bool:
    """
    保存流程执行状态

    Args:
        execution_id: 执行ID
        state: 执行状态数据

    Returns:
        bool: 是否保存成功
    """
    pass

load_execution_state abstractmethod

load_execution_state(execution_id: str) -> Optional[ExecutionState]

加载流程执行状态

参数：

名称	类型	描述	默认
execution_id	str	执行ID	必需

返回：

类型	描述
Optional[ExecutionState]	ExecutionState or None: 执行状态数据，如果不存在则返回None

源代码位于： plaita/storage/base.py

@abstractmethod
def load_execution_state(self, execution_id: str) -> Optional[ExecutionState]:
    """
    加载流程执行状态

    Args:
        execution_id: 执行ID

    Returns:
        ExecutionState or None: 执行状态数据，如果不存在则返回None
    """
    pass

delete_execution_state abstractmethod

delete_execution_state(execution_id: str) -> bool

删除流程执行状态

参数：

名称	类型	描述	默认
execution_id	str	执行ID	必需

返回：

名称	类型	描述
bool	bool	是否删除成功

源代码位于： plaita/storage/base.py

@abstractmethod
def delete_execution_state(self, execution_id: str) -> bool:
    """
    删除流程执行状态

    Args:
        execution_id: 执行ID

    Returns:
        bool: 是否删除成功
    """
    pass

list_executions abstractmethod

list_executions(query: Optional[Any] = None, order_by: Optional[str] = None, limit: int = 100, offset: int = 0) -> List[ExecutionState]

列出执行ID

参数：

名称	类型	描述	默认
query	Optional[Any]	查询条件	None
order_by	Optional[str]	排序字段	None
limit	int	最大返回数量	100
offset	int	偏移量	0

返回：

类型	描述
List[ExecutionState]	List[ExecutionState]: 执行状态列表

源代码位于： plaita/storage/base.py

@abstractmethod
def list_executions(self, query: Optional[Any] = None, order_by: Optional[str] = None, limit: int = 100, offset: int = 0) -> List[ExecutionState]:
    """
    列出执行ID

    Args:
        query: 查询条件
        order_by: 排序字段
        limit: 最大返回数量
        offset: 偏移量

    Returns:
        List[ExecutionState]: 执行状态列表
    """
    pass

load_execution_tombstone

load_execution_tombstone(execution_id: str) -> Optional[Dict[str, Any]]

读取执行删除墓碑（A51-01）；无墓碑/后端不支持时返回 None。

删除是持久语义：状态行被删掉之后，任何仍在队列/PEL 里的 start 载体都会让 start_flow 走「盘上无状态 = 全新执行」分支，把已删 执行重建并重放旧输入。墓碑在删除时与状态删除同原子落盘，worker 在启动/重投/守卫克隆前复核；键位在平台根（不随租户命名空间清理）， 值至少含 tenant，调用方按消息体租户匹配，避免跨租户同 id 误伤。

返回 {"tenant": str, "deleted_at": str, "reason": str, ...}； 读取/反序列化失败应抛 ExecutionStateLoadError（瞬态不得当 「无墓碑」放行重建）。默认实现返回 None，保持第三方/内存后零回归。

源代码位于： plaita/storage/base.py

def load_execution_tombstone(self, execution_id: str) -> Optional[Dict[str, Any]]:
    """读取执行删除墓碑（A51-01）；无墓碑/后端不支持时返回 None。

    删除是持久语义：状态行被删掉之后，任何仍在队列/PEL 里的 ``start``
    载体都会让 ``start_flow`` 走「盘上无状态 = 全新执行」分支，把已删
    执行重建并重放旧输入。墓碑在删除时与状态删除**同原子**落盘，worker
    在启动/重投/守卫克隆前复核；键位在平台根（不随租户命名空间清理），
    值至少含 ``tenant``，调用方按消息体租户匹配，避免跨租户同 id 误伤。

    返回 ``{"tenant": str, "deleted_at": str, "reason": str, ...}``；
    读取/反序列化失败应抛 ``ExecutionStateLoadError``（瞬态不得当
    「无墓碑」放行重建）。默认实现返回 None，保持第三方/内存后零回归。
    """
    return None

serialize_state

serialize_state(state: Dict[str, Any]) -> str

序列化状态数据

参数：

名称	类型	描述	默认
state	Dict[str, Any]	状态数据	必需

返回：

名称	类型	描述
str	str	序列化后的字符串

源代码位于： plaita/storage/base.py

def serialize_state(self, state: Dict[str, Any]) -> str:
    """
    序列化状态数据

    Args:
        state: 状态数据

    Returns:
        str: 序列化后的字符串
    """
    try:
        return json.dumps(state)
    except Exception as e:
        logger.error("Failed to serialize state: %s", _scrub_url_text(e))
        raise

deserialize_state

deserialize_state(data: str) -> Dict[str, Any]

反序列化状态数据

参数：

名称	类型	描述	默认
data	str	序列化的状态数据	必需

返回：

名称	类型	描述
Dict	Dict[str, Any]	反序列化后的状态数据

源代码位于： plaita/storage/base.py

def deserialize_state(self, data: str) -> Dict[str, Any]:
    """
    反序列化状态数据

    Args:
        data: 序列化的状态数据

    Returns:
        Dict: 反序列化后的状态数据
    """
    try:
        return json.loads(data)
    except Exception as e:
        logger.error("Failed to deserialize state: %s", _scrub_url_text(e))
        raise 

FlowStorage

Bases: ABC

get_flow abstractmethod

get_flow(flow_id: str, version: Optional[str] = None) -> Optional[Dict[str, Any]]

获取流程定义

参数：

名称	类型	描述	默认
flow_id	str	流程ID	必需
version	Optional[str]	流程版本号	None

返回：

类型	描述
Optional[Dict[str, Any]]	Dict or None: 流程定义数据，如果不存在则返回None

源代码位于： plaita/storage/base.py

@abstractmethod
def get_flow(self, flow_id: str, version: Optional[str] = None) -> Optional[Dict[str, Any]]:
    """
    获取流程定义

    Args:
        flow_id: 流程ID
        version: 流程版本号

    Returns:
        Dict or None: 流程定义数据，如果不存在则返回None
    """
    pass

save_flow abstractmethod

save_flow(flow: Dict[str, Any]) -> bool

保存流程定义

源代码位于： plaita/storage/base.py

@abstractmethod
def save_flow(self, flow: Dict[str, Any]) -> bool:
    """
    保存流程定义
    """
    pass