超时不等于失败：AI Agent 缺失的关键状态

当一个 AI Agent 的 HTTP 请求或浏览器工具调用超时后，你的系统会记录什么？

如果记录为失败，那么这个 Agent 存在一个盲点。网络超时并不代表远程服务器上的操作失败了——它只意味着连接在客户端收到响应之前就关闭了。如果服务器已经处理了这个变更，盲目重试将产生重复产物：重复支付、重复工单、重复邮件，或者一篇重复的文章。

如果记录为成功，那是在虚构确定性。

缺失的状态是 outcome_unknown——这是一种一等公民的操作状态，它会阻止自动重试、记录未确认的变更，并将执行交接给一个明确的调和对账循环。

在前一篇帖子中，我们讨论了为什么 Agent 需要动作收据（action receipts）而不是纯粹语义化的记忆。在与分布式系统和记忆边界领域的从业者进行了深入讨论之后，本文将这个概念转变为一个具体的、可测试的状态机，你可以将其嵌入任何生产环境的 Agent 框架中。

临界线：发送前失败 vs 发送后歧义

并非所有异常都是平等的：

[Intent Recorded]
       |
       v
[Attempting Transport] ---> (DNS / Local Socket / Auth error) ---> [REJECTED / SAFE_TO_RETRY]
       |
  (Bytes sent)
       |
       v
[Awaiting Response]   ---> (Connection Timeout / Drop / 504)  ---> [OUTCOME_UNKNOWN]

发送前失败（Pre-Send Failures）：如果 DNS 查询失败、凭证在本地缺失，或者在单个字节离开 socket 之前连接就被拒绝，那么世界还没有发生任何改变。该动作被确定性地未执行，可以安全重试。

发送后歧义（Post-Send Ambiguity）：一旦字节上了wire，传输失败就不再是服务器状态的指标。服务器可能已经提交了变更并在响应序列化过程中崩溃了，或者一个中间代理在 30 秒后超时了，而后台 worker 其实已经完成了任务。

将发送后歧义视为失败，是自动化重复风暴的根本原因。

以下是一个受保护 Agent 动作的完整生命周期：

幂等键 vs 意图指纹

在支付工程领域，分布式共识是通过「至少一次交付」配合服务器端去重键（幂等键，Idempotency Key）来实现的。

当外部平台原生支持幂等头（如 Stripe 中的 Idempotency-Key: <uuid> 或 GitHub GraphQL 变更键）时，调和是直接的：如果你超时了，用完全相同的键重新发送即可。

然而，绝大多数 Web API、CRUD 服务和浏览器驱动的界面并不支持原生幂等键。在这些环境中，调用方必须承担这个负担：

标准化的意图指纹（Normalized Intent Fingerprint）：在发送之前，计算语义载荷（变更类型、资源目标、标准化后的 body 字段）的规范密码学哈希。

写后读调和（Read-After-Write Reconciliation）：当 outcome_unknown 触发时，Agent 查询读 API（或搜索端点），在有界的时间窗口内查找由该 Agent 账户创建的资源，并匹配意图指纹。

import hashlib
import json

def compute_intent_fingerprint(method: str, path: str, payload: dict) -> str:
    canonical = json.dumps(
        {"method": method.upper(), "path": path, "payload": payload},
        sort_keys=True,
        separators=(",", ":")
    )
    return f"sha256:{hashlib.sha256(canonical.encode('utf-8')).hexdigest()}"

保留歧义审计轨迹

状态机设计中的一个微妙陷阱是破坏性的就地更新。

如果一个动作经历了 submitted -> outcome_unknown -> reconciling -> succeeded，而你只是简单地将状态覆盖为 succeeded，那么你就摧毁了该动作在数小时内处于不确定状态的历史记录。

在事后复盘时（或者在审计另一个 worker 在那个时间窗口内观察到缺失状态的竞态条件时），了解一个动作是如何到达成功状态的，和最终状态本身同样关键。

一个健壮的 action receipt 会保留完整的转换轨迹：

