Struct Orchestrator 

pub struct Orchestrator<B: SagaBackend> { /* private fields */ }

业务作用：Saga Orchestrator——所有推进都以数据库 CAS 与单一本地事务落库。

无进程内可变业务状态（仅 fencing token 发行器）：多副本并发安全完全由 store 的 CAS、唯一键与 fencing token 承担，任何副本崩溃后另一副本可无缝接管。

Implementations

impl<B> Orchestrator<B>

where B: SagaBackend + SagaBackendFactory,

pub fn new( registry: DefinitionRegistry, config: OrchestratorConfig, ) -> Result<Self>

业务作用：构造 Orchestrator。

参数说明：

• registry: 启动阶段构建完成的 definition 注册表（运行期只读）。
• config: 运行参数。

返回：配置预算全部为正时返回 Orchestrator；零/负预算返回错误并拒绝启动。

pub fn with_datasource( registry: DefinitionRegistry, config: OrchestratorConfig, datasource: impl AsRef<str>, ) -> Result<Self>

业务作用：构造绑定命名 datasource 的 Orchestrator，使状态、Inbox、Outbox、timer 与审计共享一次本地提交。

参数说明：

• registry: 启动期已冻结的 definition 注册表。
• config: Orchestrator 运行预算与租户门禁。
• datasource: 所有 Saga 本地持久与事务边界共用的数据源。

返回：配置与 datasource 合法时返回同源 Orchestrator；否则在任何数据库 I/O 前拒绝。

pub fn datasource_ref(&self) -> &DatasourceRef

业务作用：读取 Orchestrator 全部本地持久与事务链绑定的 datasource 身份。

参数说明: 无。

返回：store、Inbox、Outbox 与 timer 共用的不可变 qualifier。

pub fn replace_registry(&self, registry: DefinitionRegistry)

业务作用：原子发布已经完成 Catalog、历史实例与 capability 门禁的新 registry generation。

参数说明：registry 同时包含 active definition 和旧实例仍需的 deprecated definition。

返回：新快照成为后续操作唯一入口后完成；进行中的操作继续持有旧快照直到事务结束。

pub async fn verify_registry_snapshot( &self, registry: &DefinitionRegistry, ) -> Result<()>

业务作用：在动态 Catalog 快照发布前确认全部非终态实例仍能找到相同摘要的 definition。

参数说明：registry 是 watcher 刚从共享 generation 完整装载的候选快照。

返回：所有在途实例均可安全继续时成功；缺失或摘要漂移时拒绝切换。

pub async fn verify_startup(&self) -> Result<()>

业务作用：启动预检——非终态实例引用的 definition 必须可用且摘要一致，否则拒绝 Ready。

参数说明: 无。

返回：预检通过返回 Ok；缺失或摘要漂移返回错误，调用方不得宣告 Ready。

pub fn definition_digest( &self, workflow: &WorkflowName, version: DefinitionVersion, ) -> Option<String>

业务作用：读取当前 registry 快照中指定 definition 的 canonical 摘要，供协议兼容门禁在创建前校验。

参数说明：workflow 与 version 共同定位不可变流程版本。

返回：当前快照已注册时返回摘要；未知或未激活版本返回空。

pub fn definition_digest_for_tenant( &self, tenant: &TenantId, workflow: &WorkflowName, version: DefinitionVersion, ) -> Option<String>

业务作用：按租户读取当前可用于运行期复验的 definition 摘要。

参数说明：tenant、workflow 与 version 共同定位租户专属流程版本。

返回：active、deprecated 或静态全局定义存在时返回摘要；未知版本为空。

pub async fn load_instance( &self, tenant: &TenantId, saga_id: &SagaId, ) -> Result<Option<SagaInstanceRow>>

业务作用：为管理面读取指定租户拥有的 Saga 快照，阻止仅凭 saga_id 跨租户枚举。

参数说明：

• tenant: 已认证管理主体被授权访问的租户。
• saga_id: 实例身份。

返回：实例存在且属于该租户时返回快照；不存在或属于其他租户均返回 None，防止 saga_id 成为跨租户存在性 oracle。

pub async fn load_audit_trail( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, after_transition_seq: u64, after_control_seq: u64, limit: u32, ) -> Result<SagaAuditTrail>

业务作用：按租户和 saga.audit.read 权限读取一份有界、可恢复的完整审计视图。

参数说明：

• management: 已认证主体与权限快照；reason 用于宿主管理访问日志。
• tenant: 主体获授权访问的租户。
• saga_id: 目标实例。
• after_transition_seq: 业务迁移分页游标。
• after_control_seq: 控制迁移分页游标。
• limit: 每类事实最大返回数，范围 1..=1000。

