SafeAgent — Execution Guard for AI Agents

SafeAgent — AI Agent 执行防护

POST /claim · safeagent-production.up.railway.app
控制面板：safeagent-dashboard-2.vercel.app

# Claim an action — free, no payment header required
curl -s -X POST https://safeagent-production.up.railway.app/claim \
  -H "Content-Type: application/json" \
  -d '{"request_id":"order:TQQQ:buy:6:2026-05-19T13:31:00-04:00","action":"order"}'
# → {"status":"PROCEED","request_id":"..."}

# Settle after the action fires
curl -s -X POST https://safeagent-production.up.railway.app/settle/order:TQQQ:buy:6:2026-05-19T13:31:00-04:00 \
  -H "Content-Type: application/json" \
  -d '{"result":{"status":"completed"}}'
# → {"status":"committed","request_id":"..."}

# Retry with the same request_id — returns SKIP
curl -s -X POST https://safeagent-production.up.railway.app/claim \
  -H "Content-Type: application/json" \
  -d '{"request_id":"order:TQQQ:buy:6:2026-05-19T13:31:00-04:00","action":"order"}'
# → {"status":"SKIP","request_id":"...","existing":{...}}

# Free test endpoint — same logic, limited to 10 calls per IP
curl -s -X POST https://safeagent-production.up.railway.app/claim/test \
  -H "Content-Type: application/json" \
  -d '{"agent_id":"bot-1","action_type":"order","scope":"TQQQ:buy:bar:2026-05-19T13:31:00-04:00"}'
# → {"status":"PROCEED","request_id":"...","test":true,"calls_remaining":9}

已在 Bazaar 编入索引。

CrewAI 重试 A/B 实测

一项无需凭据的实验，测试了 CrewAI 1.15.21 在同一进程中实际执行的 `ToolUsage._use` 重试，以及 SafeAgent 0.1.23。在 30 次执行效果后、确认前的试验中，未受保护的路径每个逻辑操作记录了两次效果；SafeAgent 路径记录了一次效果，并保留了 `PENDING` 声明以供协调。

**范围：**使用脚本化确定性 LLM，并通过进程外、能力隔离的账本记录无害效果。本实验未验证进程终止、新工作进程恢复、真实外部提供方或 SafeAgent Control 协调。

方法、源码、测试、哈希值和 180 次试验的证据

SafeAgent Control 创始试点

现向外部运营方开放两个范围受限的只读协调试点。只有当 SafeAgent Control 发现此前未知且可复现、并经运营方同意确属可操作的问题时，才会收取费用。

范围及基于结果的条款

---

---

Mycelium Trails

此实现是一个 Mycelium 提供方。 每个已完成的操作都会向 ARGENTUM 提交一条 trail，其中包含内容寻址的 action_ref （基于四个原像字段计算的 JCS+SHA-256）。返回的 mycelium_trail_id 会包含在 每张回执中，以供独立验证。

---

已在 Soma 上验证

SafeAgent 是 Soma（Mycelium Agent 目录）中首个通过验证的外部集成方。每次生产执行都会通过 Mycelium Trails 锚定到链上，并且无需经由运营方即可独立验证。

• 集成方徽章 — 已验证的外部客户端，在生产环境中运行 Mycelium Trails
• 实时 trails — https://argentum-api.rgiskard.xyz/dashboard/trails?client=safeagent-prod
• 公开审计 — 任何审计人员都可以独立验证 SafeAgent 执行记录

---

功能

SafeAgent 是一款持久化执行声明防护工具。它会抑制使用相同稳定 request_id 的并发和重复尝试，并保留尚未解决的尝试以供协调处理，而不是悄然重试。

每项操作都会根据其逻辑操作生成一个稳定的 request_id。新键返回 PROCEED；已结算的键返回 SKIP，并附带存储的结果。超时或崩溃后仍处于 PENDING 状态的键会继续被阻止，因为仅凭本地状态无法证明提供商是否已接受该操作。只有在检查提供商证据或使用提供商原生幂等机制后，才能恢复处理。

状态机： CLAIMABLE → PENDING → COMMITTED；后续声明会看到 PENDING 或 SKIP。