{
  "operation_id": "20260818T190000Z-a1b2c3d4e5",
  "operation": "articles.create",
  "state": "succeeded",
  "intent_fingerprint": "sha256:4d8a...",
  "state_history": [
    { "state": "planned", "recorded_at": "2026-08-18T19:00:00Z" },
    { "state": "submitted", "recorded_at": "2026-08-18T19:00:01Z" },
    { 
      "state": "outcome_unknown", 
      "recorded_at": "2026-08-18T19:00:31Z",
      "error": { "code": "timeout", "message": "Gateway Timeout 504" }
    },
    { 
      "state": "reconciling", 
      "recorded_at": "2026-08-18T19:05:00Z" 
    },
    { 
      "state": "succeeded", 
      "recorded_at": "2026-08-18T19:05:02Z",
      "reconciliation": {
        "evidence": "Readback from /api/articles matched title fingerprint",
        "external_id": 4407310
      }
    }
  ]
}

最小可行实现

以下是受保护执行与调和模式的 Python 实现：

from dataclasses import dataclass, field
from datetime import datetime, timezone
import uuid

@dataclass
class ActionReceipt:
    operation_id: str
    action: str
    target: str
    fingerprint: str
    state: str = "planned"
    external_id: str | None = None
    state_history: list[dict] = field(default_factory=list)

    def transition_to(self, new_state: str, **meta):
        self.state = new_state
        self.state_history.append({
            "state": new_state,
            "recorded_at": datetime.now(timezone.utc).isoformat(),
            **meta
        })

def execute_guarded_action(client, action: str, target: str, payload: dict) -> ActionReceipt:
    receipt = ActionReceipt(
        operation_id=f"{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}-{uuid.uuid4().hex[:8]}",
        action=action,
        target=target,
        fingerprint=compute_intent_fingerprint("POST", target, payload)
    )
    receipt.transition_to("planned")

    # 1. Record intent persistently before touching network
    persist_receipt(receipt)

    receipt.transition_to("submitted")
    persist_receipt(receipt)

    try:
        response = client.post(target, json=payload, timeout=10.0)
        receipt.external_id = response.json().get("id")
        receipt.transition_to("succeeded", status_code=response.status_code)
    except TimeoutError as exc:
        # Crucial: DO NOT retry. Mark as ambiguous.
        receipt.transition_to("outcome_unknown", error=str(exc))
    except Exception as exc:
        receipt.transition_to("rejected", error=str(exc))
    finally:
        persist_receipt(receipt)

    return receipt

def reconcile_receipt(client, receipt: ActionReceipt, read_fn) -> ActionReceipt:
    if receipt.state != "outcome_unknown":
        return receipt

    receipt.transition_to("reconciling")
    persist_receipt(receipt)

    matched_item = read_fn(client, receipt.fingerprint)
    if matched_item:
        receipt.external_id = matched_item["id"]
        receipt.transition_to("succeeded", evidence="Matched on readback query")
    else:
        # If absence is authoritatively proven, mark safe for a fresh attempt
        receipt.transition_to("safe_to_retry", evidence="Authoritative readback showed 0 records")

    persist_receipt(receipt)
    return receipt

限制与最终一致性

最终一致性滞后（Eventual Consistency Lag）：在分布式数据库中，新创建的资源可能不会立即在读副本上可见。调和循环必须用有界退避来考虑传播延迟，而不是立即断定不存在。

盲变更端点（Blind Mutation Endpoints）：如果一个 API 允许变更但不暴露列表、搜索或读回端点，调和就无法自动化。这类动作必须转换为 manual_review。

破坏性操作（Destructive Operations）：删除操作（DELETE）在调和上天生更棘手，因为「不存在」才是目标状态。一个缺失的条目可能意味着删除成功了，也可能意味着该条目根本不存在。

---

在构建与外部 API 或浏览器界面交互的自主 Agent 时：

在你的系统中，哪种外部写操作在意外超时后最难调和？你如何防止重复执行？