返回：实例存在且属于租户时返回 attempt/transition/control/management/conflict 聚合视图；无权限、跨租户、上限非法或持久层失败返回错误。

pub async fn load_audit_page( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, cursor: Option<&SagaAuditPageCursor>, limit: u32, ) -> Result<SagaAuditPage>

业务作用：按数据库全局序号读取不可变审计事件，使运行中新增事实和 attempt 终态变化 都能出现在既有游标之后。

参数说明：

• management: 已认证主体与审计读取权限。
• tenant: 主体获授权访问的租户。
• saga_id: 目标实例。
• cursor: 上一页末项的全局序号；None 表示从事件流起点开始。
• limit: 单页最大事实数，范围 1..=1000。

返回：按 audit_seq 返回至多 limit 条事实；非空页同时返回末项 checkpoint， 供历史遍历或后续追加事实续读。越权、跨租户或持久化失败返回错误。

pub async fn list_instances( &self, management: &SagaManagementContext, query: &SagaInstanceQuery<'_>, ) -> Result<Vec<SagaInstanceSummary>>

业务作用：租户受限的实例只读检索——按状态与创建时间窗过滤、keyset 分页，让运维 无需直连数据库即可定位待处置对象（如全部 MANUAL_INTERVENTION 实例）。

检索是纯读动作，使用与写动作分离的 saga.instance.list 权限；响应只含身份、 状态、当前步骤、版本、稳定原因码与时间戳，不携带业务 payload。租户过滤由查询 条件强制，不同租户实例的存在性不经本入口泄漏。

参数说明：

• management: 已认证主体与权限快照。
• query: 租户、状态集合、时间窗、cursor 与页大小。

返回：无时间条件时按 saga_id 升序；有时间条件时按创建时刻、saga_id 升序；无权限、 参数非法或持久层失败返回错误。

pub async fn tenant_quota_usage( &self, management: &SagaManagementContext, tenant: &TenantId, ) -> Result<u64>

业务作用：读取单个租户的精确在飞配额用量——只经受鉴权管理查询返回,不进指标标签。

参数说明：

• management: 已认证主体与权限快照（要求 saga.audit.read）。
• tenant: 目标租户。

返回：账本记录的在飞实例数;无权限返回错误。

pub async fn tenant_action_rate_usage( &self, management: &SagaManagementContext, tenant: &TenantId, ) -> Result<Option<(u64, u64)>>

业务作用：读取单个租户当前窗口的管理动作用量——只经受鉴权管理查询返回,不进指标标签。

参数说明：

• management: 已认证主体与权限快照（要求 saga.audit.read）。
• tenant: 目标租户。

返回：Some((窗口起点毫秒, 已用动作数));该租户未配置速率时返回 None。

pub async fn load_operational_metrics( &self, now_ms: i64, ) -> Result<SagaOperationalMetrics>

业务作用：读取一份不含业务身份标签的 Saga 运行指标快照。

生命周期与状态计数来自当前 Saga 后端的已提交事实；Kafka 处理指标来自当前进程。 宿主应将该入口放在受保护的运维 /metrics 端口，不要与业务 API 公网暴露。

参数说明：

• now_ms: 当前 epoch 毫秒，用于统计已到期 durable timer。

返回：数据库聚合成功时返回可渲染 Prometheus 文本的快照；连接或查询失败返回错误。

pub fn verify_result_producer( &self, envelope: &SagaResultEnvelope, producer: &ServiceIdentity, ) -> Result<()>

业务作用：在占用 Orchestrator Inbox 身份前，把 transport 认证出的 producer 与 definition 登记的步骤 owner 绑定，阻止共享 result topic 上的越权作证。

该校验只接受 transport 信任策略给出的逻辑身份；envelope 自报字段会先完成派生复验， 随后再定位固定版本的 definition。失败时调用方必须隔离消息，不得进入 Inbox claim， 否则攻击者可以用合法 event id 抢先污染真实 owner 的去重身份。

参数说明：

• envelope: transport 已解码但尚未进入持久化路径的结果 envelope。
• producer: mTLS/SASL principal 或独占 topic 经信任策略映射出的逻辑服务身份。

返回：producer 与步骤 owner 精确一致时返回成功；身份、definition、步骤或 owner 任一不一致时返回协议错误，且没有任何数据库副作用。

pub async fn handle_authenticated_result( &self, envelope: &SagaResultEnvelope, producer: &ServiceIdentity, now_ms: i64, ) -> Result<HandleOutcome>

业务作用：消费一条已经获得 transport producer 身份的结果，并在认证通过后推进 Saga。

参数说明：

• envelope: 参与方结果 envelope。
• producer: transport 信任策略解析出的逻辑服务身份。
• now_ms: 当前 epoch 毫秒。

返回：认证和本地事务均成功时返回可 ACK 结论；认证失败无持久化副作用，推进失败则 事务回滚，两者都返回错误供 transport 选择 DLT 或保留 offset 重投。

pub async fn handle_authenticated_result_traced( &self, envelope: &SagaResultEnvelope, producer: &ServiceIdentity, receipt_trace: Option<&TraceContext>, now_ms: i64, ) -> Result<HandleOutcome>

业务作用：携带 transport 收据链路上下文消费结果——参与方到编排端的 trace 由此续接。

receipt_trace 必须来自 transport 收据（如 Kafka 消息 header）的显式解析，框架不 读取 ambient 状态。推进真实生效时，该上下文在同一事务内写入实例作为最新因果上下文， 本事务内发出的下一条命令与之后的 timer/恢复命令都从它派生 child span。

参数说明：

• envelope: 参与方结果 envelope。
• producer: transport 信任策略解析出的逻辑服务身份。
• receipt_trace: 收据中已校验的链路上下文；None 表示收据未携带。
• now_ms: 当前 epoch 毫秒。

返回：语义与 handle_authenticated_result 完全一致；trace 不改变认证、去重与推进边界。

pub async fn handle_authenticated_result_authorized_traced( &self, envelope: &SagaResultEnvelope, producer: &ServiceIdentity, receipt_trace: Option<&TraceContext>, now_ms: i64, authorize: &(dyn Fn() -> Result<()> + Send + Sync), ) -> Result<HandleOutcome>

业务作用：在宿主冻结的资格下接收已认证结果，数据库等待期间失权时回滚全部推进事实。

参数说明：envelope 是原结果，producer 是认证身份，receipt_trace 是可选受信链路， now_ms 是裁决时刻，authorize 同步复验同一次操作的期限、撤销代际和安全合同。

返回：资格持续有效且事务提交后才返回应用或重复收据；失权回滚，COMMIT 已发出后的不确定性保持原分类。 加入已有同源事务时，调用方仍须在外层最终提交前复验自己的执行资格。

pub async fn start_saga( &self, request: &StartSagaRequest<'_>, ) -> Result<StartOutcome>

业务作用：以受保护事务幂等创建 Saga 并发布首步命令。

创建事务同时提交：version/transition_seq = 1 的 NONE -> RUNNING、步骤骨架、 首步 execute 命令 Outbox、step 超时 timer 与（可选）实例 deadline timer。 业务幂等键冲突时返回既有实例，不产生任何新副作用。

参数说明：

• request: 创建请求。

返回：真实创建返回 StartOutcome::Started；幂等命中返回 AlreadyExists； 确定性拒绝可下转为 StartSagaError；事务失败保留回滚、提交结果不明或回滚失败的封闭分类。

pub async fn start_saga_traced( &self, request: &StartSagaRequest<'_>, trace: Option<&TraceContext>, ) -> Result<StartOutcome>

业务作用：携带显式链路上下文创建 Saga——发起入口的调用链由此接入 Saga 全程追踪。

框架不读取线程或任务 ambient 状态：调用方（Web 入口、调度任务）显式传入已校验的 TraceContext，其 canonical traceparent 随实例同事务持久化；首步命令与之后 每一条自动命令（含 timer、崩溃恢复发出的）都从已提交实例读取同一 trace 并派生 新 child span。缺少上下文时传 None，实例照常创建与投递。

参数说明：

• request: 创建请求。
• trace: 发起入口已校验的链路上下文；None 表示调用链未提供。

返回：语义与 start_saga 完全一致；trace 不改变创建、幂等 与拒绝边界。

pub async fn start_saga_authorized_traced( &self, request: &StartSagaRequest<'_>, trace: Option<&TraceContext>, authorize: &(dyn Fn() -> Result<()> + Send + Sync), ) -> Result<StartOutcome>

业务作用：在调用方冻结的执行资格下裁决 Saga 创建与重复收据，连接池、行锁和事务等待不能延长权限。

参数说明：request 是创建事实，trace 是受信链路上下文，authorize 同步复验本次操作的资格与期限。

返回：资格有效且创建事务提交后返回创建或重复收据；失权走事务回滚，底层提交不确定分类保持不变。 deprecated 定义仍按持久请求摘要裁决重复与冲突；只有真正的新建适用 active 门禁。 此门禁覆盖本次创建操作；加入已有同源事务时，外层事务还须在其最终提交前复验自己的执行资格。

pub async fn run_due_timers(&self, owner: &str, now_ms: i64) -> Result<u32>

业务作用：领取到期 timer 并逐个裁决，多副本可并发竞争。

必须在 ambient 事务之外调用：租约领取独立提交；每个 timer 的裁决各自开启 单一本地事务。实例处于 PAUSED 时不消费，按轮询退避交还（只是轮询节奏， 不顺延业务 deadline）。

参数说明：

• owner: 本副本稳定标识。
• now_ms: 当前时刻（epoch 毫秒）。

返回：本轮实际完成业务裁决的 timer 数；领取或裁决基础设施失败返回错误 （已领取未处理的 timer 由租约到期自动回收）。

pub async fn fire_timer( &self, timer: &SagaTimerRow, token: &TimerFencingToken, now_ms: i64, ) -> Result<TimerOutcome>

业务作用：裁决单个已领取 timer；消费与它触发的状态迁移同一事务。

参数说明：

• timer: 已领取的 timer 行。
• token: 本轮 fencing token。
• now_ms: 当前时刻（epoch 毫秒）。

返回：完成裁决返回 TimerOutcome::Applied；暂停/失权/语义过时返回对应跳过 结论；CAS 冲突返回错误（事务回滚，timer 留在租约内等待重试或回收）。

pub async fn pause( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, operation_id: &str, ) -> Result<()>

业务作用：管理面暂停实例——只停止自动动作，不改业务状态、不停业务时钟。

参数说明：

• management: 已认证主体、原因与 saga.pause 权限快照。
• tenant: 已认证管理主体被授权访问的租户。
• saga_id: 实例身份。
• operation_id: 稳定管理 operation id；与控制态切换同事务写入本地幂等审计链。

返回：暂停生效返回 Ok；实例版本或控制状态过期返回错误，调用方重读后重试。

pub async fn pause_with_expectation( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, operation_id: &str, expectation: SagaManagementExpectation, ) -> Result<()>

业务作用：在事务内复验调用方版本后暂停实例，避免旧管理快照覆盖并发状态或控制迁移。

参数说明：管理上下文、租户、实例和操作身份定位动作，expectation 约束事务内实例版本。

返回：版本匹配且暂停提交时成功；初始控制态不允许动作返回管理前置条件错误， 提交 CAS 失去竞争返回可重试快照过期，权限或持久化失败返回对应错误。

pub async fn resume( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, operation_id: &str, ) -> Result<()>

业务作用：管理面恢复实例；已逾期的 deadline 由 timer 在恢复后立即合并触发一次， 不重置到未来，也不逐 tick 补烧。

参数说明：

• management: 已认证主体、原因与 saga.resume 权限快照。
• tenant: 已认证管理主体被授权访问的租户。
• saga_id: 实例身份。
• operation_id: 稳定管理 operation id；重放已提交操作不会再次改变控制态。

返回：恢复生效返回 Ok；实例版本或控制状态过期返回错误。

pub async fn resume_with_expectation( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, operation_id: &str, expectation: SagaManagementExpectation, ) -> Result<()>

业务作用：在事务内复验调用方版本后恢复实例，并保持原业务 deadline 语义。

参数说明：管理上下文、租户、实例和操作身份定位动作，expectation 约束事务内实例版本。

返回：版本匹配且恢复提交时成功；初始控制态不允许动作返回管理前置条件错误， 提交 CAS 失去竞争返回可重试快照过期，权限或持久化失败返回对应错误。

pub async fn retry_compensation( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, operation_id: &str, now_ms: i64, ) -> Result<SagaStatus>

业务作用：管理面在人工介入后重试补偿——基于持久化事实继续冻结计划内的补偿。

只允许 MANUAL_INTERVENTION -> COMPENSATING 且实例已有冻结计划；计划首项为 HALTED 时，同事务管理审计会显式重开 Orchestrator 投影，并把 operation id 写入 command 供 Participant 二次门禁。Unknown 仍禁止直接重做；无剩余计划项时收敛。

参数说明：

• management: 已认证主体、原因与 saga.retry_compensation 权限快照。
• tenant: 已认证管理主体被授权访问的租户。
• saga_id: 实例身份。
• operation_id: 稳定管理 operation id，作为迁移触发身份进入审计链。
• now_ms: 当前时刻（epoch 毫秒）。

返回：推进后的实例状态；PAUSED、实例不在人工介入、无冻结计划或 CAS 冲突返回错误。

pub async fn retry_compensation_with_expectation( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, operation_id: &str, now_ms: i64, expectation: SagaManagementExpectation, ) -> Result<SagaStatus>

业务作用：在事务内命中调用方版本后重开冻结补偿计划，禁止旧快照发布新的外部命令。

参数说明：管理上下文、实例、操作身份和时刻定位动作，expectation 约束事务内实例版本。

返回：提交后的状态；过期快照、权限、恢复前置条件或持久化失败返回错误。

pub async fn retry_resolution( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, operation_id: &str, now_ms: i64, ) -> Result<SagaStatus>

业务作用：在 Unknown 解决硬预算耗尽后，由经授权的操作员恢复一次查询周期。

本入口只重发 typed resolve；若查询投影为 HALTED，只在同事务审计成立后重开查询， UNKNOWN/HALTED 补偿事实本身绝不被直接改回 PENDING，不会二次退款。

参数说明：

• management: 已认证主体、原因与 saga.retry_resolution 权限快照。
• tenant: 已授权租户。
• saga_id: 人工介入实例身份。
• operation_id: 稳定管理操作身份，重放不再发新命令。
• now_ms: 当前 epoch 毫秒，用于新的 resolve timeout 与硬预算。

返回：成功返回 WAITING_RESOLUTION；PAUSED、非人工态、当前没有 Unknown 事实、 步骤不支持 Poll、租户/权限不匹配或 CAS 冲突返回错误。

pub async fn retry_resolution_with_expectation( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, operation_id: &str, now_ms: i64, expectation: SagaManagementExpectation, ) -> Result<SagaStatus>

业务作用：在事务内命中调用方版本后重开 Unknown 解决周期，禁止旧快照发布新的查询命令。

参数说明：管理上下文、实例、操作身份和时刻定位动作，expectation 约束事务内实例版本。

返回：提交后的状态；过期快照、权限、解决前置条件或持久化失败返回错误。

pub async fn manual_close( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, operation_id: &str, ) -> Result<SagaStatus>

业务作用：系统外人工处置完成后关闭自动化——MANUAL_INTERVENTION -> MANUALLY_CLOSED 的唯一管理入口，实例由此离开非终态集合，nasaga_manual_intervention 回落。

关闭只记录“自动化已由人工关闭“这一事实：不写 COMPLETED/COMPENSATED、不伪造 未经系统证明的业务结果；未完成的补偿计划仍以快照保留。动作与 MANUAL_CLOSE_ACTION 审计同事务提交， 状态机受限出边只放行携带该同事务证据的迁移，自动 worker 无法触发。

参数说明：

• management: 已认证主体、强制原因与 saga.manual_close 权限快照。
• tenant: 已认证管理主体被授权访问的租户。
• saga_id: 实例身份。
• operation_id: 一次性管理 operation 身份；重放按幂等成功返回，不产生第二次迁移。

返回：关闭后（或幂等重放时）的实例状态。能力开关未开启、无权限、实例不在 MANUAL_INTERVENTION、PAUSED 或 CAS 冲突返回错误。

pub async fn manual_close_with_expectation( &self, management: &SagaManagementContext, tenant: &TenantId, saga_id: &SagaId, operation_id: &str, expectation: SagaManagementExpectation, ) -> Result<SagaStatus>

业务作用：在事务内命中调用方版本后人工关闭自动化，避免旧快照终结已并发变化的实例。

参数说明：管理上下文、租户、实例和操作身份定位动作，expectation 约束事务内实例版本。

返回：关闭后或幂等重放状态；过期快照、权限、状态或持久化失败返回错误。

impl<B> Orchestrator<B>

where B: SagaBackend + SagaBackendFactory,

pub async fn start_scheduled_batch( &self, spec: &ScheduledBatchSpec<'_>, items: &[ScheduledItem], still_leader: impl FnMut() -> bool, ) -> Result<ScheduledBatchReport>

业务作用：在一次已取得触发权的名义调度时刻内，按预算幂等发起一批 Saga。

触发权合同：调用方必须已持有领导权与该名义时刻的 FireLog claim；still_leader 在每个对象之间复验——失去领导权立即停止领取新项，已提交创建的实例由 Saga 自身推进，不随领导权变化回滚。重复触发（重试、换届重扫、崩溃后补漏）由派生 业务幂等键吸收；同 key 的创建摘要漂移由创建入口 fail-closed，以带稳定原因码的 错误停止本批（输入口径漂移是配置事故，不是可跳过的单行失败）。

参数说明：

• spec: 批次身份、目标 workflow 与有界预算。
• items: 本名义时刻扫描出的业务对象列表（来自已提交业务事实）。
• still_leader: 领导权探针；返回假即停止领取新项。

返回：批次结果；创建路径的基础设施失败返回错误（已创建实例保持已提交）。