SafeAgent 可防止的常见故障模式：

场景	不使用 SafeAgent	使用 SafeAgent
Stripe 受理扣款后请求超时	重试可能导致重复扣款	同一键会保持 PENDING，直到与提供商协调确认
用户注册时重试发送欢迎邮件	用户会收到两封欢迎邮件	第二次发送返回 SKIP
Webhook 被重复投递（Stripe/GitHub/Twilio 保证至少投递一次）	事件被处理两次	第二次处理返回 SKIP
重试时配置工作区	创建两个工作区	第二次配置返回 SKIP
AI Agent 在崩溃后重试工具调用	产生重复副作用	同一键会继续被阻止；结果明确标记为尚未解决

保证边界： SafeAgent 能证明本地声明状态，但不能证明外部结果。COMMITTED 表示调用方已结算一份回执。PENDING 表示尚未解决——既不代表失败，也不代表已确认成功。要实现端到端的恰好一次效果，必须使用提供商原生幂等机制，和／或与提供商的权威记录进行协调核对。

---

技术栈

SafeAgent 是 Agent 执行完整性技术栈中的持久化声明与重复抑制层：

Polaris (commit-gated authorization)
└── AgentGraph safety verdict (pre-execution safety gate)
    └── SafeAgent (durable claim guard)  ← you are here
        └── Mycelium Trails (on-chain anchor)

每一层均由独立主体编写，并可独立验证。各层之间互不信任。

---

API

POST /claim

为操作设置门控。首次调用返回 PROCEED，任何重复调用均返回 SKIP。

请求正文：

{
  "request_id": "order:TQQQ:buy:6:2026-05-19T13:31:00-04:00",
  "action": "order"
}

• request_id — 根据 Agent 正在执行的操作生成的稳定标识符。相同输入 = 相同键。
• action — 设置门控的操作类型（例如 payment.send、email、trade、order）

响应：

{ "status": "PROCEED", "request_id": "order:TQQQ:buy:6:2026-05-19T13:31:00-04:00" }

使用相同的负载重试：

{ "status": "SKIP", "request_id": "order:TQQQ:buy:6:2026-05-19T13:31:00-04:00", "existing": {...} }

POST /settle/{request_id}

操作执行后结算声明。将状态从 PENDING 推进到 COMMITTED。

curl -s -X POST https://safeagent-production.up.railway.app/settle/order:TQQQ:buy:6:2026-05-19T13:31:00-04:00 \
  -H "Content-Type: application/json" \
  -d '{"result":{"status":"completed","txid":"abc123"}}'

{ "status": "committed", "request_id": "..." }

POST /claim/test

免费测试端点——逻辑相同，无需付款。每个 IP 总共最多调用 10 次。

curl -s -X POST https://safeagent-production.up.railway.app/claim/test \
  -H "Content-Type: application/json" \
  -d '{"agent_id":"bot-1","action_type":"order","scope":"TQQQ:buy:6:bar:2026-05-19T13:31:00-04:00"}'

{ "status": "PROCEED", "request_id": "...", "test": true, "calls_remaining": 9 }

GET /audit

完整的声明历史。可按 agent_id、action、status 或时间戳范围筛选。

curl "https://safeagent-production.up.railway.app/audit?agent_id=bot-1&status=COMMITTED"

{"items": [...], "total": 5, "limit": 100, "offset": 0}

参数：agent_id、action、status、from_ts、to_ts、limit（最大值为 1000）、offset。

---

集成

Python

import requests

def claim(request_id: str, action: str) -> dict:
    r = requests.post(
        "https://safeagent-production.up.railway.app/claim",
        headers={"Content-Type": "application/json"},
        json={"request_id": request_id, "action": action}
    )
    return r.json()  # {"status": "PROCEED"|"SKIP", "request_id": "..."}

def settle(request_id: str, result: dict) -> dict:
    r = requests.post(
        f"https://safeagent-production.up.railway.app/settle/{request_id}",
        headers={"Content-Type": "application/json"},
        json={"result": result}
    )
    return r.json()

