工具检查拦在了哪一步：把 Guardrail 和副作用排到同一条时间线上

给一个写入工具加上输出检查，看起来很稳妥：结果不合格，就抛出异常。直到你打开业务表，发现刚才生成的记录还在那里。异常发生时，函数已经做完了写入；检查挡住的是返回值继续被使用的那一步。

这篇从这个时间差开始。我们用一个原创的离线草稿账本，把参数校验、输入检查、业务函数、输出检查和审计回执排在同一条线上。你会亲手让后检拒绝、让调用在提交后取消，再查询账本，确认到底留下了什么。程序不使用 Agents SDK，不请求模型，不调用真实 API，也没有发信、付款或远程写入。唯一业务效果是在临时 SQLite 文件里增加一条虚构草稿，演示与测试结束后临时目录自动清理。

本文假设你能运行 Python 文件，了解函数和异常。只需 Python 3.10 以上及标准库；实际在 Python 3.12.14 验证。代码、输入和测试都在本页展开。运行前后不需要配置账号或密钥。

这里固定核对 OpenAI Agents SDK Python v0.23.1，对应提交 c4a1d047b22488234b2ba81056e1b9b8db2127a5。版本是为了让结论能复查，并不表示它永远是最新版。该版本 Guardrails 文档区分 Agent 级输入、最终输出，以及函数工具调用前后的检查。

Agent 的 input guardrail 只在链路的第一个 Agent 上检查初始输入。默认 run_in_parallel=True，检查和 Agent 并行推进；等 tripwire 触发时，模型可能已消耗 token，工具也可能已执行。改成 run_in_parallel=False，会等输入检查完成再启动 Agent；若此时拒绝，Agent 不执行。这个差异直接影响“前置”两个字到底能承诺什么。

Agent 的 output guardrail 针对产生最终输出的 Agent，发生在最终输出产生之后，没有 run_in_parallel 选项。它不能代替每次工具调用的参数校验。若一次运行已经使用多个工具，最终回答被拒绝也不会让这些调用自动倒着执行。文档还区分输出 tripwire 与检查函数自身抛异常的会话保存行为；会话里保留或替换哪些消息，与业务系统里撤销一笔操作，是两件分别要处理的事。

工具级检查挂在 FunctionTool 执行链上。输入检查可放行、返回拒绝消息以跳过此次调用，或抛出工具输入 tripwire；输出检查拿到已产生的结果后，可放行、替换结果为拒绝消息，或抛出工具输出 tripwire。tool_guardrails.py定义 allow、reject_content、raise_exception 三种行为。本文的 input_rejected、output_rejected 等名称是离线实验自己的回执字段，并非 SDK 状态名。

实际调用位置可沿 tool_execution.py 的输入检查入口向下追踪：拒绝时返回，不进入正常业务调用；正常调用产生 real_result 后，才进入 输出检查入口。这说明后检拥有的是结果处理时机。不能把 tripwire 当成数据库事务回滚指令，更不能期待它撤销已被另一家服务接受的请求。若检查里用了模型，模型的判断仍然只是检查结果；回滚能力来自业务系统本身。

同一版本文档说明，FunctionTool 可直接配置工具护栏；本地 MCP server 配置的护栏列表会附到其暴露的工具上。handoff 调用走另一条管线。托管工具，以及 ComputerTool、ShellTool、ApplyPatchTool、LocalShellTool 等内建执行工具，不走这条工具护栏管线；Agent.as_tool() 也未直接暴露工具护栏选项。把“工具”都写成一个名字，会掩盖这些差别。接入前应逐个确认实际工具类型，不能从本文推导出所有能力已经受保护。

如果函数工具需要人工批准，输入护栏通常在批准后、执行前运行。该版本可用 ToolExecutionConfig(pre_approval_tool_input_guardrails=True) 在待批准中断产生前先检查一次；通过后，批准完成仍会在执行前重检。这里点到时间位置为止，不展开批准的保存、撤销或消费规则。相关行为同样见 固定版本的 Tool guardrails 说明。

下面实验只研究单次串行调用。它不仿制 SDK 调度器、不模拟模型分类、不实现用户认证或审批。参数形状检查和业务上限检查都是确定性 Python 函数，因此测试能精确回答发生先后，不能据此宣称真实模型的安全识别率。

草稿请求只有三个字段：job 是 JOB-001 这样的本地业务键，topic 只能选 river 或 forest，copies 是整数。结构校验允许 1 到 9，接着业务输入检查把单次上限收紧到 3。这样 copies=4 能通过结构校验，再在业务规则处被挡住，两个阶段不会混在一起。布尔值即使在 Python 中属于整数的子类，也明确不接受。

business 函数把固定模板“river / copies=2”存入 drafts 表。job 是主键，digest 由规范化请求计算。同键、同参数会读回旧草稿；同键换参数则报 key_conflict。这个键是我们定义的业务去重键，与 SDK 的一次 tool_call_id 无关。调用方若每次重试都生成新 job，就会得到新草稿。

Faults 是测试端传入的故障开关，故意与工具参数分开。输入里的 skip_checks 等额外字段会直接被拒绝。bad_output 只污染函数返回副本，不改已经保存的模板内容，方便看清“存下来的草稿”和“准备交给调用方的结果”可以分别检查。

在一个新目录里建立 timing_lab.py、demo.py、test_timing_lab.py，把后面三个代码块按文件名完整复制。在这三个文件所在的目录运行以下命令，不需要 pip install。

python --version
python demo.py
python -m unittest -v

with self.db 包住的短事务只有本地草稿写入。事务内异常使未提交的行回滚；离开 with 后追加 effect_committed 标记，表示提交已完成。输出检查在它后面运行。审计 receipt 另开一次事务，记录这次尝试的状态和阶段标记，不保存参数全文或被拒绝的内容。

