findscripter Temporal 持久工作流编排（Python）
何时使用
适用（长时运行 + 需崩溃恢复/精确一次语义的编排）：
- 跨微服务的分布式事务、Saga 模式（带补偿回滚）。
- 长周期业务流程（小时到数年）：订单履约、支付处理、SLA 升级。
- 人审介入（human-in-the-loop）审批流；实体工作流（购物车/账户/库存）。
- 多步数据处理管道、基础设施自动化编排。
- 触发词：Temporal、工作流编排、durable workflow、Saga、Activity、Worker、信号查询、确定性。
不该用（负边界）：
- 无状态的请求-响应、普通 CRUD —— 直接写 service 即可，别引入 Worker/集群。
- 简单定时任务 —— 用 cron / 调度器，不必上 Temporal。
- 单机短脚本、一次性批处理 —— 编排开销不划算。
- 纯前端，或不需要持久状态/重试/恢复的场景。
核心心智模型：Workflow = 编排（确定性、可重放）；Activity = 与外部世界交互（可重试、需幂等）。工作流代码靠事件历史重放恢复，因此必须确定性；副作用一律放进 Activity。
步骤 / 指令
1. 建模：识别编排逻辑（→ Workflow）与外部副作用（→ Activity）。
- Workflow 用 @workflow.defn，入口方法 @workflow.run（async def）。
- Activity 用 @activity.defn，从 workflow 内只能用 workflow.execute_activity 调用。
2. 选 Activity 执行模型（关键反模式：阻塞异步事件循环 → 串行化）：
- 异步 I/O（API/异步 DB）→ async Activity（asyncio）。
- 阻塞 I/O（同步 DB 客户端/文件/旧库）→ 同步 Activity + ThreadPoolExecutor。
- CPU 密集（数据处理/ML 推理）→ 同步 Activity + ProcessPoolExecutor。
3. 配超时与重试（超时是必填项，不是可选）：
- start_to_close_timeout（单次尝试上限，几乎总要设）。
- schedule_to_close_timeout（含重试的总时长上限）。
- heartbeat_timeout（长 Activity 检测卡死，配合 activity.heartbeat()）。
- RetryPolicy(initial_interval, backoff_coefficient, maximum_interval,
maximum_attempts, non_retryable_error_types)。
4. 保证确定性（见“注意事项”清单），保证 Activity 幂等（重试可能重复执行）。
5. Saga：为每步登记补偿动作；任一步失败按逆序执行补偿。
6. 信号/查询：@workflow.signal 收外部事件改状态；@workflow.query 只读快照。
7. 启动 Worker：注册全部 workflows + activities，绑定 task_queue（须与 client 一致）。
8. 测试：用 time-skipping 的 WorkflowEnvironment（sleep 瞬时跳过）+ mock Activity；
CI 跑 replay 测试校验确定性。
9. 部署：容器化 Worker，水平扩容，优雅停机，监控队列深度/成功率，做 workflow 版本治理。
硬规则：
- 工作流内禁止
datetime.now()/random.random()/ 线程锁 / 全局可变状态 / 直接外部调用 —— 否则重放破裂。 - 改工作流逻辑用
workflow.get_version()做版本门，保证旧历史可重放。 - 单参数 payload ≤ 2MB；超限改传引用（如对象存储 key）。
- 区分瞬时失败（让重试）与永久失败（
ApplicationError(non_retryable=True)）。
示例
最小工作流 + Activity（确定性时间、显式超时与重试）：
from datetime import timedelta
from temporalio import workflow, activity
from temporalio.common import RetryPolicy
@activity.defn
async def charge_payment(order_id: str, amount: int) -> str:
# 幂等：用 order_id 做去重键，重试不会重复扣款
return f"txn-{order_id}"
@workflow.defn
class OrderWorkflow:
def __init__(self) -> None:
self._approved = False
@workflow.run
async def run(self, order_id: str, amount: int) -> str:
# 用 workflow.now() 而非 datetime.now()，保证可重放
started = workflow.now()
txn = await workflow.execute_activity(
charge_payment, args=[order_id, amount],
start_to_close_timeout=timedelta(seconds=30),
retry_policy=RetryPolicy(
initial_interval=timedelta(seconds=1),
backoff_coefficient=2.0,
maximum_interval=timedelta(seconds=60),
maximum_attempts=5,
non_retryable_error_types=["CardDeclined"],
),
)
return txn
@workflow.signal
def approve(self) -> None: # 外部事件：人审通过
self._approved = True
@workflow.query
def is_approved(self) -> bool: # 只读状态快照
return self._approved
Saga 补偿（任一步失败逆序回滚）：
@workflow.run
async def run(self, order):
compensations = []
try:
await workflow.execute_activity(reserve_inventory, order, start_to_close_timeout=timedelta(seconds=10))
compensations.append(release_inventory)
await workflow.execute_activity(charge_payment, order, start_to_close_timeout=timedelta(seconds=30))
compensations.append(refund_payment)
await workflow.execute_activity(ship_order, order, start_to_close_timeout=timedelta(seconds=10))
except Exception:
for comp in reversed(compensations): # 逆序补偿
await workflow.execute_activity(comp, order, start_to_close_timeout=timedelta(seconds=10))
raise
等待人审信号（带超时）：
@workflow.run
async def run(self, doc_id):
try:
await workflow.wait_condition(lambda: self._approved, timeout=timedelta(days=3))
except TimeoutError:
return "auto-rejected"
return "approved"
启动 Worker（注册 + 绑定 task queue）：
from temporalio.client import Client
from temporalio.worker import Worker
async def main():
client = await Client.connect("localhost:7233")
async with Worker(client, task_queue="orders",
workflows=[OrderWorkflow],
activities=[charge_payment]):
await asyncio.Event().wait() # 常驻；生产中接优雅停机
时间跳跃测试（月级 sleep 瞬时执行 + mock Activity）：
from temporalio.testing import WorkflowEnvironment
async def test_order():
async with await WorkflowEnvironment.start_time_skipping() as env:
async with Worker(env.client, task_queue="t",
workflows=[OrderWorkflow], activities=[charge_payment]):
result = await env.client.execute_workflow(
OrderWorkflow.run, args=["o1", 100],
id="wf-1", task_queue="t")
assert result.startswith("txn-")
本地起 Temporal 做集成测试：temporal server start-dev（或 Docker Compose）。
注意事项
确定性违规（最常见的崩溃源，工作流内一律禁止）：
datetime.now()→ 用workflow.now()；random.random()→ 用workflow.random()。- 线程/锁/全局可变状态；从工作流直接发 HTTP/查 DB（必须经 Activity）。
- 非纯函数逻辑；依赖迭代顺序不稳定的结构。
Activity 实现：
- 必须幂等——重试会重复执行；用业务键去重，外部写操作要可安全重放。
- 漏配超时 = 隐患；长 Activity 要
activity.heartbeat()否则 worker 卡死无法检测。 - 别在 async Activity 里跑阻塞/重计算代码，会冻结事件循环 → 改用同步 Activity + 线程/进程池。
- 单参数 payload ≤ 2MB。
测试与部署：
- 不用 time-skipping 环境 → 测一个 30 天工作流真等 30 天；务必用
start_time_skipping()。 - 工作流测试要 mock Activity，只验编排逻辑。
- CI 必须跑 replay 测试，用生产历史校验代码改动不破坏确定性。
- Worker 上必须注册全部用到的 workflow/activity，且
task_queue与 client 严格一致，否则任务永远 pending。 - 改工作流逻辑用
workflow.get_version()灰度，避免在途实例重放失败；部署做优雅停机。
通用约束：本技能输出不替代针对具体环境的验证、测试与专家评审；关键输入（领域边界、权限、超时/重试预算、成功标准）缺失时先停下澄清。
官方参考：python.temporal.io、docs.temporal.io/workflows、docs.temporal.io/develop/python/testing-suite。
互见
- related：
async-python-patterns（Activity 的 async/await 与 executor 选型基础）、saga-orchestration（Saga 模式与补偿设计的更深展开）、event-sourcing-cqrs（同属持久化/事件回放范式，对比取舍）、distributed-tracing（为工作流/Activity 加可观测性）。 - combines_with：
microservices-patterns（用 Temporal 编排跨服务事务）、backend-architecture-patterns（在基础设施层引入持久化执行而不增加架构复杂度）。
采编自 sickn33/antigravity-awesome-skills（MIT）。