# Usage
response = claim("payment:customer-123:invoice-456", "payment.send")
if response["status"] == "PROCEED":
    result = send_payment(...)
    settle("payment:customer-123:invoice-456", {"status": "completed"})
elif response["status"] == "SKIP":
    print("Already executed, skipping")

SaaS（Stripe／邮件／Webhook）

对于生产环境中的 Stripe PaymentIntents，请使用文档中介绍的、由许可凭证门控的适配器： SafeAgent Boundary。 它会将 Stripe 凭证保留在 Agent 之外，将已消费的许可凭证 ID 作为原生幂等键传递给 Stripe，持久化提供商关联信息，验证 Webhook 签名，并支持权威回读。下面的简单声明／结算代码片段仅用于说明通用 API；它不是提供商适配器。

请运行范围受限的测试模式示例： examples/stripe_payment_intent_guard.py。 验证记录记录了观察到的端到端结果：一个成功创建的 Stripe 沙盒 PaymentIntent、在分发到提供商之前被阻止的完全相同重放请求，以及通过 retrieve 进行的权威协调核对。

import requests

def safe_stripe_charge(customer_id, amount, idempotency_key):
    request_id = f"stripe:charge:{customer_id}:{idempotency_key}"

    response = requests.post(
        "https://safeagent-production.up.railway.app/claim",
        headers={"Content-Type": "application/json"},
        json={"request_id": request_id, "action": "stripe_charge"}
    ).json()

    if response["status"] == "SKIP":
        return response.get("existing")

    charge = stripe.PaymentIntent.create(
        amount=amount,
        currency="usd",
        customer=customer_id,
        idempotency_key=request_id,
    )
    requests.post(
        f"https://safeagent-production.up.railway.app/settle/{request_id}",
        headers={"Content-Type": "application/json"},
        json={"result": {"charge_id": charge.id}}
    )
    return charge

Webhook 去重 — 稳定的事件 ID 可在结果结算后抑制重复处理；尚未解决的 PENDING 事件会继续被阻止，以便进行调查：

def handle_stripe_webhook(event):
    response = requests.post(
        "https://safeagent-production.up.railway.app/claim",
        headers={"Content-Type": "application/json"},
        json={"request_id": event["id"], "action": "stripe_event"}
    ).json()

    if response["status"] == "SKIP":
        return {"ok": True}

    provision_subscription(event)
    requests.post(
        f"https://safeagent-production.up.railway.app/settle/{event['id']}",
        headers={"Content-Type": "application/json"},
        json={"result": {"processed": True}}
    )

CrewAI

from crewai import Agent
import sqlite3

guard_con = sqlite3.connect("safeagent_orders.db", check_same_thread=False)
# ... same guard pattern, keyed on tool_name + input hash + session_id

PR crewAIInc/crewAI#5822 添加了可插拔的幂等性后端。SafeAgent 的 SQLite 架构与其兼容。

WisePick

WisePick 是面向 Agent 运行时的确定性路由层。此集成将能力选择与持久化、幂等执行拆分开来——WisePick 决定做什么以及使用哪个提供方，SafeAgent 则判断这项逻辑工作是否已执行过。

技术栈：

WisePick /v1/decide → DashClaw → SafeAgent → Mycelium Trails → Base/Arbitrum

• 适配器：adapters/safeagent_adapter.py
• 集成文档：docs/integrations/safeagent.md
• 集成讨论：SafeAgent #9

---

本地防护（嵌入式 SQLite）

无需网络调用，即可实现相同的持久化声明行为。将以下代码放入任意 Python Agent 中：

import json
import sqlite3

_SA_DB = "safeagent_orders.db"
_sa_con = sqlite3.connect(_SA_DB, check_same_thread=False)
_sa_con.execute("""CREATE TABLE IF NOT EXISTS orders (
    request_id TEXT PRIMARY KEY,
    result     TEXT,
    status     TEXT DEFAULT 'PENDING',
    created_at TEXT DEFAULT (datetime('now'))
)""")
_sa_con.commit()