effect_exists 回答“这个 job 当前有没有草稿”。它不表示一定是当前尝试新写入：同键复用、参数冲突也可能看到旧行。判断本次行为，要同时看 effect_committed 或 effect_reused。已保存的回执按 attempt 编号，一条草稿可以对应多次尝试。这个编号不是完整调用次数；回执未写入的尝试不会因此得到一条可查询的记录。

"""Original offline timing experiment; no SDK, network, model or external tools."""
import asyncio
import hashlib
import json
import re
import sqlite3
from dataclasses import dataclass


class InputRejected(Exception):
    pass


class OutputRejected(Exception):
    pass


class Conflict(Exception):
    pass


class AuditUnavailable(Exception):
    pass


@dataclass(frozen=True)
class Faults:
    # Trusted test harness only. These fields are never tool arguments.
    cancel_at: str = ""
    fail_at: str = ""
    bad_output: bool = False
    audit_failure: bool = False


def checked_args(raw):
    if type(raw) is not dict or set(raw) != {"job", "topic", "copies"}:
        raise InputRejected("shape")
    if type(raw["job"]) is not str or not re.fullmatch(r"JOB-[0-9]{3}", raw["job"]):
        raise InputRejected("job")
    if type(raw["topic"]) is not str or raw["topic"] not in ("river", "forest"):
        raise InputRejected("topic")
    if type(raw["copies"]) is not int or not 1 <= raw["copies"] <= 9:
        raise InputRejected("copies")
    return dict(raw)


def input_check(args):
    # Application rule, separate from structural validation.
    if args["copies"] > 3:
        raise InputRejected("batch_limit")


def output_check(value, args):
    expected = f"{args['topic']} / copies={args['copies']}"
    if (type(value) is not dict or set(value) != {"job", "text"}
            or value["job"] != args["job"] or value["text"] != expected):
        raise OutputRejected("output_contract")


