你部署了一个用 LangGraph 写的基础设施开通 agent，它本应在调用任何 Terraform 之前先验证 AWS 凭证、确认目标环境不是生产环境、再核对资源配额。在一个特别长的任务里，编排器的 pre-flight 步骤超时了，流水线的兜底逻辑把它标成「skipped (non-fatal)」就继续往下走。Terraform agent 接着执行，把该区域的 EC2 配额耗尽，留下一个开通了一半的环境，要花好几个小时才能清理干净。Pre-flight 检查只有在被无条件强制执行时才有价值。任何允许它被跳过的代码路径，迟早都会被触发。
**最快的修复：**把 pre-flight 写成一个「失败就抛异常」的函数（包括超时、以及「这个任务类型没有定义检查项」也要抛），然后在任务入口和重试入口的第一行无条件调用它。只要 pre-flight 在结构上还能返回一个调用方可以忽略的结果，它就是建议性的，而不是强制性的。下面的内容都是在加固这一个核心思想：按任务类型注册检查、用环境变量守卫开关、在工作流图层面强制、以及写 CI 测试。
你属于哪一种情况？
抓一次真实「跳过了检查」的运行记录，把现象对应到下面六种原因之一。绝大多数「pre-flight 被跳过」的事故都属于其中之一。
| 运行记录里的现象 | 可能的原因 | 对应修复 |
|---|---|---|
| Pre-flight 跑了，记录了失败，但任务仍继续执行 | 结果被当成建议性（if passed: log()） | Step 1 |
| Pre-flight 记录了超时，然后任务继续了 | except TimeoutError 里设了 passed=True | Step 1 |
| 系统提示里有这个检查，但部分运行的工具调用记录里没有 | 只写在 prompt 里，没在代码里强制 | Step 4 |
| 只在 staging 或某些机器上被跳过 | 那里设了 FAST_MODE / SKIP_PREFLIGHT 开关 | Step 3 |
| 新任务类型完全没有任何检查 | 注册表没为新类型更新 | Step 2 |
| Pre-flight「跑了」，但失败是在执行开始后才记录的 | 用 create_task 触发后没有 await | Step 1 |
常见原因
1. Pre-flight 结果在编排逻辑里被当成可选的
编排代码写的是 preflight_result = run_preflight(task)，但只用 if preflight_result.passed: log_success() 来消费结果。当 preflight_result.passed 是 False、或者 preflight_result 是 None（超时或出错）时，它从不阻断执行。任务总是继续往下走。
怎么判断：找到消费 pre-flight 结果的代码。如果不管结果如何（包括 None、出错、False）执行都照样继续，这个检查就是建议性的，而不是强制性的。
2. Pre-flight 超时，超时被当成「可以跳过」处理
Pre-flight 调用了一个很慢的外部校验服务。30 秒后检查超时，异常处理里捕获了 TimeoutError，并为了「稳妥起见、不阻塞用户」把 preflight_result = PreflightResult(passed=True, skipped=True)。这是反的：超时应该阻断，而不是放行。一个你没能跑完的检查，就是一个没有通过的检查。
怎么判断：找到 pre-flight 步骤的 except TimeoutError（或等价的）处理分支。如果它设了 passed=True 或直接落到执行流程，它就把超时变成了自动批准。
3. Pre-flight 写在 prompt 里，但代码里没有强制
系统提示写着「做任何事之前，永远先检查 X、Y、Z」。Agent 有时会跳过 Z 这一步，因为它判断「Z 跟这个任务无关」。基于 prompt 的 pre-flight 不可靠：模型总能给自己找到跳过某一步的理由。
怎么判断：把 prompt 里列出的 pre-flight 步骤，和每次运行开头实际发出的工具调用对比。如果 Z 出现在 prompt 里、却在部分运行的工具调用记录里缺席，那它就只在 prompt 里、没有被代码强制。
4. 某些模式下 pre-flight 被硬编码跳过
为了开发提速加了一个「快速模式」开关：if fast_mode: skip_preflight()。这个开关通过环境变量设置，有人带着 FAST_MODE=true 部署到了 staging，却没意识到 pre-flight 被关掉了。这个开关后来也一直没清理。
怎么判断：在代码库里搜 skip_preflight、FAST_MODE、bypass_checks 之类。如果有任何开关能关掉 pre-flight，检查它是否可能在非开发环境里被设上。
5. 新增了任务类型，但 pre-flight 没为它更新
原来的 pre-flight 检查的是「AWS 凭证是否有效」和「环境是否非生产」。流水线新增了一个任务类型（数据库迁移）。数据库任务需要额外的 pre-flight 检查（备份存在、迁移幂等、回滚方案就绪），但没人为新任务类型更新 pre-flight。
怎么判断：列出流水线处理的每一种任务类型，以及各自需要的 pre-flight 检查。任何没有完整 pre-flight 规格的任务类型，都是检查不足的。
6. Pre-flight 跑了，但 Agent 没等结果
在异步流水线里，asyncio.create_task(run_preflight(task)) 把检查丢到后台。主流程立刻继续往下执行，没有 await。Pre-flight 检查确实在跑、也可能会失败，但等到失败被记录下来时，执行早已经开始了。
怎么判断：看 pre-flight 是被 await 的（await run_preflight(task)），还是作为后台任务触发的。后台跑的 pre-flight 等于没有 pre-flight。
最短修复路径
Step 1：把 pre-flight 做成会阻断的门禁，而不是建议性的步骤
class PreflightError(Exception):
"""当某个必需的 pre-flight 检查失败时抛出。执行必须停止。"""
def require_preflight(task: dict) -> None:
"""
必须在任何执行动作之前调用。任何失败都会抛 PreflightError。
永远不返回 None。永远不吞掉异常。
"""
checks = get_required_checks(task["type"])
results = []
for check in checks:
try:
result = check.run(task, timeout=30)
except TimeoutError:
raise PreflightError(
f"Pre-flight check '{check.name}' timed out — execution blocked. "
"Fix the check or resolve the underlying connectivity issue."
)
except Exception as e:
raise PreflightError(
f"Pre-flight check '{check.name}' raised an error: {e}"
) from e
if not result.passed:
raise PreflightError(
f"Pre-flight check '{check.name}' failed: {result.reason}"
)
results.append(result)
logger.info("All %d pre-flight checks passed for task %s", len(results), task["id"])Step 2：用注册表按任务类型定义必需的检查项
PREFLIGHT_REGISTRY: dict[str, list[PreflightCheck]] = {
"terraform_provision": [
AWSCredentialsCheck(),
NonProductionEnvironmentCheck(),
ResourceQuotaCheck(min_remaining={"ec2": 10, "vpc": 2}),
TerraformValidateCheck(),
],
"database_migration": [
DatabaseConnectionCheck(),
BackupExistsCheck(max_age_hours=24),
MigrationIdempotencyCheck(),
RollbackPlanCheck(),
],
"code_deploy": [
GitBranchCheck(allowed_branches=["main", "release/*"]),
TestSuitePassCheck(),
SecretScanCheck(),
],
}
def get_required_checks(task_type: str) -> list[PreflightCheck]:
if task_type not in PREFLIGHT_REGISTRY:
raise ValueError(
f"No pre-flight checks defined for task type '{task_type}'. "
"Add an entry to PREFLIGHT_REGISTRY before adding new task types."
)
return PREFLIGHT_REGISTRY[task_type]新任务类型如果没注册，这里会直接报错（强制你显式注册 pre-flight），所以你不可能忘记加检查项。
Step 3：移除所有「跳过」开关，或把它们限制在本地模式
import os
def require_preflight_with_env_guard(task: dict) -> None:
if os.environ.get("SKIP_PREFLIGHT") == "true":
env = os.environ.get("ENVIRONMENT", "unknown")
if env not in ("local", "test"):
raise ConfigurationError(
"SKIP_PREFLIGHT=true is not allowed in non-local environments. "
f"Current environment: {env}"
)
logger.warning("SKIP_PREFLIGHT=true — only allowed in local/test environments")
return
require_preflight(task)在 staging 和生产环境里硬性阻断这个跳过开关。
Step 4：在工作流图的定义层面强制 pre-flight，而不只在代码里
在 LangGraph v1.0（2025 年 10 月起为稳定版）里，set_entry_point() 已弃用，改为用 START 哨兵节点 显式连边。把入口连成 preflight 是从起点唯一可达的节点，失败则路由到 END：
from langgraph.graph import StateGraph, START, END
graph = StateGraph(State)
graph.add_node("preflight", preflight_node)
graph.add_node("execute", execute_node)
# preflight 是从 START 唯一可达的节点
graph.add_edge(START, "preflight")
# 根据 preflight 结果分支；绝不让 START 直接到达 execute
graph.add_conditional_edges(
"preflight",
lambda s: "execute" if s["preflight_passed"] else "abort",
{"execute": "execute", "abort": END},
)让 preflight_node 在任何失败或超时时把 preflight_passed=False（或直接抛异常）。因为没有从 START 到 execute 的边，所以没有任何路径能在不经过 preflight 的情况下到达执行。
如果用 Temporal，把 pre-flight 做成一个会快速失败的独立 activity。截至 2026 年 6 月，maximum_attempts=1 表示只尝试一次、不重试；但更干净的做法是在 activity 内部抛一个不可重试的 ApplicationError，让编排器立刻停下，而不是在一个根本无法恢复的前提条件上把整个重试预算耗完：
from datetime import timedelta
from temporalio import activity, workflow
from temporalio.common import RetryPolicy
from temporalio.exceptions import ApplicationError
@activity.defn
async def run_preflight_activity(task: dict) -> None:
failures = run_required_checks(task) # 你的检查执行器：抛异常或收集失败项
if failures:
# non_retryable：缺失的备份、过期的 token 不会因为重试就自己好了
raise ApplicationError(
f"Pre-flight failed: {failures}", type="PreflightFailed", non_retryable=True
)
@workflow.defn
class ProvisionWorkflow:
@workflow.run
async def run(self, task: dict):
# 必需的第一个 activity；由 start_to_close_timeout 限定时长
await workflow.execute_activity(
run_preflight_activity,
task,
retry_policy=RetryPolicy(maximum_attempts=1),
start_to_close_timeout=timedelta(seconds=60),
)
await workflow.execute_activity(
execute_activity, task, start_to_close_timeout=timedelta(minutes=30)
)Step 5：为每个 pre-flight 检查写一个 CI 测试
def test_preflight_blocks_on_missing_backup():
task = {"type": "database_migration", "db": "prod_replica"}
# 模拟最近 24 小时内没有备份
with patch("checks.backup_check.get_latest_backup", return_value=None):
with pytest.raises(PreflightError, match="BackupExistsCheck failed"):
require_preflight(task)
def test_preflight_blocks_on_timeout():
task = {"type": "terraform_provision", "env": "staging"}
with patch("checks.aws_check.validate_credentials", side_effect=TimeoutError):
with pytest.raises(PreflightError, match="timed out"):
require_preflight(task)如何确认已经修好
在你信任这个门禁之前，跑下面三项检查：
- **负向测试能通过。**主动制造一个真实的失败场景（吊销 token、删掉备份、指向生产环境），确认任务在任何会产生副作用的工具调用 之前 就中止了。execute 步骤的日志根本不应该出现。
- **没有残留的绕过路径。**在仓库里 grep
skip_preflight、FAST_MODE、SKIP_PREFLIGHT、bypass、if .*preflight。每一处命中要么被移除，要么被一个硬性阻断非本地使用的环境检查守住。 - **重试路径也被覆盖。**对一个原本能通过、但现在前提条件被破坏的运行触发重试。重试必须重新跑 pre-flight 并中止，而不是基于过期的结果直接恢复。
现在在运行日志里，每一次被阻断的运行都应该看到带着失败检查名和原因的 PreflightError（或不可重试的 PreflightFailed），每一次成功的运行都应该看到一行 all N checks passed。如果某次运行进入了执行、却两行都没有，那就说明还有代码路径在绕过这个门禁。
预防建议
- 把 pre-flight 实现成一个会在任何失败时抛异常的强制阻断函数；它绝不应返回一个可以被忽略的结果。
- 按任务类型注册必需的 pre-flight 检查；新任务类型缺失注册时应该大声报错，而不是悄悄通过。
- 绝不允许在非本地环境里使用
SKIP_PREFLIGHT之类开关；用显式的环境检查守住。 - 把 pre-flight 设为工作流图里无条件的第一个节点；移除任何能在不经过 pre-flight 的情况下到达执行的路径。
- 把 pre-flight 超时当作阻断性失败，而不是放行。一个你没能跑完的检查，就是一个没有通过的检查。
- 写 CI 测试，验证每个 pre-flight 检查在它要拦截的每一种失败场景下都能阻断执行。
- 每当流水线新增任务类型或新能力时，复查并更新 pre-flight 检查。
- 为每一次运行记录带时间戳和结果的 pre-flight 日志；审计需要确认检查确实跑过了，而不只是「通过了」。
常见问答 (FAQ)
Q: 怎么避免 pre-flight 太严，把正常工作也挡掉？ A: 让每个检查只在「确实会阻碍安全执行」的条件下失败，而不是在「只是不够理想」的条件下失败。「24 小时内没有备份」会挡住迁移；「备份是 6 小时前的（在 24 小时上限内）」则不会。阈值要根据真实事故来标定，而不是理论上的极值。
Q: Pre-flight 检查应该重试吗？ A: 针对外部资源的检查（API 可达性、凭证有效性）可以带短延迟重试一次。针对逻辑前提的检查（备份存在、环境非生产）不应该重试：如果条件没满足，重试也不会改变它。快速失败，让人来修复这个前提条件。
Q: 怎么把 pre-flight 的结果传给主 Agent，让它不用重新推导一遍？ A: 把 pre-flight 结果放进交给执行 Agent 的 handoff 上下文里。执行 Agent 从上下文读到「backup verified at 2026-05-25T14:30:00Z, quota remaining: 15 EC2 instances」，就能在推理时直接引用，而不必重新检查。
Q: 如果某个 pre-flight 检查需要 LLM 推理（比如「这次迁移安全吗？」）怎么办？
A: 基于 LLM 的 pre-flight 检查适合做定性判断，但要给它配一个确定性的兜底。「这次迁移安全吗？」要拆成：LLM 评审 + --dry-run 执行 + schema diff 对比。绝不要让一个会阻断生产工作的门禁只依赖 LLM 的判断。
Q: 我的编排器本来就会重试失败的 activity，为什么 pre-flight 失败还是要几分钟才暴露？
A: 因为默认的重试策略会一直重试。一个前提条件失败（token 过期、备份缺失）永远不会因为重试就成功，于是它会在拖垮整个重试预算之后才让运行失败。把 pre-flight 失败标记为终止性的：在 Temporal 里抛 ApplicationError(..., non_retryable=True) 或设 maximum_attempts=1；在 LangGraph 里把失败的 preflight 节点直接路由到 END。不可重试的机制可参考 Temporal 的 retry policy 文档。