def place_order_with_guard(symbol, qty, side, bar_ts):
    request_id = f"order:{symbol}:{side}:{qty}:{bar_ts}"

    cur = _sa_con.execute(
        "INSERT OR IGNORE INTO orders (request_id, status) VALUES (?, 'PENDING')",
        (request_id,)
    )
    _sa_con.commit()

    if cur.rowcount == 0:
        row = _sa_con.execute(
            "SELECT status, result FROM orders WHERE request_id = ?",
            (request_id,)
        ).fetchone()
        if row and row[0] == 'COMMITTED':
            print(f"SAFEAGENT SKIP: {request_id}")
            return row[1]
        raise RuntimeError(
            f"SAFEAGENT BLOCKED: unresolved PENDING claim for {request_id}"
        )

    result = place_order(symbol, qty, side)

    _sa_con.execute(
        "UPDATE orders SET status='COMMITTED', result=? WHERE request_id=?",
        (json.dumps(str(result)), request_id)
    )
    _sa_con.commit()
    return result

---

案例研究：防止崩溃重试导致重复操作

一个交易机器人发出市价单，买入 6 股 TQQQ。经纪商接受了订单。机器人在更新状态前崩溃。重启后——信号相同，K 线相同——机器人再次发出订单。经纪商成交了两次。Agent 现在持有 12 股，而它原本只打算持有 6 股。

这并非纸上谈兵。在提交订单与持久化状态之间发生任何未处理的异常时，都可能出现这种情况。

SafeAgent 会在联系经纪商前生成稳定的键：

request_id = "order:TQQQ:buy:6:2026-05-19T13:31:00-04:00"

首次调用：PROCEED。触发副作用。settle 写入 COMMITTED。
崩溃并重试：SKIP。不再联系经纪商。

---

案例研究：实时阻止重复操作——2026 年 5 月 21 日

完整技术栈在实时会话中确认了 6 次 SKIP 事件：DashClaw、SafeAgent、Mycelium Trails、Base/Arbitrum、经纪商 Alpaca。

• 0942 ET：阻止重复买入 TQQQ，数量=6，$452
• 0947 ET：阻止重复加仓 TQQQ，数量=6，$452
• 0949 ET：反转时阻止重复卖出 TQQQ，数量=12，$902
• 1000 ET：阻止重复买入 TQQQ，数量=6，$454
• 1014 ET：V20 反转时阻止重复卖出 TQQQ，数量=18，$1,350
• 1106 ET：分批建仓期间阻止重复加仓 SQQQ，$43

完整会话数据：https://gist.github.com/azender1/b9112b6519c935df4a75cb05cd250e26

---

一致性验证

SafeAgent 参与了 A2A 跨实现一致性验证语料库。四个独立实现的夹具均通过字节级验证：

• kenneives (agentgraph) — verifier-attestation-v0：30/30 ✓
• evidai (LemonCake) — gated-preflight-v1：33/33 ✓
• haroldmalikfrimpong-ops (agentid) — 独立验证器 ✓
• SafeAgent — claim-lifecycle conformance v1.1 ✓

A2A #1920 已关闭。四个实现均满足新鲜度和重放要求。

---

规范的 action_ref 派生

JCS (RFC 8785)，与 argentum-core、Nobulex 和 APS 保持一致：

import hashlib
import rfc8785

def compute_action_ref(
    agent_id: str,
    action_type: str,
    scope: str,
    timestamp: str,  # RFC 3339 UTC, 3-digit ms: "2026-05-15T10:00:00.123Z"
) -> str:
    payload = {
        "agent_id": agent_id,
        "action_type": action_type,
        "scope": scope,
        "timestamp": timestamp,
    }
    canonical = rfc8785.dumps(payload)
    return hashlib.sha256(canonical).hexdigest()

timestamp 是 RFC 3339 UTC 字符串，且必须恰有 3 位毫秒数字。末尾的 Z 为必填项。

一致性验证夹具：giskard09/argentum-core tag action-ref-v1.0

---

部署

Railway · Serverless OFF · 始终运行
PyPI：pip install safeagent-exec-guard
npm：npm install n8n-nodes-safeagent
MCP Registry：io.github.azender1/safeagent
仪表板：safeagent-dashboard-2.vercel.app

---

许可证

Apache-2.0