def fingerprint(args):
    wire = json.dumps(args, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(wire.encode()).hexdigest()


class Lab:
    def __init__(self, database):
        self.db = sqlite3.connect(database)
        self.db.executescript("""
            CREATE TABLE IF NOT EXISTS drafts(
                job TEXT PRIMARY KEY, digest TEXT NOT NULL, body TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS receipts(
                attempt INTEGER PRIMARY KEY AUTOINCREMENT,
                job TEXT, status TEXT NOT NULL, effect_exists INTEGER NOT NULL,
                events TEXT NOT NULL);
        """)

    def close(self):
        self.db.close()

    def count(self):
        return self.db.execute("SELECT COUNT(*) FROM drafts").fetchone()[0]

    def exists(self, job):
        return bool(self.db.execute("SELECT 1 FROM drafts WHERE job=?", (job,)).fetchone())

    async def checkpoint(self, name, faults):
        if faults.cancel_at == name:
            raise asyncio.CancelledError()
        if faults.fail_at == name:
            raise RuntimeError(name)
        # A real cancellation request can also be delivered at this await.
        await asyncio.sleep(0)

    def business(self, args, events, faults):
        digest = fingerprint(args)
        with self.db:
            old = self.db.execute(
                "SELECT digest, body FROM drafts WHERE job=?", (args["job"],)
            ).fetchone()
            if old:
                if old[0] != digest:
                    raise Conflict("same_key_different_arguments")
                events.append("effect_reused")
                return json.loads(old[1])
            value = {"job": args["job"],
                     "text": f"{args['topic']} / copies={args['copies']}"}
            self.db.execute("INSERT INTO drafts VALUES(?,?,?)",
                            (args["job"], digest, json.dumps(value)))
            if faults.fail_at == "inside_transaction":
                raise RuntimeError("inside_transaction")
        # The SQLite transaction has committed when this marker is added.
        events.append("effect_committed")
        if faults.fail_at == "after_commit":
            raise RuntimeError("after_commit")
        return value

    def receipt(self, job, status, events, faults):
        # Audit is a later, separate local transaction. It cannot undo drafts.
        if faults.audit_failure:
            raise AuditUnavailable("audit write unavailable")
        present = self.exists(job) if job is not None else False
        with self.db:
            cursor = self.db.execute(
                "INSERT INTO receipts(job,status,effect_exists,events) VALUES(?,?,?,?)",
                (job, status, int(present), json.dumps(events)))
            # A no-op INSERT can leave lastrowid unchanged; require a new row.
            if cursor.rowcount != 1:
                raise AuditUnavailable("audit insert did not create one row")
        return {"attempt": cursor.lastrowid, "job": job, "status": status,
                "effect_exists": present, "events": list(events)}

    async def run(self, raw, faults=Faults()):
        events, job, status, released = [], None, "error", None
        try:
            args = checked_args(raw)
            job = args["job"]
            events.append("arguments_valid")
            input_check(args)
            events.append("input_allowed")
            await self.checkpoint("before_business", faults)
            events.append("business_started")
            value = self.business(args, events, faults)
            await self.checkpoint("after_business", faults)
            events.append("output_started")
            if faults.bad_output:
                value = {**value, "text": "SYNTHETIC-BAD-OUTPUT"}
            await self.checkpoint("output_check", faults)
            output_check(value, args)
            events.append("output_allowed")
            status, released = "accepted", value
        except InputRejected:
            status = "input_rejected"
        except OutputRejected:
            status = "output_rejected"
        except Conflict:
            status = "key_conflict"
        except asyncio.CancelledError:
            # Preserve cancellation semantics; do not convert it to success.
            self.receipt(job, "cancelled", events, faults)
            raise
        except Exception:
            status = "error"
        result = self.receipt(job, status, events, faults)
        result["released"] = released
        return result

这是教学用串行实现，所有已知输入都来自本页夹具。未知异常被折叠成 error，避免把异常内容当成合格输出；生产中还应有经过脱敏的运维诊断。本例用 audit_failure 注入 AuditUnavailable；真实的 SQLite 回执写入错误则直接以数据库异常向外传播，都不会返回 accepted。已经提交的草稿仍在。取消路径上的回执写入失败也会让该审计错误优先冒出；下文同时验证预写入故障开关和 SQLite 执行时的失败。原取消异常仍可留在异常链中，不能因为最外层是审计错误就认定取消从未发生。

没有抛异常，也未必写入了回执

原实现只取 lastrowid，没有检查这次 INSERT 是否实际新增一行。用仅存在于临时测试库的 BEFORE INSERT 触发器执行 RAISE(IGNORE)，可以让 SQLite 跳过回执写入而不抛异常。修正前实测返回 status=accepted、attempt=1 和 released 内容，但另一条连接读到的账本是 1 条草稿、0 条回执。这里复现的是教学代码缺少插入结果检查，不是 SDK 或线上服务的漏洞。

SQLite 的 RAISE 说明区分 IGNORE 与返回错误的 ABORT；Python 的 lastrowid 说明也不能作为“这次写入成功”的独立凭据。本例使用普通 INSERT，无 RETURNING、无 CTE，因此在语句执行完毕后要求 rowcount 恰为 1。否则就在回执事务内部抛出 AuditUnavailable，不返回带回执的结果；数据库本身抛出的异常仍原样传播。

这个检查不会撤销前一个事务已经提交的草稿。恢复测试仅移除可信测试端安装的故障触发器，随后以原 job 和原参数重试，复用已有草稿并重新检查输出，最终只有 1 条草稿和 1 条回执。不要把触发器故障夹具当作业务操作；本实验也没有验证任意触发器篡改、磁盘故障或持久化审计系统。

演示依次执行三次 JOB-001 请求。第一次 copies=4，在前检拒绝；第二次提交草稿后污染返回值，后检拒绝；第三次用相同参数重试，复用原行并重新检查输出。最后为 JOB-002 注入提交后的取消。

import asyncio
import json
import tempfile
from pathlib import Path
from timing_lab import Lab, Faults


async def main():
    # Fresh local files, removed on exit; no user directory is touched.
    with tempfile.TemporaryDirectory(prefix="guardrail-timing-") as folder:
        lab = Lab(Path(folder) / "lab.sqlite")
        args = {"job": "JOB-001", "topic": "river", "copies": 2}
        try:
            for label, raw, faults in [
                ("input_rejected", {**args, "copies": 4}, Faults()),
                ("output_rejected", args, Faults(bad_output=True)),
                ("retry_same_request", args, Faults()),
            ]:
                print(label)
                print(json.dumps(await lab.run(raw, faults), ensure_ascii=False))
            try:
                await lab.run({**args, "job": "JOB-002"},
                              Faults(cancel_at="after_business"))
            except asyncio.CancelledError:
                print("cancelled_after_business: effect_exists="
                      + str(lab.exists("JOB-002")))
            print("draft_rows=" + str(lab.count()))
        finally:
            lab.close()


if __name__ == "__main__":
    asyncio.run(main())

实际输出如下。第二次的 released 为 null，而 effect_exists 为 true；第三次出现 effect_reused。最后草稿总数为 2，对应两个不同的 job。

input_rejected
{"attempt": 1, "job": "JOB-001", "status": "input_rejected", "effect_exists": false, "events": ["arguments_valid"], "released": null}
output_rejected
{"attempt": 2, "job": "JOB-001", "status": "output_rejected", "effect_exists": true, "events": ["arguments_valid", "input_allowed", "business_started", "effect_committed", "output_started"], "released": null}
retry_same_request
{"attempt": 3, "job": "JOB-001", "status": "accepted", "effect_exists": true, "events": ["arguments_valid", "input_allowed", "business_started", "effect_reused", "output_started", "output_allowed"], "released": {"job": "JOB-001", "text": "river / copies=2"}}
cancelled_after_business: effect_exists=True
draft_rows=2

这里用本地事务人为确定了提交点，所以我们能够直接查询数据库确定结果。真实远程调用若在发送后超时，应用可能不知道对方是否接受请求，应先按稳定业务键查询、对账或使用对方的幂等能力。仅凭一个超时异常就再次发送，可能造成重复效果。

有三个容易混淆的结果。检查明确返回拒绝，代表拿到了否定结论；检查函数抛异常，代表没有顺利得到结论；任务取消，代表后续执行被请求停止。本例都不把待检内容当合格输出，但保留各自状态。SDK 的真实异常传播还会受工具错误处理器、运行方式和取消入口影响，不能把本例的 catch 分支搬过去就声称语义相同。

下面 30 项测试既断言返回状态，也直接查询草稿表。取消测试大部分通过 Faults 精确注入 CancelledError，另有 test_real_task_cancel_after_commit 使用 asyncio.Task.cancel()：等待业务提交后，在协程等待点真正发送取消，再检查行仍存在。这个测试验证 Python 本地协作取消；它没有验证 SDK、网络请求或远端服务的取消行为。

import asyncio
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from timing_lab import Lab, Faults, AuditUnavailable


class TimingTests(unittest.IsolatedAsyncioTestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.path = Path(self.tmp.name) / "lab.sqlite"
        self.lab = Lab(self.path)
        self.args = {"job": "JOB-001", "topic": "river", "copies": 2}

    def tearDown(self):
        self.lab.close()
        self.tmp.cleanup()

    def install_receipt_fault(self, action):
        # Trusted fixture on this test's temporary database, never tool input.
        if action == "ignore":
            statement = "SELECT RAISE(IGNORE);"
        elif action == "abort":
            statement = "SELECT RAISE(ABORT, 'fixture_receipt_abort');"
        else:
            raise ValueError("unknown fixture")
        self.lab.db.execute("CREATE TRIGGER fixture_receipt_fault "
                            "BEFORE INSERT ON receipts BEGIN " + statement + " END")

    def persisted_counts(self):
        # A separate connection verifies committed state, not a pending write.
        observer = sqlite3.connect(self.path)
        try:
            return (observer.execute("SELECT COUNT(*) FROM drafts").fetchone()[0],
                    observer.execute("SELECT COUNT(*) FROM receipts").fetchone()[0])
        finally:
            observer.close()

    async def test_sqlite_ignored_receipt_and_same_key_recovery(self):
        self.install_receipt_fault("ignore")
        with self.assertRaisesRegex(AuditUnavailable, "audit insert"):
            await self.lab.run(self.args)
        self.assertEqual(self.persisted_counts(), (1, 0))
        self.assertFalse(self.lab.db.in_transaction)
        # Remove only the fixture fault; do not rewrite the business request.
        self.lab.db.execute("DROP TRIGGER fixture_receipt_fault")
        result = await self.lab.run(self.args)
        self.assertEqual(result["status"], "accepted")
        self.assertIn("effect_reused", result["events"])
        self.assertNotIn("effect_committed", result["events"])
        self.assertIn("output_allowed", result["events"])
        self.assertEqual(result["released"], {
            "job": "JOB-001", "text": "river / copies=2"})
        self.assertEqual(self.persisted_counts(), (1, 1))

    async def test_sqlite_ignored_receipt_cannot_reuse_old_receipt(self):
        first = await self.lab.run(self.args)
        self.install_receipt_fault("ignore")
        with self.assertRaisesRegex(AuditUnavailable, "audit insert"):
            await self.lab.run({**self.args, "job": "JOB-002"})
        self.assertEqual(self.persisted_counts(), (2, 1))
        self.assertEqual(self.lab.db.execute(
            "SELECT attempt,job,status FROM receipts").fetchall(),
            [(first["attempt"], "JOB-001", "accepted")])
        self.assertFalse(self.lab.db.in_transaction)

    async def test_sqlite_aborted_receipt_preserves_committed_effect(self):
        self.install_receipt_fault("abort")
        with self.assertRaisesRegex(sqlite3.IntegrityError, "fixture_receipt_abort"):
            await self.lab.run(self.args)
        self.assertEqual(self.persisted_counts(), (1, 0))
        self.assertFalse(self.lab.db.in_transaction)

    async def test_sqlite_ignored_receipt_during_cancel_takes_precedence(self):
        self.install_receipt_fault("ignore")
        with self.assertRaises(AuditUnavailable) as caught:
            await self.lab.run(self.args, Faults(cancel_at="after_business"))
        self.assertIsInstance(caught.exception.__context__, asyncio.CancelledError)
        self.assertEqual(self.persisted_counts(), (1, 0))
        self.assertFalse(self.lab.db.in_transaction)

    async def test_success_order(self):
        r = await self.lab.run(self.args)
        self.assertEqual(r["status"], "accepted")
        self.assertEqual(r["events"], ["arguments_valid", "input_allowed",
            "business_started", "effect_committed", "output_started", "output_allowed"])
        self.assertTrue(r["effect_exists"])
        self.assertEqual(r["released"], {"job": "JOB-001", "text": "river / copies=2"})

    async def test_missing_field(self):
        r = await self.lab.run({"job": "JOB-001"})
        self.assertEqual(r["status"], "input_rejected")
        self.assertEqual(self.lab.count(), 0)

    async def test_extra_field(self):
        r = await self.lab.run({**self.args, "skip_checks": True})
        self.assertEqual(r["status"], "input_rejected")
        self.assertEqual(self.lab.count(), 0)

    async def test_bool_is_not_count(self):
        r = await self.lab.run({**self.args, "copies": True})
        self.assertEqual(r["status"], "input_rejected")

    async def test_invalid_job(self):
        r = await self.lab.run({**self.args, "job": "../report"})
        self.assertEqual(r["status"], "input_rejected")

    async def test_unknown_topic(self):
        r = await self.lab.run({**self.args, "topic": "unknown"})
        self.assertEqual(r["status"], "input_rejected")

    async def test_batch_policy(self):
        r = await self.lab.run({**self.args, "copies": 4})
        self.assertEqual(r["events"], ["arguments_valid"])
        self.assertEqual(r["status"], "input_rejected")
        self.assertFalse(r["effect_exists"])

    async def test_output_rejection_preserves_effect(self):
        r = await self.lab.run(self.args, Faults(bad_output=True))
        self.assertEqual(r["status"], "output_rejected")
        self.assertTrue(r["effect_exists"])
        self.assertIsNone(r["released"])
        self.assertNotIn("output_allowed", r["events"])

    async def test_before_business_exception(self):
        r = await self.lab.run(self.args, Faults(fail_at="before_business"))
        self.assertEqual(r["status"], "error")
        self.assertFalse(r["effect_exists"])

    async def test_local_transaction_rollback(self):
        r = await self.lab.run(self.args, Faults(fail_at="inside_transaction"))
        self.assertEqual(r["status"], "error")
        self.assertFalse(r["effect_exists"])
        self.assertNotIn("effect_committed", r["events"])

    async def test_after_commit_exception(self):
        r = await self.lab.run(self.args, Faults(fail_at="after_commit"))
        self.assertEqual(r["status"], "error")
        self.assertTrue(r["effect_exists"])
        self.assertNotIn("output_started", r["events"])

    async def test_output_check_exception(self):
        r = await self.lab.run(self.args, Faults(fail_at="output_check"))
        self.assertEqual(r["status"], "error")
        self.assertTrue(r["effect_exists"])
        self.assertIsNone(r["released"])

    async def test_cancel_before_business(self):
        with self.assertRaises(asyncio.CancelledError):
            await self.lab.run(self.args, Faults(cancel_at="before_business"))
        self.assertEqual(self.lab.count(), 0)
        self.assertEqual(self.lab.db.execute("SELECT status FROM receipts").fetchone()[0],
                         "cancelled")

    async def test_cancel_after_business(self):
        with self.assertRaises(asyncio.CancelledError):
            await self.lab.run(self.args, Faults(cancel_at="after_business"))
        self.assertEqual(self.lab.count(), 1)

    async def test_cancel_during_output_check(self):
        with self.assertRaises(asyncio.CancelledError):
            await self.lab.run(self.args, Faults(cancel_at="output_check"))
        self.assertEqual(self.lab.count(), 1)

    async def test_idempotent_retry_after_rejection(self):
        await self.lab.run(self.args, Faults(bad_output=True))
        r = await self.lab.run(self.args)
        self.assertEqual(self.lab.count(), 1)
        self.assertIn("effect_reused", r["events"])
        self.assertIn("output_allowed", r["events"])
        self.assertEqual(r["attempt"], 2)

    async def test_retry_runs_output_check_again(self):
        await self.lab.run(self.args)
        r = await self.lab.run(self.args, Faults(bad_output=True))
        self.assertEqual(r["status"], "output_rejected")
        self.assertIsNone(r["released"])
        self.assertEqual(self.lab.count(), 1)

    async def test_same_key_conflict(self):
        await self.lab.run(self.args)
        r = await self.lab.run({**self.args, "copies": 3})
        self.assertEqual(r["status"], "key_conflict")
        self.assertEqual(self.lab.count(), 1)
        self.assertIsNone(r["released"])

    async def test_new_key_is_new_effect(self):
        await self.lab.run(self.args)
        await self.lab.run({**self.args, "job": "JOB-002"})
        self.assertEqual(self.lab.count(), 2)

    async def test_restart_same_database(self):
        await self.lab.run(self.args, Faults(fail_at="after_commit"))
        self.lab.close()
        self.lab = Lab(self.path)
        r = await self.lab.run(self.args)
        self.assertIn("effect_reused", r["events"])
        self.assertEqual(self.lab.count(), 1)

    async def test_audit_failure_never_claims_success(self):
        with self.assertRaises(AuditUnavailable):
            await self.lab.run(self.args, Faults(audit_failure=True))
        self.assertEqual(self.lab.count(), 1)
        self.assertEqual(self.lab.db.execute("SELECT COUNT(*) FROM receipts").fetchone()[0], 0)

    async def test_real_task_cancel_after_commit(self):
        reached = asyncio.Event()
        wait_forever = asyncio.Event()
        original = self.lab.checkpoint

        async def pause(name, faults):
            if name == "after_business":
                reached.set()
                await wait_forever.wait()
            await original(name, faults)

        self.lab.checkpoint = pause
        task = asyncio.create_task(self.lab.run(self.args))
        await asyncio.wait_for(reached.wait(), timeout=2)
        task.cancel()
        with self.assertRaises(asyncio.CancelledError):
            await task
        self.assertEqual(self.lab.count(), 1)
        self.assertEqual(self.lab.db.execute("SELECT status FROM receipts").fetchone()[0],
                         "cancelled")

    async def test_zero_count(self):
        r = await self.lab.run({**self.args, "copies": 0})
        self.assertEqual(r["status"], "input_rejected")
        self.assertEqual(self.lab.count(), 0)

    async def test_non_string_topic(self):
        r = await self.lab.run({**self.args, "topic": ["river"]})
        self.assertEqual(r["status"], "input_rejected")

    async def test_audit_error_during_cancel_takes_precedence(self):
        with self.assertRaises(AuditUnavailable):
            await self.lab.run(self.args, Faults(cancel_at="after_business",
                                                audit_failure=True))
        self.assertEqual(self.lab.count(), 1)

    async def test_receipt_does_not_store_rejected_payload(self):
        await self.lab.run(self.args, Faults(bad_output=True))
        row = self.lab.db.execute("SELECT events FROM receipts").fetchone()[0]
        self.assertNotIn("SYNTHETIC-BAD-OUTPUT", row)
        self.assertIn("effect_committed", json.loads(row))


if __name__ == "__main__":
    unittest.main()

本节通过标准是 30 项全通过，演示最后输出 draft_rows=2。检查代码时尤其留意 test_local_transaction_rollback 与 test_after_commit_exception：相同 error 状态，一次没有行，一次有行。也要看 test_retry_runs_output_check_again：即使旧草稿已存在，重试仍经过本次输出检查，没有直接把旧结果放出去。

四项 test_sqlite_ 开头的新增测试把故障真正放进 SQLite：忽略插入后同键恢复、已有旧回执时拒绝冒用旧编号、ABORT 错误保留已提交效果、取消后的回执无写入错误。persisted_counts 使用另一个连接核对已提交行数，并检查失败后没有悬挂事务。只移除 rowcount 检查时，其中三项会失败；ABORT 测试仍通过，说明仅覆盖“数据库抛异常”不足以发现无异常的空写入。这里的取消仍是本地注入，真实 Task.cancel 的覆盖由原有独立测试提供。

前面的输入检查使用固定的 copies 上限，适合把各阶段分开观察。把上限改成业务表里的可变值，便出现另一个需要验收的空隙：前检看到上限为 3，允许生成 2 份草稿；真正写入前，上限已经改成 1。即使前检判断没有错误，继续照旧写入仍会违反提交时的规则。

这里补一份独立的小实验，不改原来的 timing_lab.py。连接 A 先读取规则，连接 B 提交规则修改，A 再尝试插入草稿。两个连接操作同一个临时 SQLite 文件，先后顺序由普通函数调用确定；没有线程竞速、sleep 或锁等待时长的假设。唯一业务效果仍是虚构的本地草稿行。

初始规则固定为 revision=1、max_copies=3，请求固定为 JOB-001、copies=2。对照实现 insert_naive 只信任旧结论；修正入口 insert_checked 同时要求当前修订号等于前检所见修订号、当前上限仍容纳本次 copies。这是一种刻意保守的约定：任何规则修改都要求重新检查，包括放宽上限和把值改回原样。修订号由受控业务写入口每次递增，不能把旧修订号恢复或复用。

2026-10-14 核对的 Agents SDK 官方文档仍把工具输入检查放在函数执行之前、输出检查放在函数执行之后；v0.23.1 发布页与原文版本一致。本补充只实现应用自己的提交条件，没有安装或执行 SDK，也没有调用模型。这不是对 SDK 漏洞或真实权限系统的测试。

将 policy_boundary.py 另存到前文三个文件的同一目录。它使用不同的临时数据库结构，不继承原 Lab。Snapshot 只是一次规则读取结果，既不是授权凭证，也不证明调用方身份。change_policy 是可信测试端的布置函数，不应作为模型可调用工具暴露。

关键在 INSERT … SELECT：当前规则不匹配时，SELECT 没有结果行，因而没有草稿被插入。这个语义来自 SQLite INSERT 文档。规则表与草稿表位于同一数据库；BEGIN IMMEDIATE 成功后，在同一写事务中判断条件并写入。SQLite 的写入隔离意味着其他连接不能在这次写事务里插入一次规则修改。不要把前检期间也持有写锁，等待模型或人工输入。

本例是普通 INSERT，没有 RETURNING 或 CTE；执行完毕后检查 Cursor.rowcount。等于 1 才尝试提交并返回 True；不等于 1 则回滚并返回 False。False 只表示这次条件写入没有匹配，不能单凭它断言“上限降低了”，因为修订号变化或规则行缺失也会得到相同结果。

连接使用显式事务，入口假定调用前没有未结束事务。锁忙可能让 BEGIN IMMEDIATE 报错；重复 job 产生约束错误；提交前注入异常则回滚。这些异常继续向调用方传播，不折叠成 False，也不假装成功。此小例未实现原文的幂等复用或审计回执，不能直接取代原 Lab。

policy_boundary.py

"""Local teaching fixture. No SDK, network, authentication, or external effect."""
import re
import sqlite3
from dataclasses import dataclass


class InputRejected(Exception):
    pass


@dataclass(frozen=True)
class Request:
    job: str
    copies: int

    def validate(self):
        if type(self.job) is not str or not re.fullmatch(r"JOB-[0-9]{3}", self.job):
            raise InputRejected("job")
        if type(self.copies) is not int or not 1 <= self.copies <= 9:
            raise InputRejected("copies")


@dataclass(frozen=True)
class Snapshot:
    revision: int
    max_copies: int


def connect(path):
    # Use explicit transactions; do not depend on Python's default mode.
    return sqlite3.connect(path, isolation_level=None, timeout=0)


def initialize(db):
    db.executescript("""
        CREATE TABLE policy(
            singleton INTEGER PRIMARY KEY CHECK(singleton=1),
            revision INTEGER NOT NULL CHECK(revision>0),
            max_copies INTEGER NOT NULL CHECK(max_copies BETWEEN 0 AND 9));
        INSERT INTO policy VALUES(1, 1, 3);
        CREATE TABLE drafts(
            job TEXT PRIMARY KEY, copies INTEGER NOT NULL,
            checked_revision INTEGER NOT NULL);
    """)


def change_policy(db, max_copies):
    """Trusted fixture setup only; each change advances the revision."""
    if type(max_copies) is not int or not 0 <= max_copies <= 9:
        raise ValueError("max_copies")
    db.execute("UPDATE policy SET revision=revision+1, max_copies=? "
               "WHERE singleton=1", (max_copies,))


def precheck(db, request):
    request.validate()
    row = db.execute("SELECT revision,max_copies FROM policy WHERE singleton=1").fetchone()
    if row is None:
        raise InputRejected("missing_policy")
    snapshot = Snapshot(*row)
    if request.copies > snapshot.max_copies:
        raise InputRejected("current_limit")
    return snapshot


def insert_naive(db, request, snapshot):
    """Deliberately wrong control: trusts an earlier check indefinitely."""
    request.validate()
    db.execute("INSERT INTO drafts VALUES(?,?,?)",
               (request.job, request.copies, snapshot.revision))


def insert_checked(db, request, snapshot, *, fail_before_commit=False):
    request.validate()
    if type(snapshot) is not Snapshot or type(snapshot.revision) is not int:
        raise ValueError("snapshot")
    db.execute("BEGIN IMMEDIATE")
    try:
        cursor = db.execute("""
            INSERT INTO drafts(job,copies,checked_revision)
            SELECT ?,?,revision FROM policy
            WHERE singleton=1 AND revision=? AND max_copies>=?
        """, (request.job, request.copies, snapshot.revision, request.copies))
        if cursor.rowcount != 1:
            db.rollback()
            return False
        if fail_before_commit:
            raise RuntimeError("injected_before_commit")
        db.commit()
        return True
    except BaseException:
        db.rollback()
        raise


def rows(db):
    return [list(row) for row in db.execute(
        "SELECT job,copies,checked_revision FROM drafts ORDER BY job")]

保存下面的 policy_demo.py 和 expected-policy.json。预期文件是依据固定输入手工写出的答案，不从程序输出自动更新。每一组 with fixture 都新建数据库；reader 和 editor 是两次真实 connect，查询结果的行依次表示 job、copies、checked_revision。

前两组使用相同变更 3→1。错误对照会留下 [JOB-001, 2, 1]；条件写入返回 False，草稿表为空。随后重新检查同一请求仍得到 current_limit。恢复示范由读者明确换成 JOB-002、copies=1，按当前 revision=2 重检后写入，结果只应有 [JOB-002, 1, 2]。这一步是新的业务输入，不是系统擅自减少份数，也没有把上限改回去来迁就旧请求。

policy_demo.py

"""Fixed real-SQLite interleavings; two connections, no timing assumptions."""
import json
import tempfile
from contextlib import contextmanager
from pathlib import Path
from policy_boundary import (Request, InputRejected, change_policy, connect,
                             initialize, insert_checked, insert_naive, precheck, rows)


@contextmanager
def fixture():
    with tempfile.TemporaryDirectory(prefix="guardrail-policy-") as folder:
        path = Path(folder) / "fixture.sqlite"
        reader, editor = connect(path), connect(path)
        try:
            initialize(reader)
            yield reader, editor
        finally:
            reader.close()
            editor.close()


def main():
    request = Request("JOB-001", 2)
    result = {}
    with fixture() as (reader, editor):
        seen = precheck(reader, request)
        change_policy(editor, 1)
        insert_naive(reader, request, seen)
        result["naive_stale"] = rows(editor)
    with fixture() as (reader, editor):
        seen = precheck(reader, request)
        change_policy(editor, 1)
        result["checked_stale"] = insert_checked(reader, request, seen)
        result["after_stale"] = rows(editor)
        try:
            precheck(reader, request)
        except InputRejected as error:
            result["fresh_check_same_request"] = str(error)
        # A deliberately new business input, not an automatic rewrite.
        smaller_request = Request("JOB-002", 1)
        fresh = precheck(reader, smaller_request)
        result["new_input_accepted"] = insert_checked(reader, smaller_request, fresh)
        result["after_recovery"] = rows(editor)
    with fixture() as (reader, editor):
        seen = precheck(reader, request)
        change_policy(editor, 1)
        change_policy(editor, 3)
        result["restored_value_old_revision"] = insert_checked(reader, request, seen)
        result["after_restored_value"] = rows(editor)
    with fixture() as (reader, editor):
        seen = precheck(reader, request)
        try:
            insert_checked(reader, request, seen, fail_before_commit=True)
        except RuntimeError as error:
            result["precommit_error"] = str(error)
        result["after_precommit_error"] = rows(editor)
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

expected-policy.json

{
  "naive_stale": [["JOB-001", 2, 1]],
  "checked_stale": false,
  "after_stale": [],
  "fresh_check_same_request": "current_limit",
  "new_input_accepted": true,
  "after_recovery": [["JOB-002", 1, 2]],
  "restored_value_old_revision": false,
  "after_restored_value": [],
  "precommit_error": "injected_before_commit",
  "after_precommit_error": []
}

运行 python -B policy_demo.py，输出 JSON 应与上面的预期语义完全一致。尤其核对 after_stale 和 after_precommit_error 都为空，而 after_recovery 仅包含新请求。程序退出时清理临时目录，不碰用户原有数据库。

第三组故意让上限经历 3→1→3；虽然数值回来了，修订号已经从 1 变成 3，旧 Snapshot 仍不得继续使用。第四组在 INSERT 之后、COMMIT 之前抛出异常，另一连接应查不到草稿。这里证明的是一次受控的本地事务回滚，没有测试断电、磁盘错误或远端写入。

保存 test_policy_boundary.py 后，在所有源码所在目录运行下面命令。前文主实验现有 30 项测试，本节另有 14 项测试覆盖旧规则反例、修订变化、当前上限复核、明确的新输入恢复、回滚、重复键、锁忙、规则缺失与固定答案。本次实际环境为 Python 3.12.14、SQLite 3.53.1，44 项全部通过；优化模式也复跑通过。

python -B policy_demo.py
python -B -m unittest -v test_timing_lab test_policy_boundary
python -B -O -m unittest -v test_timing_lab test_policy_boundary

test_policy_boundary.py

import contextlib
import io
import json
import sqlite3
import unittest
from pathlib import Path
from policy_boundary import (Request, Snapshot, InputRejected, change_policy,
                             insert_checked, insert_naive, precheck, rows)
from policy_demo import fixture, main


class PolicyBoundaryTests(unittest.TestCase):
    def setUp(self):
        self.scope = fixture()
        self.reader, self.editor = self.scope.__enter__()
        self.request = Request("JOB-001", 2)

    def tearDown(self):
        self.scope.__exit__(None, None, None)

    def test_unchanged_rule_commits(self):
        seen = precheck(self.reader, self.request)
        self.assertTrue(insert_checked(self.reader, self.request, seen))
        self.assertEqual(rows(self.editor), [["JOB-001", 2, 1]])

    def test_negative_control_accepts_stale_rule(self):
        seen = precheck(self.reader, self.request)
        change_policy(self.editor, 1)
        insert_naive(self.reader, self.request, seen)
        self.assertEqual(rows(self.editor), [["JOB-001", 2, 1]])

    def test_changed_rule_inserts_nothing(self):
        seen = precheck(self.reader, self.request)
        change_policy(self.editor, 1)
        self.assertFalse(insert_checked(self.reader, self.request, seen))
        self.assertEqual(rows(self.editor), [])
        self.assertFalse(self.reader.in_transaction)

    def test_fresh_check_still_refuses_same_input(self):
        change_policy(self.editor, 1)
        with self.assertRaisesRegex(InputRejected, "current_limit"):
            precheck(self.reader, self.request)
        self.assertEqual(rows(self.editor), [])

    def test_recovery_requires_new_valid_input(self):
        seen = precheck(self.reader, self.request)
        change_policy(self.editor, 1)
        self.assertFalse(insert_checked(self.reader, self.request, seen))
        smaller = Request("JOB-002", 1)
        fresh = precheck(self.reader, smaller)
        self.assertTrue(insert_checked(self.reader, smaller, fresh))
        self.assertEqual(rows(self.editor), [["JOB-002", 1, 2]])

    def test_restoring_value_does_not_restore_revision(self):
        seen = precheck(self.reader, self.request)
        change_policy(self.editor, 1)
        change_policy(self.editor, 3)
        self.assertFalse(insert_checked(self.reader, self.request, seen))
        self.assertEqual(rows(self.editor), [])

    def test_relaxed_rule_still_requires_fresh_check(self):
        seen = precheck(self.reader, self.request)
        change_policy(self.editor, 4)
        self.assertFalse(insert_checked(self.reader, self.request, seen))
        fresh = precheck(self.reader, self.request)
        self.assertTrue(insert_checked(self.reader, self.request, fresh))
        self.assertEqual(rows(self.editor), [["JOB-001", 2, 2]])

    def test_current_predicate_rechecks_changed_arguments(self):
        seen = precheck(self.reader, self.request)
        bigger = Request("JOB-001", 4)
        self.assertFalse(insert_checked(self.reader, bigger, seen))
        self.assertEqual(rows(self.editor), [])

    def test_precommit_failure_rolls_back(self):
        seen = precheck(self.reader, self.request)
        with self.assertRaisesRegex(RuntimeError, "injected_before_commit"):
            insert_checked(self.reader, self.request, seen, fail_before_commit=True)
        self.assertEqual(rows(self.editor), [])
        self.assertFalse(self.reader.in_transaction)

    def test_duplicate_key_is_error_not_rule_change(self):
        seen = precheck(self.reader, self.request)
        self.assertTrue(insert_checked(self.reader, self.request, seen))
        with self.assertRaises(sqlite3.IntegrityError):
            insert_checked(self.reader, self.request, seen)
        self.assertEqual(rows(self.editor), [["JOB-001", 2, 1]])
        self.assertFalse(self.reader.in_transaction)

    def test_busy_is_error_not_rule_change(self):
        seen = precheck(self.reader, self.request)
        self.editor.execute("BEGIN IMMEDIATE")
        try:
            with self.assertRaises(sqlite3.OperationalError):
                insert_checked(self.reader, self.request, seen)
            self.assertFalse(self.reader.in_transaction)
        finally:
            self.editor.rollback()
        self.assertEqual(rows(self.editor), [])

    def test_missing_policy_does_not_insert(self):
        seen = precheck(self.reader, self.request)
        self.editor.execute("DELETE FROM policy")
        self.assertFalse(insert_checked(self.reader, self.request, seen))
        self.assertEqual(rows(self.editor), [])

    def test_bad_request_shapes(self):
        for request in [Request("../x", 2), Request("JOB-001", True),
                        Request("JOB-001", 0), Request("JOB-001", 10)]:
            with self.subTest(request=request):
                with self.assertRaises(InputRejected):
                    precheck(self.reader, request)
        self.assertEqual(rows(self.editor), [])

    def test_manual_oracle(self):
        stream = io.StringIO()
        with contextlib.redirect_stdout(stream):
            main()
        expected = json.loads(Path(__file__).with_name("expected-policy.json").read_text())
        self.assertEqual(json.loads(stream.getvalue()), expected)


if __name__ == "__main__":
    unittest.main()

如果观察到 False，先重新读取当前规则和业务请求；不符合当前条件就保持无效果，不循环重放旧 Snapshot。若得到数据库异常，先按异常类型检查连接、事务与现有行，不能切换到 insert_naive 绕过条件。恢复必须有仍然有效的业务输入；本例的缩减份数由读者明确选择。

保证止于同一数据库中的条件判断和草稿提交，并依赖所有相关规则写入都推进修订号。之后规则再改，不会追溯撤销已提交行。多张规则表、异步复制、外部服务和分布式一致性都未覆盖；现有输入、输出护栏仍各自承担原来的检查职责。本文程序没有模型、网络、身份认证、审批或真实操作。

下载完整文章与可运行源码（ZIP）。解压后阅读 README.txt；包内含原文代码、新实验、测试和手工预期，不需要账号或密钥。

本地账本把效果和幂等记录放在同一个 SQLite 行里，避免在这份小实验中出现“草稿写了、去重键没写”的双写缺口。保证只适用于保留这份数据库、沿用同一 job、串行调用本程序的范围。删除数据库、换业务键、换到另一套存储，都不在这项保证之内。它也没有证明并发请求下的生产吞吐与故障恢复。

前检只能依据当时拿到的事实作判断。如果检查依赖会变的库存、余额或权限，检查与提交之间还有时间窗口；真正的业务提交应在服务端重新校验关键条件，必要时用事务、条件更新或版本号。输出护栏适合检查结果能否继续使用。已经写入的内容若需要撤销，要用业务提供的取消或补偿流程，并另外验证执行结果。补偿本身也可能失败。

审计回执提供可回看的证据，前提是它确实保存成功。这个例子允许“有草稿、没回执”的明确失败窗口，并用测试暴露它。进程被强制终止、磁盘损坏、异步远端效果和跨服务一致性都未测试。正式系统遇到未知结果时，应记录待核对的业务键并对账，不能把缺回执解释为没有发生。

迁移到真实工具时，先给每个步骤标出提交点：哪一步之前还能保证没有产生效果，哪一步之后只能查询或补偿。再分别验收拒绝、检查异常、调用异常、取消与重试。这个顺序比看到一个 tripwire 名字就放心更容易发现遗漏。本文到这里结束，代码保持离线，避免教学演示替你触发真实操作。