给 Agent 工具加一份可撤销的单次审批
用一个完全离线的发布模拟器，确认批准过的内容、实际执行的参数和数据库回执始终能对应起来。
你能带走什么
- 审批绑定完整参数和业务身份
- 到期与撤销在执行入口再次检查
- 单次消费和本地效果同事务提交
- 结果未知时先查询历史回执
- 本地摘要与角色夹具均不提供认证
1. 让一次批准只对应眼前这次调用
一个写作 Agent 准备发布周报：目的地是 preview-a，标题是“合成周报”，正文只有一句话。编辑看完后点了批准。几分钟后，Agent 把目的地换成另一个频道，或者重启后把同一调用又跑了一次。最初那次点击究竟允许了什么，需要在程序里给出可以核对的答案。
本文把一份批准收紧为一个可持久化、可撤销、带期限的单次许可。它只能消费冻结的工具调用；任何目的地、正文、工具版本或运行环境变化都要重新提出候选。实验中的 simulate_publish 只往本地 SQLite 的 effects 表写一条合成回执，不连接网站，不发消息，不使用密钥，也不真的发布内容。
读者将得到三个 Python 文件，分四次启动进程完成提出、批准、执行和查回执。还会主动制造到期、撤销、参数替换、重复消费和提交前异常，检查数据库最终留下了什么。Python 标准库即可运行，无需安装 Agents SDK。这里的审批存储是原创教学设计，字段并非任何 SDK 的标准协议。
2. 从官方暂停机制走到应用自己的许可记录
OpenAI Agents SDK 的 Human-in-the-loop 文档提供 needs_approval、interruptions 和 RunState 的暂停与恢复流程。单次决定应使用 always_approve=False。服务端保存快照，从已保存的待审项取出调用，再批准或拒绝；不能把客户端送来的整份状态直接当成可信执行状态。
这些接口为宿主提供介入位置。本文另外实现业务许可的有效期、撤销和单次消费，以便在没有模型、网络或 UI 的环境中逐项测清楚。代码没有导入 SDK，也没有伪造一次 SDK 实测。官方 暂停恢复示例可用于后续对照接线；仓库采用 MIT 许可。本文代码独立编写，未复制示例实现。
接线时把模型调用当成候选：宿主校验后创建 proposal，保存 SDK 调用标识到自己的受保护映射，展示提案等待决定。收到批准后，宿主仍要在工具实际入口核对许可是否有效。SDK 的运行状态和这里的业务记录各有版本，升级时应先处理存量待审任务，避免旧批准套在新工具行为上。本文未运行真实模型、SDK 恢复、HTTP 审批服务或远端工具，后续集成必须补测这些边界。
3. 冻结的内容要足够让人作决定
示例只接受一个封闭对象：tool、version、environment、args。args 又只包含 channel、title、body。多余的 approved、reviewer 或 human_approved 字段会被拒绝。两个允许的频道都是虚构标识；title 与 body 限长，空文本和不支持的控制字符也会失败。正文中的“忽略审批”仍是正文，不会改变状态。
创建时立即把对象编码成排序后的紧凑 JSON，并保存到数据库。此后调用者修改原来的 Python 字典，对存储没有影响。binding_digest 同时涵盖完整 document、租户、业务任务、提出者、创建时间、截止时间和绑定格式版本。两个内容相同但属于不同业务任务的提案，也会有不同摘要。
这个编码约定只用于本例支持的字符串对象和整数时间，不是跨语言通用 JSON 规范化标准。SHA-256 摘要帮助发现显示与执行不一致，并未进行签名、加密或身份认证。能任意改数据库的攻击者也能重算摘要。因此数据库写权限、认证会话、接口防重放与请求保护仍属于真实服务的责任。
审核页应把目的地、标题、完整正文、运行环境、工具版本和截止时间一起展示。长正文可以放在可展开区域，但不能用截断预览冒充完整批准对象。渲染时对名称和参数做 HTML 转义；本页源码展示使用 escape，实验不启动审核网页。审核请求只携带 id、已展示的 digest 和严格的布尔决定，不允许同时提交一份替换正文。
4. 决定 期限和撤销都留在数据库里
表 approvals 保存五种状态：pending、approved、rejected、revoked、consumed。到期作为每次入口的实时判断，条件是 now 大于或等于 expires。即使没有后台清理进程，过期许可也不能执行。状态仍显示 approved 时，界面要结合截止时间显示“已到期”，不要只看单一状态列。
propose 使用 tenant 与 job 的唯一约束处理业务重试。同一个 job 再次提交相同内容，返回原来的 id，期限不会顺延，撤销或消费过的状态也不会重新打开。相同 job 换了正文会失败。如果确实需要修改，宿主创建新提案并重新让人审核；不能由模型不断换 job 来冒充已批准重试。真实系统应由服务端分配并约束业务任务身份。
decide 只接受 pending，重复点击直接拒绝，调用方重新读取显示状态。revoke 可以取消 pending 或 approved，但不能撤回已经 consumed 的效果。撤销与执行发生竞争时，谁先取得写事务并提交，就决定另一方看到的状态。编辑撤销成功后不能再执行；执行先提交以后，撤销会收到明确拒绝，需要走业务自身的补救流程。
示例时间 100、110、120、160 都是合成逻辑秒，方便稳定重现期限边界，不能解释成当前真实时间。代码拒绝早于创建或决定时间的执行。生产需要可信服务端时钟及一致的期限策略，不接收浏览器传入的 now；也不能依赖这个简化检查检测系统时钟的所有回拨。
5. 只在一个短事务里消费本地许可
execute 先验证可信调用方角色，再进入 BEGIN IMMEDIATE，回读当前记录、检查时间、摘要和状态，然后重新按当前工具规则验证已冻结的 document。实际执行只读取数据库中的参数，接口不接受另一份 payload。检查全部通过，才写入 effects，并把许可改成 consumed。两个动作同处一个事务。
SQLite 对同时写入有明确约束，事务文档说明 BEGIN IMMEDIATE 会立即尝试开启写事务，另有写事务时可能遇到忙错误。本例设置五秒连接超时；测试使用两个连接竞争同一许可，只有一个消费成功。生产应把忙错误作为可解释的短暂冲突，保留原业务标识，不能因此放弃校验直接调用工具。
fail_after_effect 在写入效果后、提交前抛出异常。事务回滚后 effects 仍为空，许可仍是 approved，之后可以正常执行一次。若提交已经成功，只是进程没收到返回值，receipt 可以回查已保存的结果；再次 execute 会被拒绝。回执记录执行当时的事实，读取它不会刷新期限，也不会再次消费。
这个原子保证严格限定在同一个 SQLite 数据库内。真实远端发布成功后，本地进程可能在写回状态前退出；数据库事务不能回滚远端动作。把 HTTP 调用直接塞进本例事务，并不能获得跨系统 exactly-once。接入远端前需要目标系统支持的幂等键、结果查询，以及“结果未知”时的对账入口；在确认之前应停止重复副作用。本例故意没有接入那段流程。
6. 先检查信任边界 再运行合成案例
Actor 是测试夹具。AUTHOR、REVIEWER、EXECUTOR 在一个 Python 模块内定义，用来检查函数是否把提出、审核和执行混在一起。任何能修改或导入同一进程代码的人都可以构造这些值，它不提供登录验证、JWT 验签、会话隔离或抵抗恶意进程的能力。测试角色拒绝，只能证明这段本地分支按预期工作。
正式服务需要认证中间件提供主体，并对每个审批对象检查租户、角色和资源权限。id 与 digest 都不是 bearer token，知道它们不能获得批准权。数据库、序列化状态和回执可能含私人正文，应限制读取并制定保留规则；不要把完整正文、密钥或快照塞进诊断日志。
在一个新目录放入下面三个文件，然后运行这些命令。demo.py 拒绝覆盖已有演示数据库，因此重复从头体验时换一个空目录即可。每条命令都是新的 Python 进程，第二步的批准仅为预设合成决定，没有模拟真实用户登录。
python3 -m unittest -v
python3 demo.py init
python3 demo.py approve
python3 demo.py execute
python3 demo.py receiptinit 输出 pending、expires=160、完整 document 与本次随机 id。approve 输出 approved (synthetic reviewer, t=110)。execute 输出 kind=local-simulation-only、executed_at=120 与冻结的 payload。receipt 返回同一份回执。随机 UUID 每次不同，不应逐字比较示例 id。
接着再次执行 python3 demo.py execute，应以 Denied: approval unavailable for consumption 退出。若在 approve 之前执行，也会得到相同拒绝且没有效果。测试文件另外覆盖恰好 t=160 的到期分支、t=159 的最后有效秒，以及修改正文但沿用 job 的冲突。
7. 完整许可存储实现
阅读时可沿 propose → view → decide → execute → receipt 走一遍。view 返回完整审核对象的副本；决定与消费入口都回读当前数据库。transaction 只包围短暂存储步骤，不覆盖等待用户的时间。
approval.py · 完整源码
"""Local teaching store. No network, model, credentials, or real publishing.
Actor and now are trusted fixture inputs, NOT authentication or a secure clock.
Do not expose these methods as unauthenticated HTTP endpoints.
"""
import hashlib
import json
import sqlite3
from contextlib import contextmanager
from dataclasses import dataclass
from uuid import uuid4
class Denied(Exception):
pass
@dataclass(frozen=True)
class Actor:
tenant: str
subject: str
role: str
AUTHOR = Actor('demo', 'writer', 'proposer')
REVIEWER = Actor('demo', 'editor', 'reviewer')
EXECUTOR = Actor('demo', 'local-worker', 'executor')
def integer(value):
if type(value) is not int or value < 0:
raise ValueError('nonnegative integer required')
return value
def normalize(payload):
if type(payload) is not dict or set(payload) != {'tool', 'version', 'environment', 'args'}:
raise ValueError('exact proposal schema required')
if (payload['tool'], payload['version'], payload['environment']) != (
'simulate_publish', 'publisher-v1', 'offline'):
raise ValueError('unsupported tool binding')
args = payload['args']
if type(args) is not dict or set(args) != {'channel', 'title', 'body'}:
raise ValueError('exact argument schema required')
if args['channel'] not in ('preview-a', 'preview-b'):
raise ValueError('unknown synthetic channel')
for key, limit in [('title', 80), ('body', 4000)]:
value = args[key]
if type(value) is not str or not value.strip() or len(value) > limit:
raise ValueError('invalid ' + key)
if any(ord(c) < 32 and c not in '\n\t' for c in value):
raise ValueError('control character')
return json.dumps(payload, ensure_ascii=False, sort_keys=True,
separators=(',', ':'), allow_nan=False)
def fingerprint(document):
return hashlib.sha256(document.encode('utf-8')).hexdigest()
def binding_digest(document, tenant, job, requester, created, expires):
envelope = dict(document=document, tenant=tenant, job=job, requester=requester,
created=created, expires=expires, binding_version=1)
return fingerprint(json.dumps(envelope, ensure_ascii=False, sort_keys=True,
separators=(',', ':')))
class Store:
def __init__(self, path):
self.db = sqlite3.connect(path, isolation_level=None, timeout=5)
self.db.row_factory = sqlite3.Row
self.db.execute('PRAGMA foreign_keys=ON')
self.db.executescript('''
CREATE TABLE IF NOT EXISTS approvals(
id TEXT PRIMARY KEY, tenant TEXT NOT NULL, job TEXT NOT NULL,
requester TEXT NOT NULL, document TEXT NOT NULL, digest TEXT NOT NULL,
created INTEGER NOT NULL, expires INTEGER NOT NULL,
status TEXT NOT NULL CHECK(status IN
('pending','approved','rejected','revoked','consumed')),
reviewer TEXT, decided INTEGER,
UNIQUE(tenant,job));
CREATE TABLE IF NOT EXISTS effects(
approval_id TEXT PRIMARY KEY REFERENCES approvals(id),
receipt TEXT NOT NULL);
''')
def close(self):
self.db.close()
@contextmanager
def transaction(self):
self.db.execute('BEGIN IMMEDIATE')
try:
yield
self.db.commit()
except BaseException:
self.db.rollback()
raise
@staticmethod
def role(actor, expected):
if actor not in (AUTHOR, REVIEWER, EXECUTOR) or actor.role != expected:
raise Denied('fixture role denied')
def row(self, ident, actor):
row = self.db.execute('SELECT * FROM approvals WHERE id=?', (ident,)).fetchone()
if row is None or row['tenant'] != actor.tenant:
raise Denied('unavailable approval')
return row
@staticmethod
def active(row, now):
integer(now)
if now < row['created'] or (row['decided'] is not None and now < row['decided']):
raise Denied('clock moved backwards')
if now >= row['expires']:
raise Denied('expired')
def propose(self, actor, job, payload, now, ttl=60):
self.role(actor, 'proposer')
integer(now)
if type(ttl) is not int or not 1 <= ttl <= 300:
raise ValueError('TTL must be 1..300')
if type(job) is not str or not 1 <= len(job) <= 100:
raise ValueError('job required')
document = normalize(payload)
digest = binding_digest(document, actor.tenant, job, actor.subject, now, now + ttl)
with self.transaction():
old = self.db.execute('SELECT * FROM approvals WHERE tenant=? AND job=?',
(actor.tenant, job)).fetchone()
if old is not None:
if old['document'] != document or old['requester'] != actor.subject:
raise Denied('job already bound to another proposal')
return old['id'] # Never refresh expiry or reopen terminal states.
ident = str(uuid4())
self.db.execute('INSERT INTO approvals VALUES(?,?,?,?,?,?,?,?,?,?,?)',
(ident, actor.tenant, job, actor.subject, document,
digest, now, now + ttl, 'pending', None, None))
return ident
def view(self, actor, ident):
self.role(actor, 'reviewer')
row = self.row(ident, actor)
return {k: row[k] for k in ('id','job','requester','document','digest',
'created','expires','status','reviewer','decided')}
def decide(self, actor, ident, shown_digest, approve, now):
self.role(actor, 'reviewer')
if type(approve) is not bool:
raise ValueError('boolean decision required')
with self.transaction():
row = self.row(ident, actor)
self.active(row, now)
if shown_digest != row['digest']:
raise Denied('review display mismatch')
if row['status'] != 'pending':
raise Denied('decision already resolved')
self.db.execute('UPDATE approvals SET status=?,reviewer=?,decided=? WHERE id=?',
('approved' if approve else 'rejected', actor.subject, now, ident))
def revoke(self, actor, ident, now):
self.role(actor, 'reviewer')
integer(now)
with self.transaction():
row = self.row(ident, actor)
if now < row['created'] or (row['decided'] is not None and now < row['decided']):
raise Denied('clock moved backwards')
if row['status'] not in ('pending', 'approved'):
raise Denied('cannot revoke terminal approval')
self.db.execute('UPDATE approvals SET status=?,reviewer=?,decided=? WHERE id=?',
('revoked', actor.subject, now, ident))
def execute(self, actor, ident, expected_digest, now, fail_after_effect=False):
self.role(actor, 'executor')
with self.transaction():
row = self.row(ident, actor)
self.active(row, now)
if expected_digest != row['digest']:
raise Denied('execution binding mismatch')
if row['status'] != 'approved':
raise Denied('approval unavailable for consumption')
# Revalidate frozen server-owned data against CURRENT supported code.
document = normalize(json.loads(row['document']))
digest = binding_digest(document, row['tenant'], row['job'], row['requester'],
row['created'], row['expires'])
if document != row['document'] or digest != row['digest']:
raise Denied('stored binding mismatch')
receipt = json.dumps({'kind': 'local-simulation-only', 'approval_id': ident,
'digest': row['digest'], 'executed_at': now,
'payload': json.loads(document)}, ensure_ascii=False, sort_keys=True)
self.db.execute('INSERT INTO effects VALUES(?,?)', (ident, receipt))
if fail_after_effect:
raise RuntimeError('injected pre-commit failure')
self.db.execute('UPDATE approvals SET status=? WHERE id=?', ('consumed', ident))
return json.loads(receipt)
def receipt(self, actor, ident):
self.role(actor, 'executor')
self.row(ident, actor)
row = self.db.execute('SELECT receipt FROM effects WHERE approval_id=?', (ident,)).fetchone()
return None if row is None else json.loads(row['receipt'])
8. 四次进程启动的演示入口
approval-demo-id.json 只存定位信息，没有授权能力；修改它不能替换数据库里的正文。demo.py 中的身份和时间用于复现，不能照搬成对外接口的参数。
demo.py · 完整源码
"""Run only in a fresh empty directory: python demo.py init|approve|execute|receipt."""
import json
import sys
from pathlib import Path
from approval import Store, AUTHOR, REVIEWER, EXECUTOR
DB = Path('approval-demo.sqlite')
INFO = Path('approval-demo-id.json')
if len(sys.argv) != 2 or sys.argv[1] not in {'init','approve','execute','receipt'}:
raise SystemExit('usage: python demo.py init|approve|execute|receipt')
command = sys.argv[1]
if command == 'init' and (DB.exists() or INFO.exists()):
raise SystemExit('Use a fresh directory; refusing to replace an existing demo.')
if command != 'init' and (not DB.exists() or not INFO.exists()):
raise SystemExit('Run init first in this directory.')
s = Store(DB)
try:
if command == 'init':
p = {'tool':'simulate_publish','version':'publisher-v1','environment':'offline',
'args':{'channel':'preview-a','title':'合成周报','body':'本周完成三个离线测试。'}}
ident = s.propose(AUTHOR, 'weekly-001', p, 100, 60)
view = s.view(REVIEWER, ident)
INFO.write_text(json.dumps({'id':ident,'digest':view['digest']}), encoding='utf-8')
print(json.dumps(view, ensure_ascii=False, indent=2))
else:
info = json.loads(INFO.read_text(encoding='utf-8'))
if command == 'approve':
# A deterministic synthetic reviewer decision, NOT an authenticated human UI.
s.decide(REVIEWER, info['id'], info['digest'], True, 110)
print('approved (synthetic reviewer, t=110)')
elif command == 'execute':
print(json.dumps(s.execute(EXECUTOR, info['id'], info['digest'], 120), ensure_ascii=False, indent=2))
else:
print(json.dumps(s.receipt(EXECUTOR, info['id']), ensure_ascii=False, indent=2))
finally:
s.close()
9. 把失败路径写成可以重新运行的断言
本次在 Python 3.12.14、SQLite 3.53.1 实际运行 33 项 unittest，全部通过。测试使用临时文件数据库，覆盖独立子进程恢复与双连接竞争。它们验证本例的应用约束，未验证实际 SDK、模型选择、登录服务、操作系统崩溃断电或远端发布。
参数替换、审批字样注入、错误角色与版本变化都应留下零条效果；重复消费与双连接竞争最多留一条。提交前故障后许可仍可用；提交后的历史回执可以单独读取。这样检查结果比只观察抛出了某个异常更有意义。
test_approval.py · 完整源码
import copy
import json
import sqlite3
import subprocess
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from approval import Store, AUTHOR, REVIEWER, EXECUTOR, Actor, Denied
def payload():
return dict(tool='simulate_publish', version='publisher-v1', environment='offline',
args=dict(channel='preview-a', title='合成周报', body='本周完成三个离线测试。'))
class ApprovalTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.path = str(Path(self.tmp.name) / 'test.sqlite')
self.s = Store(self.path)
self.ident = self.s.propose(AUTHOR, 'job-1', payload(), 100, 60)
self.digest = self.s.view(REVIEWER, self.ident)['digest']
def tearDown(self):
self.s.close()
self.tmp.cleanup()
def approve(self):
self.s.decide(REVIEWER, self.ident, self.digest, True, 110)
def run_effect(self, now=120, **kw):
return self.s.execute(EXECUTOR, self.ident, self.digest, now, **kw)
def test_pending_denied(self):
with self.assertRaises(Denied): self.run_effect()
self.assertIsNone(self.s.receipt(EXECUTOR, self.ident))
def test_happy_path(self):
self.approve()
r = self.run_effect()
self.assertEqual(r['payload'], payload())
self.assertEqual(r['kind'], 'local-simulation-only')
self.assertEqual(self.s.view(REVIEWER, self.ident)['status'], 'consumed')
def test_single_consumption(self):
self.approve(); self.run_effect()
with self.assertRaises(Denied): self.run_effect()
self.assertEqual(self.s.db.execute('SELECT count(*) FROM effects').fetchone()[0], 1)
def test_receipt_is_queryable_after_expiry(self):
self.approve(); r = self.run_effect()
with self.assertRaises(Denied): self.run_effect(now=200)
self.assertEqual(self.s.receipt(EXECUTOR, self.ident), r)
def test_rejected(self):
self.s.decide(REVIEWER, self.ident, self.digest, False, 110)
with self.assertRaises(Denied): self.run_effect()
def test_pending_revocation(self):
self.s.revoke(REVIEWER, self.ident, 105)
with self.assertRaises(Denied): self.approve()
def test_approved_revocation(self):
self.approve(); self.s.revoke(REVIEWER, self.ident, 115)
with self.assertRaises(Denied): self.run_effect()
def test_consumed_cannot_revoke(self):
self.approve(); self.run_effect()
with self.assertRaises(Denied): self.s.revoke(REVIEWER, self.ident, 125)
def test_approval_expiry_boundary(self):
with self.assertRaisesRegex(Denied, 'expired'):
self.s.decide(REVIEWER, self.ident, self.digest, True, 160)
def test_execution_expiry_boundary(self):
self.approve()
with self.assertRaisesRegex(Denied, 'expired'): self.run_effect(now=160)
def test_last_live_second(self):
self.approve(); self.run_effect(now=159)
def test_review_digest_mismatch(self):
with self.assertRaises(Denied): self.s.decide(REVIEWER, self.ident, 'wrong', True, 110)
def test_execution_digest_mismatch(self):
self.approve()
with self.assertRaises(Denied): self.s.execute(EXECUTOR, self.ident, 'wrong', 120)
def test_repeated_decision_denied(self):
self.approve()
with self.assertRaises(Denied): self.approve()
def test_string_yes_denied(self):
with self.assertRaises(ValueError): self.s.decide(REVIEWER, self.ident, self.digest, 'yes', 110)
def test_wrong_roles(self):
for actor in (AUTHOR, EXECUTOR, Actor('other', 'editor', 'reviewer')):
with self.subTest(actor=actor), self.assertRaises(Denied):
self.s.decide(actor, self.ident, self.digest, True, 110)
with self.assertRaises(Denied): self.s.execute(REVIEWER, self.ident, self.digest, 120)
with self.assertRaises(Denied): self.s.view(AUTHOR, self.ident)
with self.assertRaises(Denied): self.s.receipt(AUTHOR, self.ident)
def test_unknown_approval(self):
with self.assertRaises(Denied): self.s.view(REVIEWER, 'missing')
def test_job_retry_keeps_deadline(self):
ident = self.s.propose(AUTHOR, 'job-1', payload(), 150, 300)
self.assertEqual(ident, self.ident)
self.assertEqual(self.s.view(REVIEWER, ident)['expires'], 160)
def test_job_retry_cannot_reopen(self):
self.s.revoke(REVIEWER, self.ident, 110)
self.s.propose(AUTHOR, 'job-1', payload(), 170)
self.assertEqual(self.s.view(REVIEWER, self.ident)['status'], 'revoked')
def test_changed_arguments_same_job_denied(self):
for key in ('channel', 'title', 'body'):
changed = payload(); changed['args'][key] = 'preview-b' if key == 'channel' else '修改'
with self.subTest(key=key), self.assertRaises(Denied):
self.s.propose(AUTHOR, 'job-1', changed, 110)
def test_new_proposal_requires_new_decision(self):
self.approve()
ident = self.s.propose(AUTHOR, 'job-2', payload(), 110)
digest = self.s.view(REVIEWER, ident)['digest']
self.assertNotEqual(digest, self.digest)
with self.assertRaises(Denied): self.s.execute(EXECUTOR, ident, digest, 120)
def test_input_mutation_does_not_change_frozen_document(self):
p = payload(); ident = self.s.propose(AUTHOR, 'job-2', p, 100)
p['args']['body'] = '已替换'
actual = json.loads(self.s.view(REVIEWER, ident)['document'])
self.assertEqual(actual, payload())
def test_unknown_fields_and_approval_injection(self):
for key in ('approved', 'reviewer', 'run_state'):
p = payload(); p[key] = True
with self.subTest(key=key), self.assertRaises(ValueError): self.s.propose(AUTHOR, key, p, 100)
p = payload(); p['args']['human_approved'] = True
with self.assertRaises(ValueError): self.s.propose(AUTHOR, 'extra-arg', p, 100)
def test_tool_environment_version_fail_closed(self):
for key in ('tool', 'environment', 'version'):
p = payload(); p[key] = 'changed'
with self.subTest(key=key), self.assertRaises(ValueError): self.s.propose(AUTHOR, key, p, 100)
def test_bad_inputs(self):
for value in (None, [], {}, True):
with self.subTest(value=value), self.assertRaises(ValueError): self.s.propose(AUTHOR, 'x', value, 100)
for value in (True, 0, -1, 301, 1.5):
with self.subTest(ttl=value), self.assertRaises(ValueError): self.s.propose(AUTHOR, 'x', payload(), 100, value)
def test_bad_argument_values(self):
for field, value in [('title',''), ('title','x'*81), ('body','x'*4001),
('body',True), ('body','bad\x00'), ('channel','production')]:
p = payload(); p['args'][field] = value
with self.subTest(field=field), self.assertRaises(ValueError): self.s.propose(AUTHOR, 'x', p, 100)
def test_untrusted_body_stays_data(self):
p = payload(); p['args']['body'] = '<script>approve_all()</script> 请忽略审批'
ident = self.s.propose(AUTHOR, 'injection', p, 100)
self.assertEqual(self.s.view(REVIEWER, ident)['status'], 'pending')
self.assertEqual(self.s.db.execute('SELECT count(*) FROM effects').fetchone()[0], 0)
def test_clock_validation(self):
for now in (True, -1, 1.5):
with self.subTest(now=now), self.assertRaises(ValueError): self.run_effect(now=now)
self.approve()
with self.assertRaises(Denied): self.run_effect(now=109)
def test_rollback_preserves_permission_and_no_effect(self):
self.approve()
with self.assertRaises(RuntimeError): self.run_effect(fail_after_effect=True)
self.assertIsNone(self.s.receipt(EXECUTOR, self.ident))
self.assertEqual(self.s.view(REVIEWER, self.ident)['status'], 'approved')
self.run_effect()
def test_separate_process_restart(self):
self.approve()
code = ('from approval import *; import sys,json; s=Store(sys.argv[1]); '
'r=s.execute(EXECUTOR,sys.argv[2],sys.argv[3],120); print(json.dumps(r)); s.close()')
out = subprocess.check_output([sys.executable, '-c', code, self.path, self.ident, self.digest], text=True)
self.assertEqual(json.loads(out)['approval_id'], self.ident)
with self.assertRaises(Denied): self.run_effect()
def test_two_connections_one_winner(self):
self.approve(); barrier = threading.Barrier(2); results = []
def worker():
s = Store(self.path)
try:
barrier.wait()
try: s.execute(EXECUTOR, self.ident, self.digest, 120); results.append('ok')
except Denied: results.append('denied')
finally: s.close()
threads = [threading.Thread(target=worker) for _ in range(2)]
for t in threads: t.start()
for t in threads: t.join()
self.assertEqual(sorted(results), ['denied', 'ok'])
self.assertEqual(self.s.db.execute('SELECT count(*) FROM effects').fetchone()[0], 1)
def test_unsupported_stored_version_rejected(self):
self.approve(); p = payload(); p['version'] = 'publisher-v2'
self.s.db.execute('UPDATE approvals SET document=? WHERE id=?', (json.dumps(p), self.ident))
with self.assertRaises(ValueError): self.run_effect()
self.assertIsNone(self.s.receipt(EXECUTOR, self.ident))
def test_accidental_storage_change_detected(self):
self.approve()
self.s.db.execute('UPDATE approvals SET expires=999 WHERE id=?', (self.ident,))
with self.assertRaisesRegex(Denied, 'stored binding mismatch'): self.run_effect()
if __name__ == '__main__':
unittest.main()
10. 补充：获得写锁以后，时间参数可能已经过时
把前文的逻辑时间接到真实服务时，还有一个很容易漏掉的顺序问题。假设宿主先读取当前时间，再调用 execute(..., now)。传入的是一个已经算好的整数。即使执行入口内部有到期判断，进入 BEGIN IMMEDIATE 时发生的锁等待，也不会让这个整数继续前进。
前文用固定的 100、110、120、160 来复现状态机，原有 33 项测试仍然成立。这里补的是另一条接线约定：如果业务要求按“获得写事务、准备消费许可”的时点判断期限，就应在成功获得写事务之后读取宿主时钟。下面保留原实现，另外增加 LockedClockStore，对照同一份 expires=160 的许可。
这个位置确实有可能等待。SQLite 事务文档说明同一时刻只允许一个写事务，BEGIN IMMEDIATE 会尝试立即开启写事务；另一个写者占用时可能返回忙错误。Python 的连接文档则解释了 timeout 所控制的等待。原 Store 设置了五秒连接超时，这个值既不会延长批准期限，也不会替我们重新取时间。
实验里，连接 A 在逻辑时间 159 持有写事务，连接 B 随后尝试 BEGIN IMMEDIATE。宿主把注入的逻辑时钟推进到 160，再释放 A。旧入口在尝试之前取到 159，最后仍消费成功，留下 consumed 和一条 effects；新入口获得写事务后才读到 160，抛出 expired，保留 approved 记录，效果为零。决定入口同样如此：旧入口写入 decided=159，新入口保留 pending。
这不是已确认的生产漏洞，也没有证明线程真实等待了一秒。两条数据库连接是真实的，159→160 是测试主动推进的逻辑时钟。我们验证的是锁竞争边界两侧的取样顺序，而不是操作系统调度、SQLite 忙处理器耗时或某台机器的物理时钟精度。
11. 另存一个在锁后取样的审批入口
将下面代码保存为 locked_clock.py，与原来的 approval.py 放在同一目录。宿主在构造对象时安装时钟回调；新的 decide 与 execute 都不接收每次请求携带的 now。execute 把故障注入参数设为仅限关键字，旧代码误传一个位置参数时会收到 TypeError，不会把旧 now 悄悄当成另一个开关。
两个入口都先校验角色，再进入原有事务上下文。只有 BEGIN IMMEDIATE 成功返回，才调用一次时钟，然后回读记录，执行原有时间、摘要、状态及冻结内容校验。effects 插入和 consumed 更新仍在同一个事务中，没有加入网络调用。时钟回调抛异常，或者返回布尔值、负数、小数、NaN、空值等无效时间，都会退出并回滚，不留下效果。
locked_clock.py
"""Separate teaching adapter; clock is installed by the trusted local host.
Only decision and consumption sample clock after BEGIN IMMEDIATE succeeds.
This is an admission-time check, not a deadline for commit or external effects.
The parent Store's fixture actors are still not authentication.
"""
import json
from approval import Store, Denied, normalize, binding_digest
class LockedClockStore(Store):
def __init__(self, path, clock):
if not callable(clock):
raise TypeError('host clock must be callable')
self._clock = clock
super().__init__(path)
def decide(self, actor, ident, shown_digest, approve):
self.role(actor, 'reviewer')
if type(approve) is not bool:
raise ValueError('boolean decision required')
with self.transaction():
# BEGIN IMMEDIATE has succeeded; any lock wait is already over.
now = self._clock()
row = self.row(ident, actor)
self.active(row, now)
if shown_digest != row['digest']:
raise Denied('review display mismatch')
if row['status'] != 'pending':
raise Denied('decision already resolved')
self.db.execute('UPDATE approvals SET status=?,reviewer=?,decided=? WHERE id=?',
('approved' if approve else 'rejected', actor.subject, now, ident))
def execute(self, actor, ident, expected_digest, *, fail_after_effect=False):
self.role(actor, 'executor')
with self.transaction():
now = self._clock()
row = self.row(ident, actor)
self.active(row, now)
if expected_digest != row['digest']:
raise Denied('execution binding mismatch')
if row['status'] != 'approved':
raise Denied('approval unavailable for consumption')
document = normalize(json.loads(row['document']))
digest = binding_digest(document, row['tenant'], row['job'], row['requester'],
row['created'], row['expires'])
if document != row['document'] or digest != row['digest']:
raise Denied('stored binding mismatch')
receipt = json.dumps({'kind': 'local-simulation-only', 'approval_id': ident,
'digest': row['digest'], 'executed_at': now,
'payload': json.loads(document)}, ensure_ascii=False, sort_keys=True)
self.db.execute('INSERT INTO effects VALUES(?,?)', (ident, receipt))
if fail_after_effect:
raise RuntimeError('injected pre-commit failure')
self.db.execute('UPDATE approvals SET status=? WHERE id=?', ('consumed', ident))
return json.loads(receipt)
这里故意只覆盖决定与消费入口。propose 和 revoke 仍继承前文的教学夹具接口；Actor 常量也仍然不是认证。把 callable 注入构造函数不会自动使它可信，不能让客户端提交回调、替换宿主时钟，或者绕回旧 Store.execute 取得另一条执行路径。部署中的主体、业务任务身份、时钟来源和入口路由都必须由受保护的宿主管理。
时间尺度也需要事先约定。若 created、decided、expires 使用一种持久化时间尺度，回调必须返回同尺度、同单位的非负整数。Python 的 monotonic 文档说明其参考起点没有定义，适合计算两次读数的差；不能把原始 monotonic 值直接拿来比较另一种尺度的到期时间。本适配器保留原有“早于创建或决定就拒绝”的检查，没有增加持久化的历史最高时间，也不能识别所有系统时钟回拨。
12. 用两个真实连接重现期限边界
下面的 lock_case.py 给出完整输入和调度过程。PAYLOAD 是仅写入本地的合成正文；created=100、ttl=60，执行案例先在 110 批准。每个案例使用新的临时文件数据库，工作线程在自己的线程内打开连接，避免把一个连接跨线程传递。
四个 Event 分别协调连接准备好、开始调用、BEGIN 尝试和调用结束。SQL trace 回调在连接 B 尝试 BEGIN IMMEDIATE 时发出信号；此时 A 仍持有真实写事务。主线程收到信号后才推进逻辑时钟并释放锁。信号证明尝试发生在另一写者持锁期间，不证明 B 已在忙处理器里耗费了某个时长。没有用 sleep 猜测先后；等待超时只用于让损坏的测试及时报错。
lock_case.py
"""Two real SQLite connections; Event gates, not guessed sleep durations."""
import json
from pathlib import Path
import sqlite3
import tempfile
import threading
from approval import Store, AUTHOR, REVIEWER, EXECUTOR
from locked_clock import LockedClockStore
PAYLOAD = {'tool': 'simulate_publish', 'version': 'publisher-v1',
'environment': 'offline', 'args': {'channel': 'preview-a',
'title': '锁等待样例', 'body': '这是一条合成数据。'}}
class Clock:
def __init__(self, now=159):
self.now = now
self.samples = []
def __call__(self):
self.samples.append(self.now)
return self.now
def contention_case(fixed, operation='execute', release_time=160, busy=False):
with tempfile.TemporaryDirectory() as root:
path = Path(root) / 'approval.sqlite'
setup = Store(path)
ident = setup.propose(AUTHOR, 'lock-case', PAYLOAD, 100, 60)
digest = setup.view(REVIEWER, ident)['digest']
if operation == 'execute':
setup.decide(REVIEWER, ident, digest, True, 110)
setup.close()
ready, start, attempted, finished = [threading.Event() for _ in range(4)]
clock, result, errors = Clock(), {}, []
timeline = []
def worker():
store = None
try:
store = LockedClockStore(path, clock) if fixed else Store(path)
if busy:
store.db.execute('PRAGMA busy_timeout=0')
ready.set()
if not start.wait(5):
raise RuntimeError('start barrier timed out')
def trace(sql):
if sql == 'BEGIN IMMEDIATE':
timeline.append('attempt_begin_while_other_writer_holds_lock')
attempted.set()
store.db.set_trace_callback(trace)
kwargs = {} if fixed else {'now': clock()}
try:
if operation == 'execute':
receipt = store.execute(EXECUTOR, ident, digest, **kwargs)
result['sampled_at'] = receipt['executed_at']
else:
store.decide(REVIEWER, ident, digest, True, **kwargs)
result['sampled_at'] = store.view(REVIEWER, ident)['decided']
result['outcome'] = 'accepted'
except Exception as exc:
result['outcome'] = type(exc).__name__ + ': ' + str(exc)
except BaseException as exc:
errors.append(repr(exc))
finally:
if store is not None:
store.close()
finished.set()
thread = threading.Thread(target=worker)
thread.start()
blocker = None
try:
if not ready.wait(5):
raise RuntimeError('worker setup barrier timed out: ' + repr(errors))
blocker = sqlite3.connect(path, isolation_level=None)
blocker.execute('BEGIN IMMEDIATE')
timeline.append('other_writer_lock_acquired_at_159')
start.set()
if not attempted.wait(5):
raise RuntimeError('BEGIN barrier timed out: ' + repr(errors))
if busy and not finished.wait(5):
raise RuntimeError('zero-timeout call did not finish')
clock.now = release_time
timeline.append('host_clock_advanced_to_' + str(release_time))
blocker.commit()
timeline.append('other_writer_lock_released')
finally:
if blocker is not None:
if blocker.in_transaction:
blocker.rollback()
blocker.close()
start.set()
thread.join(10)
if thread.is_alive() or errors:
raise RuntimeError('worker failure: ' + repr(errors))
check = Store(path)
try:
result.update(status=check.view(REVIEWER, ident)['status'],
effects=check.db.execute('SELECT count(*) FROM effects').fetchone()[0],
host_clock=clock.now, clock_samples=clock.samples,
timeline=timeline)
finally:
check.close()
return result
def main():
report = {}
for operation in ('execute', 'decide'):
for fixed in (False, True):
name = operation + ('_locked_clock' if fixed else '_captured_now')
report[name] = contention_case(fixed, operation)
report['execute_before_expiry'] = contention_case(True, release_time=159)
report['execute_busy'] = contention_case(True, busy=True)
print(json.dumps(report, ensure_ascii=False, indent=2))
if __name__ == '__main__':
main()
expected.json 是按这六种情形独立写下的固定预期，没有从实现的返回值生成答案。除两组旧／新执行与决定的对照，还保留“释放时仍为 159”和“立即忙错误”两个控制案例。忙错误案例把 busy_timeout 设为零，并等它结束才释放锁；它验证失败发生在时钟回调之前，并没有实测五秒超时。
expected.json
{
"execute_captured_now": {"outcome":"accepted","status":"consumed","effects":1,"clock_samples":[159],"sampled_at":159},
"execute_locked_clock": {"outcome":"Denied: expired","status":"approved","effects":0,"clock_samples":[160]},
"decide_captured_now": {"outcome":"accepted","status":"approved","effects":0,"clock_samples":[159],"sampled_at":159},
"decide_locked_clock": {"outcome":"Denied: expired","status":"pending","effects":0,"clock_samples":[160]},
"execute_before_expiry": {"outcome":"accepted","status":"consumed","effects":1,"clock_samples":[159],"sampled_at":159},
"execute_busy": {"outcome":"OperationalError: database is locked","status":"approved","effects":0,"clock_samples":[]}
}
python3 -B lock_case.py
python3 -B -m unittest -v test_locked_clock
本次 lock_case.py 的完整标准输出如下。sampled_at 来自回执或 decided 字段；被拒绝的调用没有这个字段。clock_samples 直接记录回调采样值，timeline 记录已协调的因果顺序。验收同时核对 outcome、最终状态和 effects 数量，不能只检查是否打印了 expired。
{
"execute_captured_now": {
"sampled_at": 159,
"outcome": "accepted",
"status": "consumed",
"effects": 1,
"host_clock": 160,
"clock_samples": [
159
],
"timeline": [
"other_writer_lock_acquired_at_159",
"attempt_begin_while_other_writer_holds_lock",
"host_clock_advanced_to_160",
"other_writer_lock_released"
]
},
"execute_locked_clock": {
"outcome": "Denied: expired",
"status": "approved",
"effects": 0,
"host_clock": 160,
"clock_samples": [
160
],
"timeline": [
"other_writer_lock_acquired_at_159",
"attempt_begin_while_other_writer_holds_lock",
"host_clock_advanced_to_160",
"other_writer_lock_released"
]
},
"decide_captured_now": {
"sampled_at": 159,
"outcome": "accepted",
"status": "approved",
"effects": 0,
"host_clock": 160,
"clock_samples": [
159
],
"timeline": [
"other_writer_lock_acquired_at_159",
"attempt_begin_while_other_writer_holds_lock",
"host_clock_advanced_to_160",
"other_writer_lock_released"
]
},
"decide_locked_clock": {
"outcome": "Denied: expired",
"status": "pending",
"effects": 0,
"host_clock": 160,
"clock_samples": [
160
],
"timeline": [
"other_writer_lock_acquired_at_159",
"attempt_begin_while_other_writer_holds_lock",
"host_clock_advanced_to_160",
"other_writer_lock_released"
]
},
"execute_before_expiry": {
"sampled_at": 159,
"outcome": "accepted",
"status": "consumed",
"effects": 1,
"host_clock": 159,
"clock_samples": [
159
],
"timeline": [
"other_writer_lock_acquired_at_159",
"attempt_begin_while_other_writer_holds_lock",
"host_clock_advanced_to_159",
"other_writer_lock_released"
]
},
"execute_busy": {
"outcome": "OperationalError: database is locked",
"status": "approved",
"effects": 0,
"host_clock": 160,
"clock_samples": [],
"timeline": [
"other_writer_lock_acquired_at_159",
"attempt_begin_while_other_writer_holds_lock",
"host_clock_advanced_to_160",
"other_writer_lock_released"
]
}
}
将以下测试也保存在同一目录。六项竞争测试读取上面的固定预期，另十二项检查时间参数误传、角色与摘要、存储绑定、撤销、回滚、单次消费与回执、错误时钟和重新审核。原有 test_approval.py 仍单独运行；它验证原 Store，不应把原有 33 项声称为新适配器已经通过的 33 项。
test_locked_clock.py
import json
from pathlib import Path
import tempfile
import unittest
from approval import AUTHOR, REVIEWER, EXECUTOR, Denied
from locked_clock import LockedClockStore
from lock_case import Clock, PAYLOAD, contention_case
EXPECTED = json.loads(Path(__file__).with_name('expected.json').read_text())
class ContentionTests(unittest.TestCase):
def check_case(self, name, *args, **kwargs):
got = contention_case(*args, **kwargs)
want = EXPECTED[name]
self.assertEqual({key: got[key] for key in want}, want)
self.assertEqual(got['timeline'][0], 'other_writer_lock_acquired_at_159')
self.assertEqual(got['timeline'][1], 'attempt_begin_while_other_writer_holds_lock')
self.assertEqual(got['timeline'][-1], 'other_writer_lock_released')
def test_original_execution_uses_stale_time(self):
self.check_case('execute_captured_now', False)
def test_locked_execution_refuses_at_expiry(self):
self.check_case('execute_locked_clock', True)
def test_original_decision_uses_stale_time(self):
self.check_case('decide_captured_now', False, 'decide')
def test_locked_decision_refuses_at_expiry(self):
self.check_case('decide_locked_clock', True, 'decide')
def test_last_live_second(self):
self.check_case('execute_before_expiry', True, release_time=159)
def test_busy_error_precedes_clock_and_effect(self):
self.check_case('execute_busy', True, busy=True)
class AdapterTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.clock = Clock(110)
self.path = Path(self.tmp.name) / 'test.sqlite'
self.store = LockedClockStore(self.path, self.clock)
self.ident = self.store.propose(AUTHOR, 'original-job', PAYLOAD, 100, 60)
self.digest = self.store.view(REVIEWER, self.ident)['digest']
def tearDown(self):
self.store.close()
self.tmp.cleanup()
def approve(self):
self.store.decide(REVIEWER, self.ident, self.digest, True)
def execute(self, **kwargs):
return self.store.execute(EXECUTOR, self.ident, self.digest, **kwargs)
def effects(self):
return self.store.db.execute('SELECT count(*) FROM effects').fetchone()[0]
def test_no_per_call_time_override(self):
self.approve()
with self.assertRaises(TypeError):
self.execute(now=100)
with self.assertRaises(TypeError):
self.store.execute(EXECUTOR, self.ident, self.digest, 100)
self.assertEqual(self.effects(), 0)
def test_role_and_digest_binding(self):
self.approve()
with self.assertRaises(Denied):
self.store.execute(REVIEWER, self.ident, self.digest)
with self.assertRaises(Denied):
self.store.execute(EXECUTOR, self.ident, 'different-digest')
self.assertEqual(self.effects(), 0)
def test_frozen_payload_revalidated(self):
self.approve()
self.store.db.execute('UPDATE approvals SET expires=999 WHERE id=?', (self.ident,))
with self.assertRaisesRegex(Denied, 'stored binding mismatch'):
self.execute()
self.assertEqual(self.effects(), 0)
def test_revocation_preserved(self):
self.approve()
self.store.revoke(REVIEWER, self.ident, 111)
self.clock.now = 112
with self.assertRaises(Denied):
self.execute()
self.assertEqual(self.effects(), 0)
def test_rollback_preserves_approval(self):
self.approve()
with self.assertRaises(RuntimeError):
self.execute(fail_after_effect=True)
self.assertEqual(self.effects(), 0)
self.assertEqual(self.store.view(REVIEWER, self.ident)['status'], 'approved')
self.execute()
self.assertEqual(self.effects(), 1)
def test_replay_and_receipt_preserved(self):
self.approve()
receipt = self.execute()
with self.assertRaises(Denied):
self.execute()
self.assertEqual(self.effects(), 1)
self.assertEqual(self.store.receipt(EXECUTOR, self.ident), receipt)
def test_invalid_clock_values_rollback(self):
self.approve()
for value in (True, -1, 1.5, float('nan'), None, '120'):
self.clock.now = value
with self.subTest(value=repr(value)), self.assertRaises(ValueError):
self.execute()
self.assertEqual(self.effects(), 0)
self.assertFalse(self.store.db.in_transaction)
def test_backward_clock_rejected(self):
self.approve()
self.clock.now = 109
with self.assertRaisesRegex(Denied, 'clock moved backwards'):
self.execute()
self.assertEqual(self.effects(), 0)
def test_clock_exception_rolls_back(self):
self.approve()
def broken():
raise RuntimeError('clock unavailable')
self.store._clock = broken
with self.assertRaisesRegex(RuntimeError, 'clock unavailable'):
self.execute()
self.assertFalse(self.store.db.in_transaction)
self.assertEqual(self.effects(), 0)
self.store._clock = self.clock
self.execute()
self.assertEqual(self.effects(), 1)
def test_expired_recovery_requires_new_proposal_and_review(self):
self.approve()
self.clock.now = 160
with self.assertRaisesRegex(Denied, 'expired'):
self.execute()
same = self.store.propose(AUTHOR, 'original-job', PAYLOAD, 160, 60)
self.assertEqual(same, self.ident)
self.assertEqual(self.store.view(REVIEWER, same)['expires'], 160)
new = self.store.propose(AUTHOR, 'renewed-job', PAYLOAD, 160, 60)
digest = self.store.view(REVIEWER, new)['digest']
with self.assertRaisesRegex(Denied, 'approval unavailable'):
self.store.execute(EXECUTOR, new, digest)
self.clock.now = 161
self.store.decide(REVIEWER, new, digest, True)
self.clock.now = 162
receipt = self.store.execute(EXECUTOR, new, digest)
self.assertEqual(receipt['executed_at'], 162)
self.assertEqual(receipt['payload'], PAYLOAD)
self.assertEqual(self.effects(), 1)
self.assertIsNone(self.store.receipt(EXECUTOR, self.ident))
def test_expired_decision_does_not_mutate_pending(self):
self.clock.now = 160
with self.assertRaisesRegex(Denied, 'expired'):
self.approve()
self.assertEqual(self.store.view(REVIEWER, self.ident)['status'], 'pending')
self.assertEqual(self.effects(), 0)
def test_server_clock_must_be_callable(self):
with self.assertRaises(TypeError):
LockedClockStore(self.path, 159)
if __name__ == '__main__':
unittest.main()
13. 保留到期记录，用新提案和新审核恢复
遇到 expired，先读取原记录和历史回执。approved 只是数据库状态，不能覆盖已经达到的 expires；pending 也不会因为审核动作曾经排队就获得延期。相同 job 再 propose，仍返回同一份记录和原期限。不要改 expires、把 consumed 重置为 approved，或复制旧批准来续期。
若业务确实还需要执行，由宿主明确创建新提案，重新展示完整目的地、内容、版本、环境和截止时间，再取得新的审核决定。下面演示保留旧记录，在 160 创建新的 pending 提案；新提案在审核前仍被拒绝，经 161 的合成审核后才在 162 产生一条本地效果。代码里的自动调用只是在测试中扮演审核人，不代表生产可以自动代人批准或任由模型换 job。
recovery_demo.py
"""No automatic renewal: preserve the old proposal, then explicitly review a new one."""
import json
from pathlib import Path
import tempfile
from approval import AUTHOR, REVIEWER, EXECUTOR, Denied
from locked_clock import LockedClockStore
from lock_case import Clock, PAYLOAD
def main():
with tempfile.TemporaryDirectory() as root:
clock = Clock(110)
store = LockedClockStore(Path(root) / 'recovery.sqlite', clock)
try:
old = store.propose(AUTHOR, 'job-before-wait', PAYLOAD, 100, 60)
old_digest = store.view(REVIEWER, old)['digest']
store.decide(REVIEWER, old, old_digest, True)
clock.now = 160
try:
store.execute(EXECUTOR, old, old_digest)
except Denied as exc:
print('old execution:', str(exc))
same = store.propose(AUTHOR, 'job-before-wait', PAYLOAD, 160, 60)
print('same job:', same == old, 'expiry:', store.view(REVIEWER, same)['expires'])
new = store.propose(AUTHOR, 'job-after-new-review', PAYLOAD, 160, 60)
new_digest = store.view(REVIEWER, new)['digest']
try:
store.execute(EXECUTOR, new, new_digest)
except Denied as exc:
print('new before review:', str(exc))
clock.now = 161
store.decide(REVIEWER, new, new_digest, True)
clock.now = 162
receipt = store.execute(EXECUTOR, new, new_digest)
print('new after synthetic review:', receipt['executed_at'])
print('payload:', json.dumps(receipt['payload'], ensure_ascii=False, sort_keys=True))
print('old receipt:', store.receipt(EXECUTOR, old))
print('effects:', store.db.execute('SELECT count(*) FROM effects').fetchone()[0])
finally:
store.close()
if __name__ == '__main__':
main()
运行 python3 -B recovery_demo.py，完整输出应与以下 expected-recovery.txt 相同。旧记录没有回执，唯一效果属于新提案；这一步没有重播旧效果或重置旧许可。
expected-recovery.txt
old execution: expired
same job: True expiry: 160
new before review: approval unavailable for consumption
new after synthetic review: 162
payload: {"args": {"body": "这是一条合成数据。", "channel": "preview-a", "title": "锁等待样例"}, "environment": "offline", "tool": "simulate_publish", "version": "publisher-v1"}
old receipt: None
effects: 1
遇到 database is locked，可以在宿主允许的重试预算内保留同一业务标识重新进入受检入口。下一次获得锁之后仍须重新读时钟；忙错误不是绕过校验、延长期限或直接调用工具的理由。时钟返回无效值或抛异常时，先修复时钟来源并读取状态，不能用客户端给出的旧时间兜底。提交前注入故障会回滚本地效果和状态；恢复后也只有尚未到期的许可才可能消费。若已提交但返回值丢失，仍按前文查询 receipt，不再次执行。
测试还需要能抓住写坏的版本。下面四个变体依次把执行取样移到锁前、把到期条件改成严格大于、删掉决定时的期限检查、删掉异常回滚。每个变体只在临时副本里修改代码，然后运行一项针对性测试。只有出现一次断言失败、没有 ERROR 且退出码为 1 才算抓住；导入失败、崩溃或超时不会被算作成功。
check_clock_mutants.py
"""Run targeted tests against deliberate defects in temporary source copies."""
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
ROOT = Path(__file__).resolve().parent
MUTANTS = [
('sample_before_lock', 'locked_clock.py',
' with self.transaction():\n now = self._clock()\n',
' now = self._clock()\n with self.transaction():\n',
'ContentionTests.test_locked_execution_refuses_at_expiry'),
('allow_exact_expiry', 'approval.py',
" if now >= row['expires']:",
" if now > row['expires']:",
'ContentionTests.test_locked_execution_refuses_at_expiry'),
('skip_decision_expiry', 'locked_clock.py',
' self.active(row, now)\n',
' # Deliberate defect: omit the first active check.\n',
'AdapterTests.test_expired_decision_does_not_mutate_pending'),
('leave_transaction_open', 'approval.py',
' self.db.rollback()\n',
' # Deliberate defect: do not roll back.\n',
'AdapterTests.test_clock_exception_rolls_back'),
]
def main():
results = []
for name, file, old, new, test in MUTANTS:
with tempfile.TemporaryDirectory() as temp:
target = Path(temp)
for source in ROOT.iterdir():
if source.suffix in ('.py', '.json'):
shutil.copy2(source, target / source.name)
path = target / file
text = path.read_text(encoding='utf-8')
if old not in text:
raise RuntimeError('mutation anchor missing: ' + name)
path.write_text(text.replace(old, new, 1), encoding='utf-8')
command = [sys.executable, '-B', '-m', 'unittest', '-v',
'test_locked_clock.' + test]
run = subprocess.run(command, cwd=target, text=True,
capture_output=True, timeout=20)
# An import error or crash is not evidence that the oracle caught it.
caught = (run.returncode == 1 and '\nFAIL: ' in run.stderr
and '\nERROR: ' not in run.stderr
and 'FAILED (failures=1)' in run.stderr)
results.append({'mutant': name, 'test': test,
'caught_by_assertion': caught, 'exit': run.returncode})
if not caught:
print(run.stdout, file=sys.stderr)
print(run.stderr, file=sys.stderr)
raise SystemExit('mutant was not caught cleanly: ' + name)
print(json.dumps(results, ensure_ascii=False, indent=2))
if __name__ == '__main__':
main()
python3 -B -m unittest -v test_approval
python3 -B -m unittest -v test_locked_clock
python3 -B check_clock_mutants.py
本次在 Linux、Python 3.12.14、SQLite 3.53.1 运行：原有 33 项通过，新适配器相关 18 项通过，四个错误变体均被预期断言捕获，变体检查器自身退出 0。另以 PYTHONOPTIMIZE=1 复跑，让其子进程也继承优化模式。源码目录、从本页代码块提取的文件和离线 ZIP 解压目录分别复跑并核对输出，未改动固定预期。
[
{
"mutant": "sample_before_lock",
"test": "ContentionTests.test_locked_execution_refuses_at_expiry",
"caught_by_assertion": true,
"exit": 1
},
{
"mutant": "allow_exact_expiry",
"test": "ContentionTests.test_locked_execution_refuses_at_expiry",
"caught_by_assertion": true,
"exit": 1
},
{
"mutant": "skip_decision_expiry",
"test": "AdapterTests.test_expired_decision_does_not_mutate_pending",
"caught_by_assertion": true,
"exit": 1
},
{
"mutant": "leave_transaction_open",
"test": "AdapterTests.test_clock_exception_rolls_back",
"caught_by_assertion": true,
"exit": 1
}
]
最后把保证的时间点说清楚：新适配器保证在成功获得写事务后取样，并据此决定这次本地准入。时钟回调自身可能耗时，检查之后时钟也仍可能继续前进，所以它不保证 COMMIT 在 expires 之前完成，更不保证远端请求在到期前发出或完成。若业务要求这些更强的截止语义，需要另外定义协议与失败处理，不能从这 18 项测试推出来。
本补充的官方资料核对日期为 2026-10-13；当时 Python 3.12 文档显示 3.12.15，实际运行版本为 3.12.14。没有运行真实模型、SDK、HTTP 审批入口、外部工具、可信时间服务或物理时钟跳变测试。自然尺寸 SVG 已离线渲染检查，没有启动浏览器做页面布局验收。本地 SQLite 事务内的效果保证、身份夹具限制和远端结果未知时先核对的要求仍与前文相同。
下载完整审批时钟实验离线包（SHA-256：b3e1228787c92f958ca0cadc6bf631051ec12117aa5a908628993ba7bc9bcd25）。包中包含完整文章、原有三个文件、新适配器、完整竞争与恢复案例、固定预期、测试、错误变体检查器和原创图；不含运行缓存、账户信息或内部执行记录。
14. 接入真实工具前还要补哪些证据
先用当前安装的 SDK 固定版本跑通一次暂停与恢复，保存工具定义版本和审批对象映射，再验证恢复后的实际调用参数逐项等于审核对象。把禁止绕行写进工具入口：所有执行路径都必须经过同一个许可检查，不能另留“内部可信”直通函数。
然后分别验证真实身份、对象级权限、跨租户读取拒绝、审批请求保护、存储访问控制、部署升级后的待审项处理和日志脱敏。若工具会传输数据或产生费用，审核页需要显示相应接收者、数据和承诺范围。某些高影响动作必须交由用户本人完成，产品也应保留这种限制。
最后对真实目标系统设计提交后的失联恢复。没有幂等或查询能力时，结果未知应进入人工核对，不能把“没有收到响应”当成失败再试。到这一层，许可记录才开始与具体业务效果衔接。本文提供的三个文件可以继续作为回归测试底座，远端部分需要自己的集成证据。
摘要和 Actor 是否构成安全认证？
不构成。摘要仅用于一致性检查，Actor 是离线测试夹具；真实服务必须认证主体并保护数据库。
重启以后还能消费许可吗？
可以。尚未撤销、未消费且未到期的批准保存在 SQLite 中；测试使用独立子进程验证。
同一个许可可以重复执行吗？
本例只允许消费一次。重复执行被拒绝，已提交结果通过 receipt 单独查询。
这个示例保证远端发布 exactly-once 吗？
不保证。示例没有远端发布，原子性只涵盖同一 SQLite 事务中的模拟效果与许可状态。