给 Agent 的 SQLite 报表工具划定只读边界
从一份包裹统计报表出发，验证只读打开模式限制了什么，再把工具收窄到固定查询和可检查的运行预算。
1. 一个报表工具需要多大的数据库能力
给 Agent 接一个物流报表工具，需求很简单：按地区统计已发送或待发送的包裹数量与单位数。表里还放着内部备注，另一张表保存了不属于报表的标签。工具要返回统计值，却没有理由把整份数据库交给调用方。
实现时很容易先写一个接受 sql 字符串的入口，再补上 SELECT 开头、LIMIT 和只读连接。本文从这个小报表出发，逐项验证这些措施到底限制了什么，最后把入口收窄到 state 与 limit 两个参数。所有例子只使用自己生成的临时库；敏感列里的文字也是 synthetic-note-A 这样的合成标记。
实验运行于 Python 3.12.14、SQLite 3.53.1，23 项断言通过。没有连接真实 MCP 服务或模型，没有读取用户数据库、系统数据库、凭据或外部文件，也没有加载扩展。本文提供的是本地执行边界实验，真实用户认证、租户隔离与部署安全需要另行实现。
2. 先把“只读”的含义拆到可验证的操作
SQLite URI 文档定义了 mode=ro 这一打开模式。用它打开本例数据库后，主库 INSERT 会报 attempt to write a readonly database。接着查询 private_note，却仍能得到合成备注；这次读取没有写入主库，因此没有触碰这条限制。
同一连接还能创建 TEMP 表并写入一个数字，也能 ATTACH 一个全新的 :memory: 数据库。后者仅附加空内存库，测试没有尝试读取任何已有文件。这里的观察提醒我们检查连接能力，而不能从一次 INSERT 被拒绝推导出所有操作都已受控。
还有一种边界藏在宿主注册的函数里。反例注册 remember，把传入的 synthetic 追加到 Python list，再返回原值。SELECT remember('synthetic') LIMIT 1 成功执行，list 中增加一个元素。副作用只发生在当前进程的普通列表中，足以说明 SQL 文本的开头和结果行数不能约束函数体。
sqlite3_stmt_readonly 文档同样区分直接数据库修改与函数、虚拟表带来的间接作用。本文没有调用该 C API，也没有验证虚拟表，只借它解释“语句只读”一词的官方边界。准备接入用户定义函数、扩展或非可信数据库文件时，应分别审查实现与来源，不能把已有测试成绩直接移植过去。
3. 固定工具合同，参数只表示值
报表的 SQL、表名、列名、排序以及数据库路径都由宿主固定。state 先检查精确字符串类型，拒绝 null、布尔、数字、数组和对象；调用方只能选择 queued 或 sent，再给出 1 到 20 的整数 limit。布尔值在 Python 中属于 int 的子类，所以示例使用 type(limit) is int，避免把 True 当作 1 接受。非法选项在数据库执行前抛出 ValueError。
state 与 limit 使用问号占位符。测试把看似 SQL 的字符串作为 state 绑定到查询中，结果为空；那段字符串没有成为语法。另一个测试执行 SELECT ? FROM parcels，绑定 units 后拿到的是字符串 units，而不是 units 列里的数字。SQL 表达式文档中的占位符表示运行时值，不能用来选择列或表。需要可选排序时，应由宿主把小型枚举映射到预先审核的固定 SQL。
因此，下面的 report_tool 不是一个通用数据库终端。测试文件会直接执行若干 SQL 来观察连接防护，但模型入口只应暴露 report_tool 的业务参数。path 出现在 Python 函数签名里，是为了让临时 fixture 能传入路径；真实工具注册时必须由可信宿主闭包或配置绑定它，不能把 path 加进模型参数模式。
4. 让 authorizer 看见表、列与函数
连接采用默认拒绝：仅放行 SELECT、main.parcels 上的 zone、units、state，以及 count、sum 函数。READ 回调的空列名用于表级读取情形，不表示开放全部列；本例仍要求 main 数据库和 parcels 表。其他表、private_note、写操作、PRAGMA、ATTACH、TEMP DDL、扩展函数及 random 都被测试为拒绝。
SQLite authorizer在准备语句及必要的重新准备阶段检查操作。回调持续保留到连接关闭；它没有拿到本次业务用户身份，也不会逐行执行租户鉴权。本文选 SQLITE_DENY 使整条语句失败，未采用会让部分读取被替换为 NULL 的 SQLITE_IGNORE。
每次工具调用新建连接，并将 cached_statements 设为 0，省去教学代码中动态权限与语句复用的额外状态。宿主仍掌握连接对象，能够移除回调或注册同名函数。允许 sum 这个名称的前提包含宿主没有把它替换成另一种有副作用的实现；该连接不能作为能力沙箱交给不可信 Python 代码。
本例在安装回调前开启 query_only、关闭 trusted_schema，显式关闭扩展加载，再缩小 SQL 长度、值长度与附加库数量上限。这些是附加措施，不应拿来替代表列允许名单；可对照 query_only 说明与 SQLite 安全建议检查各层职责。Python 的扩展支持取决于构建；如果 enable_load_extension 不存在或调用失败，应停止并检查运行环境，不能为了继续演示而静默跳过。参考 Python sqlite3 API核对本机能力。
5. 保留一个真实的兼容性边缘：COUNT(*)
实验专门记录独立语句 SELECT COUNT(*) FROM parcels 的 READ 回调。本机观察到参数为 (20, 'parcels', '', None, None)：20 对应 SQLITE_READ，列名为空，数据库名为 None。严格要求 database == 'main' 的回调因此拒绝这条语句。完整 GROUP BY 报表中的 COUNT(*) 则成功运行，两者不能只凭表面相似就断言行为相同。
测试会先观察参数，只有当前构建确实给出缺失数据库名时，才检查这条失败关闭分支；其他构建的检查总数可能不同。官方文档要求调用方能处理回调中的空指针，但不能据此把所有 database=None 请求都列入允许名单。
本例还验证 COUNT(units) 可以执行。换用显式列计数必须检查语义：它忽略 NULL，而 COUNT(*) 统计行。本 fixture 的 units 没有 NULL，所以这里得到相同的 3；真实数据不能直接照搬这一替换。若报表以后改变，先用实际 SQL 和目标 SQLite 构建补测试，再决定是否扩展规则，遇到不认识的情况继续拒绝。
6. LIMIT 管输出，progress 管一部分运行预算
向自建表追加一万条合成记录后，测试执行 SELECT SUM(units) FROM parcels LIMIT 1。虽然最多返回一行，SUM 仍需处理聚合输入。将 progress 间隔设为约 100 条 VM 指令，在第 20 次回调返回非零，这条查询报 interrupted。它证明输出行数无法替代执行预算，没有比较索引方案，也没有把数字包装成性能结论。
progress handler 文档把回调间隔定义为近似 VM 指令数，非零返回值用于中断。正式示例选择每约 100 条指令检查一次，到第 1000 次回调中断，约十万条指令只是教学阈值。它并非精确计数器；SQLite 3.41 起，复杂语句准备阶段也可能调用回调。
另一个反例在未防护连接注册一个只 sleep 约 30ms 的函数。间隔设为 1000 时，函数执行完毕，回调次数仍是零。函数体消耗的墙钟时间与 SQLite 指令推进不是同一件事。工具连接没有注册这个函数；它只是说明本实现没有硬墙钟 deadline。
Python connect 的 timeout 主要控制锁等待，不是整个查询的总截止时间。需要更强的运行约束时，应另行设计进程级资源隔离、可靠终止与结果交付流程。本例也未实现总响应字节上限、全应用速率限制和内存隔离；LIMIT 限的是分组数，单个值的长度限制也不能代替总响应大小预算。
7. 从空目录复跑
2026-10-14 清理路径修正：连接打开后，如果 PRAGMA、扩展开关、限额或 authorizer 安装失败，原 connection() 会抛出异常，却没有显式关闭已打开的连接。现在只在初始化区间捕获失败，先尝试 close，再重新抛出原异常；成功 return 后清理由调用方负责。原 report_tool 的 progress 注册也移入受保护区间。若失败清理本身再次报错，本例保留原异常；这表示已经尝试清理，不保证失败的 close 仍能关闭资源。固定 SQL、参数、允许名单、预算与报表输出不变。Python close 文档说明显式关闭连接；连接的 with 块只管理事务，不自动关闭连接。
准备一个空目录，复制本节后面的四个 Python 文件。它们只依赖标准库，不需要安装包或填写密钥。建议使用 Python 3.12 或更新版本，并先记录实际 SQLite 版本。测试仅创建 TemporaryDirectory 内的原创数据，运行结束后清理；请保持这条路径纪律，不要把示例改指向自己的业务数据库。
python -B -c "import platform, sqlite3; print(platform.python_version(), sqlite3.sqlite_version)"
python -B demo.py
python -B test_readonly.pydemo.py 的 rows 应为 east: 1 件、2 单位，以及 west: 1 件、3 单位。progress_callbacks 的数值与 SQLite 构建有关，不应作为业务结果。test_readonly.py 会输出版本、每项名称、通过状态和观察值；断言失败会终止运行。无需逐字比较耗时、错误消息或回调细节，先核对版本与具体断言。
保存为 fixture.py：
"""Create only a new synthetic fixture inside a caller-owned temporary directory."""
import sqlite3
def create_fixture(path):
# Refuse an existing path; callers in this tutorial own the fresh directory.
if path.exists():
raise FileExistsError(path)
db = sqlite3.connect(path)
try:
db.executescript("CREATE TABLE parcels(zone TEXT,units INTEGER,state TEXT,private_note TEXT); CREATE TABLE hidden(label TEXT);")
db.executemany('INSERT INTO parcels VALUES (?,?,?,?)', [
('east', 2, 'sent', 'synthetic-note-A'),
('west', 3, 'sent', 'synthetic-note-B'),
('east', 4, 'queued', 'synthetic-note-C'),
])
db.execute("INSERT INTO hidden VALUES ('synthetic-hidden')")
db.commit()
finally:
db.close()
保存为 readonly_tool.py：
"""Fixed, host-configured analysis tool. No raw SQL argument."""
import sqlite3
from pathlib import Path
READ_COLUMNS = {'zone', 'units', 'state'}
REPORT = '''SELECT zone, COUNT(*) AS parcels, SUM(units) AS units
FROM parcels WHERE state = ?
GROUP BY zone ORDER BY zone LIMIT ?'''
def authorizer(action, arg1, arg2, database, source):
if source is not None: # This example does not need views/triggers.
return sqlite3.SQLITE_DENY
if action == sqlite3.SQLITE_SELECT:
return sqlite3.SQLITE_OK
if action == sqlite3.SQLITE_READ:
# Empty column means table access such as COUNT(*), not every column.
# A missing database name still fails closed; observed on 3.53.1.
return sqlite3.SQLITE_OK if (database == 'main' and arg1 == 'parcels'
and arg2 in READ_COLUMNS | {''}) else sqlite3.SQLITE_DENY
if action == sqlite3.SQLITE_FUNCTION:
return sqlite3.SQLITE_OK if arg2 in {'count', 'sum'} else sqlite3.SQLITE_DENY
return sqlite3.SQLITE_DENY
def connection(path, guarded=False):
db = sqlite3.connect(path.as_uri() + '?mode=ro', uri=True,
isolation_level=None, cached_statements=0)
try:
if guarded:
db.execute('PRAGMA query_only=ON')
db.execute('PRAGMA trusted_schema=OFF')
db.enable_load_extension(False)
db.setlimit(sqlite3.SQLITE_LIMIT_SQL_LENGTH, 8000)
db.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, 100000)
db.setlimit(sqlite3.SQLITE_LIMIT_ATTACHED, 0)
db.set_authorizer(authorizer)
except BaseException:
# Setup owns the connection until successful return. Preserve its error.
try:
db.close()
except BaseException:
pass
raise
return db
def report_tool(path, state, limit=10):
# path is host configuration, never a model argument; no raw-SQL input.
if type(state) is not str or state not in {'queued', 'sent'} or type(limit) is not int or not 1 <= limit <= 20:
raise ValueError('invalid query options')
db = connection(path, guarded=True)
calls = 0
def progress():
nonlocal calls
calls += 1
return calls >= 1000
try:
db.set_progress_handler(progress, 100)
rows = db.execute(REPORT, (state, limit)).fetchall()
except BaseException:
# Registration/query failures keep their original exception.
try:
db.close()
except BaseException:
pass
raise
else:
db.close()
return {'rows': rows, 'progress_callbacks': calls}
保存为 demo.py：
import json
import tempfile
from pathlib import Path
from fixture import create_fixture
from readonly_tool import report_tool
with tempfile.TemporaryDirectory(prefix='readonly-demo-') as folder:
path = Path(folder) / 'synthetic.sqlite'
create_fixture(path)
print(json.dumps(report_tool(path, 'sent'), ensure_ascii=False, indent=2))
保存为 expected-demo.json：
{
"rows": [
[
"east",
1,
2
],
[
"west",
1,
3
]
],
"progress_callbacks": 0
}
8. 独立测试：同时检查成功路径与拒绝路径
以下测试单独导入已保存的 fixture 与工具模块，不把结果预先写成“通过”。拒绝测试只有捕获 DatabaseError 才成功；SQL 意外执行将抛出 AssertionError。测试连接包含无防护反例与加固连接两组，阅读时注意它们的用途。
八个拒绝用例验证的是组合连接策略的可见结果，不能据此把每次拒绝都单独归因于 authorizer：query_only、附加库上限及扩展开关也可能参与限制。测试名称保留 authorizer_denies 前缀以标记这组策略，但没有隔离测量每一层的贡献。
保存为 test_readonly.py：
"""Independent assertions against the shipped tool and synthetic fixture."""
import json
import platform
import sqlite3
import tempfile
import time
from pathlib import Path
from fixture import create_fixture
from readonly_tool import connection, report_tool, REPORT
def run():
results = {'python': platform.python_version(), 'sqlite': sqlite3.sqlite_version, 'checks': []}
def check(name, condition, observation):
assert condition, (name, observation)
results['checks'].append({'name': name, 'pass': True, 'observation': observation})
def denied(db, sql, params=()):
try:
db.execute(sql, params).fetchall()
except sqlite3.DatabaseError as e:
return str(e)
raise AssertionError('Unexpectedly allowed: ' + sql)
with tempfile.TemporaryDirectory(prefix='sql-ro-probe-') as folder:
path = Path(folder) / 'synthetic.sqlite'
create_fixture(path)
db = connection(path)
check('mode_ro_rejects_main_insert', True, denied(db, "INSERT INTO parcels VALUES ('x',1,'sent','synthetic')"))
check('mode_ro_can_read_private_column', db.execute('SELECT private_note FROM parcels LIMIT 1').fetchone() == ('synthetic-note-A',), 'SELECT + LIMIT still reads synthetic private column')
db.execute('CREATE TEMP TABLE scratch(x)')
db.execute('INSERT INTO temp.scratch VALUES (1)')
check('mode_ro_allows_temp_write', db.execute('SELECT * FROM temp.scratch').fetchall() == [(1,)], 'Only in-memory/temp teaching table modified')
db.execute("ATTACH DATABASE ':memory:' AS extra")
check('mode_ro_allows_attach_memory', any(r[1] == 'extra' for r in db.execute('PRAGMA database_list')), 'Attached only empty :memory: database')
effects = []
db.create_function('remember', 1, lambda v: effects.append(v) or v)
sql = "SELECT remember('synthetic') LIMIT 1"
db.execute(sql).fetchall()
check('select_prefix_and_limit_allow_python_side_effect', sql.upper().startswith('SELECT') and effects == ['synthetic'], 'Effect only appends to a Python list')
check('placeholder_is_value_not_identifier', db.execute('SELECT ? FROM parcels LIMIT 1', ('units',)).fetchone() == ('units',), 'Returns string units, not numeric column value')
calls = [0]
db.set_progress_handler(lambda: calls.__setitem__(0, calls[0]+1) or 0, 1000)
db.create_function('pause_briefly', 0, lambda: time.sleep(0.03) or 1)
start=time.monotonic();db.execute('SELECT pause_briefly()').fetchall();elapsed=time.monotonic()-start
check('progress_not_wall_clock_deadline', elapsed >= 0.025 and calls[0] == 0, {'elapsed_seconds': round(elapsed,4), 'callbacks': calls[0]})
db.close()
observed=[]
db=connection(path)
db.set_authorizer(lambda *args: observed.append(args) or sqlite3.SQLITE_OK)
db.execute('SELECT COUNT(*) FROM parcels').fetchall()
count_reads=[x for x in observed if x[0]==sqlite3.SQLITE_READ]
check('count_star_callback_observation', bool(count_reads), count_reads)
db.close()
db = connection(path, True)
for name, sql in [
('private_column', 'SELECT private_note FROM parcels LIMIT 1'),
('hidden_table', 'SELECT label FROM hidden'),
('write', "UPDATE parcels SET units=0"),
('pragma', 'PRAGMA query_only=OFF'),
('attach', "ATTACH DATABASE ':memory:' AS extra"),
('extension_function', "SELECT load_extension('never-loaded')"),
('non_allowlisted_function', "SELECT random()"),
('temp_ddl', 'CREATE TEMP TABLE scratch(x)')]:
check('authorizer_denies_'+name, True, denied(db, sql))
if any(x[3] is None for x in count_reads):
check('count_star_without_db_name_fails_closed', True, denied(db, 'SELECT COUNT(*) FROM parcels'))
check('explicit_column_count_allowed', db.execute('SELECT COUNT(units) FROM parcels').fetchone() == (3,), 'Use an explicitly authorized column when semantically suitable')
db.close()
for bad_state in (None, True, 1, [], {}):
try:
report_tool(path, bad_state)
except ValueError:
pass
else:
raise AssertionError('non-string state accepted')
check('tool_rejects_non_string_state', True, 'null, boolean, number, array, object all raise ValueError')
good = report_tool(path, 'sent')
check('fixed_report_result', good['rows']==[('east',1,2),('west',1,3)], good)
try: report_tool(path, "sent' OR 1=1 --")
except ValueError: check('tool_rejects_unrecognized_state', True, 'Enum validation fails before SQL')
else: raise AssertionError('invalid state accepted')
db = connection(path, True)
check('bound_value_has_no_sql_syntax', db.execute(REPORT, ("sent' OR 1=1 --",10)).fetchall() == [], 'Bound unmatched state returns zero rows')
db.close()
# Populate only our newly created synthetic fixture; never an existing DB.
db=sqlite3.connect(path);db.executemany('INSERT INTO parcels VALUES (?,?,?,?)',[('east',1,'sent','synthetic')]*10000);db.commit();db.close()
db=connection(path, True)
calls=[0]
def budget():
calls[0]+=1
return calls[0]>=20
db.set_progress_handler(budget,100)
error=denied(db, 'SELECT SUM(units) FROM parcels LIMIT 1')
check('limit_one_aggregate_exceeds_step_budget', calls[0]==20 and error=='interrupted', {'callbacks':calls[0],'error':error,'interval_approx_vm_instructions':100})
db.close()
return results
if __name__ == '__main__':
print(json.dumps(run(), ensure_ascii=False, indent=2))
以下是 2026-10-14 新增的初始化失败回归检查。通过 Connection 子类在自建临时库上注入 PRAGMA、扩展开关、限额、authorizer 和 progress 注册异常，检查原异常对象仍被抛出，且 close 恰好调用一次；关闭成功时连接不可再用。另注入 close 自身失败，测试只要求保留原异常与清理尝试，并由测试最后用基类关闭 fixture。它没有测试真实不支持扩展的构建、模型调用或生产环境。旧代码运行这 5 个测试方法产生 12 处失败；本次修正后 5 个方法全部通过，原 23 项检查、整数报表的 19 个方法及 8 个错误版本检查仍通过。整数报表测试中的依赖文件 SHA-256 随这次源码修正同步更新，业务断言不变。
保存为 test_connection_cleanup.py：
"""Injected lifecycle failures on fresh local synthetic SQLite connections only."""
import sqlite3
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
import exact_report
import readonly_tool
from fixture import create_fixture
REAL_CONNECT = sqlite3.connect
class CleanupTests(unittest.TestCase):
def exercise(self, entry, stage, close_fails=False, error_type=sqlite3.OperationalError):
original = error_type('synthetic ' + stage + ' failure')
opened = []
class Connection(sqlite3.Connection):
close_calls = 0
def execute(self, sql, *args):
if stage == 'pragma' and sql == 'PRAGMA query_only=ON':
raise original
return super().execute(sql, *args)
def enable_load_extension(self, enabled):
if stage == 'extension':
raise original
return super().enable_load_extension(enabled)
def setlimit(self, category, limit):
if stage == 'limit':
raise original
return super().setlimit(category, limit)
def set_authorizer(self, callback):
if stage == 'authorizer':
raise original
return super().set_authorizer(callback)
def set_progress_handler(self, callback, interval):
if stage == 'progress':
raise original
return super().set_progress_handler(callback, interval)
def close(self):
self.close_calls += 1
if close_fails:
# Failure occurs before underlying close: verify attempted cleanup,
# not a promise that failing cleanup can close the handle.
raise RuntimeError('synthetic close failure')
return super().close()
def connect(*args, **kwargs):
db = REAL_CONNECT(*args, factory=Connection, **kwargs)
opened.append(db)
return db
with tempfile.TemporaryDirectory(prefix='synthetic-cleanup-') as folder:
path = Path(folder) / 'synthetic.sqlite'
create_fixture(path)
try:
with patch.object(readonly_tool.sqlite3, 'connect', connect):
with self.assertRaises(error_type) as caught:
entry(path, 'sent')
self.assertIs(caught.exception, original)
self.assertEqual(len(opened), 1)
self.assertEqual(opened[0].close_calls, 1)
if not close_fails:
with self.assertRaises(sqlite3.ProgrammingError):
opened[0].execute('SELECT 1')
finally:
for db in opened:
# Test-owned cleanup even when overridden close failed.
sqlite3.Connection.close(db)
def test_guard_setup_closes_for_both_entrypoints(self):
for entry in (readonly_tool.report_tool, exact_report.tool_result):
for stage in ('pragma', 'extension', 'limit', 'authorizer'):
with self.subTest(entry=entry.__name__, stage=stage):
self.exercise(entry, stage)
def test_guard_cleanup_failure_preserves_original(self):
for entry in (readonly_tool.report_tool, exact_report.tool_result):
with self.subTest(entry=entry.__name__):
self.exercise(entry, 'extension', close_fails=True)
def test_original_report_registration_closes(self):
self.exercise(readonly_tool.report_tool, 'progress')
self.exercise(readonly_tool.report_tool, 'progress', close_fails=True)
def test_base_exception_during_setup_closes(self):
self.exercise(readonly_tool.report_tool, 'extension', error_type=KeyboardInterrupt)
def test_successful_ownership_transfer(self):
with tempfile.TemporaryDirectory(prefix='synthetic-cleanup-') as folder:
path = Path(folder) / 'synthetic.sqlite'
create_fixture(path)
db = readonly_tool.connection(path, guarded=True)
try:
self.assertEqual(db.execute('SELECT units FROM parcels WHERE state=?', ('sent',)).fetchall(), [(2,), (3,)])
finally:
db.close()
self.assertEqual(readonly_tool.report_tool(path, 'sent')['rows'], [('east', 1, 2), ('west', 1, 3)])
if __name__ == '__main__':
unittest.main(verbosity=2)
运行新增检查：
python -B -m unittest -v test_connection_cleanup.py9. 本次运行记录与还没有验证的范围
下面保存的是本次重跑输出。版本是 Python 3.12.14 / SQLite 3.53.1，23 项检查均通过。睡眠耗时属于单次观察，不是性能基准或保证；读者机器上的浮点时间和某些错误文字可以不同。
保存为 expected-results.json：
{
"python": "3.12.14",
"sqlite": "3.53.1",
"checks": [
{
"name": "mode_ro_rejects_main_insert",
"pass": true,
"observation": "attempt to write a readonly database"
},
{
"name": "mode_ro_can_read_private_column",
"pass": true,
"observation": "SELECT + LIMIT still reads synthetic private column"
},
{
"name": "mode_ro_allows_temp_write",
"pass": true,
"observation": "Only in-memory/temp teaching table modified"
},
{
"name": "mode_ro_allows_attach_memory",
"pass": true,
"observation": "Attached only empty :memory: database"
},
{
"name": "select_prefix_and_limit_allow_python_side_effect",
"pass": true,
"observation": "Effect only appends to a Python list"
},
{
"name": "placeholder_is_value_not_identifier",
"pass": true,
"observation": "Returns string units, not numeric column value"
},
{
"name": "progress_not_wall_clock_deadline",
"pass": true,
"observation": {
"elapsed_seconds": 0.0302,
"callbacks": 0
}
},
{
"name": "count_star_callback_observation",
"pass": true,
"observation": [
[
20,
"parcels",
"",
null,
null
]
]
},
{
"name": "authorizer_denies_private_column",
"pass": true,
"observation": "access to parcels.private_note is prohibited"
},
{
"name": "authorizer_denies_hidden_table",
"pass": true,
"observation": "access to hidden.label is prohibited"
},
{
"name": "authorizer_denies_write",
"pass": true,
"observation": "not authorized"
},
{
"name": "authorizer_denies_pragma",
"pass": true,
"observation": "not authorized"
},
{
"name": "authorizer_denies_attach",
"pass": true,
"observation": "not authorized"
},
{
"name": "authorizer_denies_extension_function",
"pass": true,
"observation": "not authorized to use function: load_extension"
},
{
"name": "authorizer_denies_non_allowlisted_function",
"pass": true,
"observation": "not authorized to use function: random"
},
{
"name": "authorizer_denies_temp_ddl",
"pass": true,
"observation": "not authorized"
},
{
"name": "count_star_without_db_name_fails_closed",
"pass": true,
"observation": "not authorized"
},
{
"name": "explicit_column_count_allowed",
"pass": true,
"observation": "Use an explicitly authorized column when semantically suitable"
},
{
"name": "tool_rejects_non_string_state",
"pass": true,
"observation": "null, boolean, number, array, object all raise ValueError"
},
{
"name": "fixed_report_result",
"pass": true,
"observation": {
"rows": [
[
"east",
1,
2
],
[
"west",
1,
3
]
],
"progress_callbacks": 0
}
},
{
"name": "tool_rejects_unrecognized_state",
"pass": true,
"observation": "Enum validation fails before SQL"
},
{
"name": "bound_value_has_no_sql_syntax",
"pass": true,
"observation": "Bound unmatched state returns zero rows"
},
{
"name": "limit_one_aggregate_exceeds_step_budget",
"pass": true,
"observation": {
"callbacks": 20,
"error": "interrupted",
"interval_approx_vm_instructions": 100
}
}
]
}
这个报表适用于可信宿主管理的普通合成表，证明了指定参数、固定 SQL 与连接防护可以共同约束本地操作。它没有验证恶意数据库文件、定制虚拟表、真实 UDF 插件、WAL 或并发写入、跨平台构建，也没有测试模型是否会正确调用工具。HTML 与 SVG 做了结构检查，未做浏览器像素验收。
如果数据来自多个用户，入口还必须从可信认证信息得到身份，并实施行级授权谓词或分库等措施。即使只返回聚合数字，也需要判断调用方是否有权看到该统计范围。数据库只读打开模式不会代替这一步。
也不要用 immutable=1 作为“更安全”的开关。该参数向 SQLite 声明文件不会发生变化；文件实际上仍被别处修改时，这个前提会破坏读取正确性。文件来源、宿主函数与扩展代码应分别审查，不能从合成库的 23 项检查推出生产沙箱保证。
实际接线时，保留本例的窄入口，先确认固定查询在目标版本与真实 schema 上的结果，再补上身份、返回大小、并发与终止测试。每增加一个表、函数或查询形状，都应把相应允许条件和拒绝用例一起交付。
10. 补充实验：报表返回的单位数从哪里来
上面的固定查询把可执行操作收窄了。接下来给同一张 parcels 表加入一条 units 为字符串 pending 的合成记录，再调用既有 report_tool。查询成功，east 的件数为 1，单位数却是 0.0。表定义中的 INTEGER 没有拒绝这次写入，SUM 也没有替业务判断 pending 能否参与统计。
本节在 2026-10-13 用 Python 3.12.14 / SQLite 3.53.1 补跑这一类输入。2026-10-13 补充时保留了原有九节、四个 Python 文件、运行输出及其日期；2026-10-14 仅修正前述连接清理路径，并同步依赖文件校验；新增 exact_report.py 给“可交付的整数单位报表”写一份更窄的数据合同。所有写入仍只发生在新建临时目录的合成库中。
SQLite 类型亲和文档说明，普通表的 INTEGER 亲和会尝试转换值，列内仍可能留下 TEXT、REAL、NULL 或 BLOB。下面每一项都是独立创建的数据库，其他列固定为 state=sent、private_note=synthetic-unused。原输出中的三元数组依次是地区、件数、单位数。
| 写入输入 | units 存储类型 | 原 report_tool | 新 tool_result |
|---|---|---|---|
无匹配行 | 无 | [] | {"status":"ok","rows":[]} |
east: 0 | integer × 1 | [["east", 1, 0]] | {"status":"ok","rows":[{"zone":"east","parcels":1,"units_decimal":"0"}]} |
east: NULL | null × 1 | [["east", 1, null]] | {"status":"rejected","reason":"missing_units"} |
east: 'pending' | text × 1 | [["east", 1, 0.0]] | {"status":"rejected","reason":"invalid_units_type"} |
east: '12units' | text × 1 | [["east", 1, 12.0]] | {"status":"rejected","reason":"invalid_units_type"} |
east: 0.5 | real × 1 | [["east", 1, 0.5]] | {"status":"rejected","reason":"invalid_units_type"} |
east: BLOB x'3132' | blob × 1 | [["east", 1, 12.0]] | {"status":"rejected","reason":"invalid_units_type"} |
zone: NULL；units: 1 | integer × 1 | [[null, 1, 1]] | {"status":"rejected","reason":"invalid_zone"} |
east: -1 | integer × 1 | [["east", 1, -1]] | {"status":"rejected","reason":"invalid_units_range"} |
east: '0007' | integer × 1 | [["east", 1, 7]] | {"status":"ok","rows":[{"zone":"east","parcels":1,"units_decimal":"7"}]} |
east: 9007199254740993 | integer × 1 | [["east", 1, 9007199254740993]] | {"status":"ok","rows":[{"zone":"east","parcels":1,"units_decimal":"9007199254740993"}]} |
east: 9223372036854775807；east: 1 | integer × 2 | OperationalError: integer overflow | {"status":"rejected","reason":"sum_out_of_range"} |
east: 2；west: 'pending' | integer × 1, text × 1 | [["east", 1, 2], ["west", 1, 0.0]] | {"status":"rejected","reason":"invalid_units_type"} |
east: 1，共 1001 行 | integer × 1001 | [["east", 1001, 1001]] | {"status":"rejected","reason":"source_limit"} |
zone00 至 zone20，各 1 行、1 单位 | integer × 21 | [["zone00", 1, 1], ["zone01", 1, 1], ["zone02", 1, 1], ["zone03", 1, 1], ["zone04", 1, 1], ["zone05", 1, 1], ["zone06", 1, 1], ["zone07", 1, 1], ["zone08", 1, 1], ["zone09", 1, 1]] | {"status":"rejected","reason":"group_limit"} |
其中 TOTAL 是另开观察连接执行的探针，没有加入原工具的允许名单。单独存入 9007199254740993 时，原 SUM 保留该整数，TOTAL 得到 9007199254740992.0；存入 9223372036854775807 与 1 时，原 SUM 在本机报 integer overflow。SUM / TOTAL 官方说明明确了整数溢出、浮点结果及空输入行为，两个函数的输出合同需要分别验收。
空集合、真实零和缺失单位数也要分开看：GROUP BY 在没有匹配行时产生零组；一行 0 产生一组零；一行 NULL 仍被 COUNT(*) 计为一件，但 SUM 为 NULL。示例观察到 pending 被累计成 0.0，12units 被累计成 12.0。这些成功返回的数值不能替代输入有效性检查。
11. 让完整整数报表有一份明确合同
新增入口只暴露 state。路径仍由可信宿主绑定，SQL 和两个容量常量也由宿主固定。查询沿用原 connection(path, guarded=True)，因此没有修改原 authorizer、schema、扩展开关或 report_tool。它读取已允许的 zone、units、state，使用默认 SQLite 到 Python 类型映射，未注册自定义 converter。
- zone 是 1 至 80 个字符的字符串。units 当前读出后必须是非负 Python int，单值不超过 9223372036854775807；缺失值、文本、小数与 BLOB 一律拒绝，不做 CAST 或默认补零。
- 一次固定 SELECT 最多取得 1001 条匹配记录。出现第 1001 条时整份报告拒绝；1000 条以内才确认本次匹配集合已经完整读出。完整结果最多 20 个地区，超过同样拒绝。
- 使用 Python 整数计数、求和，每组总量也限定在 0 至 9223372036854775807。这个总量上限是本例主动选择的业务域；Python 能计算更大的整数，这里的 sum_out_of_range 表示超出合同。
- 只有全部检查通过，才生成 status=ok 的完整结果。units_decimal 始终是十进制字符串，包括零。拒绝响应只有 status 和 reason，没有先算出的部分 rows。
这份合同检查的是已存储、再读出的值。写入字符串 0007 后，INTEGER 亲和已把它变成整数 7，所以新增工具接受它。整值浮点数、科学计数法文本或布尔输入也可能在写入前后归一化，甚至已经损失原始精度；读端无法还原这些过程。若业务要禁止某种原始拼写或输入类型，应在入库前验证并保留所需来源。本例对 zone 同样只检查存储后的字符串，按 Python 字符串精确分组及排序，适用于所示可信 schema，未适配自定义排序规则。接线时，调用方要同步采用 status 分支与 rows 中的对象字段，并按 units_decimal 字符串接收总量；原 report_tool 的三元组接口继续保留。
1000 是匹配源行上限，SQL 为找齐这些行仍可能扫描更多记录。沿用的 progress handler 继续提供近似 VM 指令预算；它仍不是硬墙钟限制或精确内存上限。SQL 没有 ORDER BY，因为成功路径读取完整的有界集合，聚合完再排序；超过容量只报拒绝，不把任意前缀当作报表。
保存为 exact_report.py：
"""Bounded synthetic report example; no schema/policy changes to original module.
Input path is trusted host configuration, never an exposed model argument.
Results are all-or-nothing; units_decimal is an exact decimal string.
"""
from collections import defaultdict
from readonly_tool import connection
MAX_I64 = 2**63 - 1
SOURCE_LIMIT = 1000
GROUP_LIMIT = 20
READ = 'SELECT zone, units FROM parcels WHERE state = ? LIMIT ?'
class ReportRejected(ValueError):
pass
def exact_report(path, state):
if type(state) is not str or state not in {'queued', 'sent'}:
raise ReportRejected('invalid_state')
db = connection(path, guarded=True)
ticks = 0
def progress():
nonlocal ticks
ticks += 1
return ticks >= 1000
try:
db.set_progress_handler(progress, 100)
# One SELECT. Fetch one extra row to distinguish complete input from truncation.
rows = db.execute(READ, (state, SOURCE_LIMIT + 1)).fetchall()
if len(rows) > SOURCE_LIMIT:
raise ReportRejected('source_limit')
totals = defaultdict(lambda: [0, 0])
# Validate every row before building any successful response.
for zone, units in rows:
if type(zone) is not str or not zone or len(zone) > 80:
raise ReportRejected('invalid_zone')
if units is None:
raise ReportRejected('missing_units')
if type(units) is not int:
raise ReportRejected('invalid_units_type')
if not 0 <= units <= MAX_I64:
raise ReportRejected('invalid_units_range')
totals[zone][0] += 1
totals[zone][1] += units
if len(totals) > GROUP_LIMIT:
raise ReportRejected('group_limit')
if any(total > MAX_I64 for _, total in totals.values()):
raise ReportRejected('sum_out_of_range')
result = {'status': 'ok', 'rows': [
{'zone': zone, 'parcels': count, 'units_decimal': str(total)}
for zone, (count, total) in sorted(totals.items())]}
except BaseException:
# Preserve the primary registration/query/validation error if cleanup fails.
try:
db.close()
except BaseException:
pass
raise
else:
db.close()
return result
def tool_result(path, state):
# Database failures deliberately propagate; never relabel them empty-success.
try:
return exact_report(path, state)
except ReportRejected as error:
return {'status': 'rejected', 'reason': str(error)}
SQLite INTEGER 默认映射为 Python int，REAL 映射为 float，NULL 映射为 None，见 Python sqlite3 类型映射。这也是代码可以直接检查 type(units) 的前提。字符串字段保住交付的十进制字符，接收端还应遵守该字段合同，不能再无条件转为浮点数；这里没有运行浏览器或跨语言客户端实验。
整数报表的失败清理还需覆盖查询主体：若 progress 注册、execute、fetchall 或合同校验已经失败，而 close 随后也失败，单纯的 finally 会让关闭异常成为抛出的异常；原失败虽在异常上下文中，调用方却无法直接按原类型处理。这里让失败分支尝试关闭后重新抛出原异常，成功分支则先关闭再交付结果。成功查询后的 close 失败仍传播，不会交付成功报告。清理失败只表示尝试过关闭，不保证资源已释放；ReportRejected 保持既有拒绝响应合同。
12. 从空目录复跑这组聚合输入
先复制上文四个 Python 文件，再把下列 probe.py、expected.json、fixtures.json、test_exact_report.py、check_mutants.py 和刚才的 exact_report.py 放在同一目录。公开离线包的 source 目录已经包含这些文件，也包含原实验的 fixture.py、demo.py、test_readonly.py 与两份原预期输出。全部仅依赖 Python 标准库。
python -B -c "import platform, sqlite3; print(platform.python_version(), sqlite3.sqlite_version)"
python -B demo.py
python -B test_readonly.py
python -B probe.py
python -B -m unittest -v test_exact_report.py
python -B check_mutants.pyprobe.py 在一个 TemporaryDirectory 中逐个生成新库，不会打开已有业务文件。每组数据只改变 zone 与 units；输入常量完整列在文件中。typeof 与 TOTAL 只在未加业务防护的观察连接上使用，随后才调用原报表与新增入口，因此没有为观察而放宽工具权限。
保存为 probe.py：
"""Creates only fresh synthetic databases, records native SQLite observations."""
import json
from pathlib import Path
import platform
import sqlite3
import tempfile
from exact_report import tool_result
from readonly_tool import report_tool
CASES = {
'empty': [],
'zero': [('east', 0)],
'missing': [('east', None)],
'text': [('east', 'pending')],
'numeric_prefix': [('east', '12units')],
'fraction': [('east', 0.5)],
'blob': [('east', b'12')],
'invalid_zone': [(None, 1)],
'negative': [('east', -1)],
'numeric_text_affinity': [('east', '0007')],
'large_exact': [('east', 9007199254740993)],
'sum_overflow': [('east', 9223372036854775807), ('east', 1)],
'late_bad_group': [('east', 2), ('west', 'pending')],
'source_limit': [('east', 1)] * 1001,
'group_limit': [('zone%02d' % n, 1) for n in range(21)],
}
def fixture(path, rows):
if path.exists():
raise FileExistsError(path)
db = sqlite3.connect(path)
try:
db.execute('CREATE TABLE parcels(zone TEXT, units INTEGER, state TEXT, private_note TEXT)')
db.executemany('INSERT INTO parcels VALUES (?, ?, ?, ?)',
[(zone, value, 'sent', 'synthetic-unused') for zone, value in rows])
db.commit()
finally:
db.close()
def inspect_case(path):
db = sqlite3.connect(path.as_uri() + '?mode=ro', uri=True)
try:
# An unguarded local observation connection, not the original public tool.
storage = db.execute('SELECT typeof(units), COUNT(*) FROM parcels GROUP BY typeof(units) ORDER BY 1').fetchall()
total = db.execute('SELECT TOTAL(units) FROM parcels').fetchone()[0]
finally:
db.close()
try:
native = {'status': 'ok', 'rows': report_tool(path, 'sent')['rows']}
except sqlite3.DatabaseError as error:
native = {'status': 'database_error', 'class': type(error).__name__, 'message': str(error)}
return {'storage_classes': storage, 'original_report': native,
'total_probe': total, 'exact_contract': tool_result(path, 'sent')}
def run():
outcomes = {}
with tempfile.TemporaryDirectory(prefix='synthetic-exact-units-') as folder:
for name, rows in CASES.items():
path = Path(folder) / (name + '.sqlite')
fixture(path, rows)
outcomes[name] = inspect_case(path)
return {'python': platform.python_version(), 'sqlite': sqlite3.sqlite_version,
'scope': 'local synthetic fixtures; no model, network or production database',
'cases': outcomes}
if __name__ == '__main__':
print(json.dumps(run(), ensure_ascii=False, indent=2))
fixtures.json 另存同一组输入的可读清单，测试会核对它与 CASES 一致；重复行使用次数描述，BLOB 使用十六进制标记。
保存为 fixtures.json：
{
"empty": [],
"zero": [
[
"east",
0
]
],
"missing": [
[
"east",
null
]
],
"text": [
[
"east",
"pending"
]
],
"numeric_prefix": [
[
"east",
"12units"
]
],
"fraction": [
[
"east",
0.5
]
],
"blob": [
[
"east",
{
"__bytes_hex__": "3132"
}
]
],
"invalid_zone": [
[
null,
1
]
],
"negative": [
[
"east",
-1
]
],
"numeric_text_affinity": [
[
"east",
"0007"
]
],
"large_exact": [
[
"east",
9007199254740993
]
],
"sum_overflow": [
[
"east",
9223372036854775807
],
[
"east",
1
]
],
"late_bad_group": [
[
"east",
2
],
[
"west",
"pending"
]
],
"source_limit": {
"repeat": 1001,
"row": [
"east",
1
]
},
"group_limit": [
[
"zone00",
1
],
[
"zone01",
1
],
[
"zone02",
1
],
[
"zone03",
1
],
[
"zone04",
1
],
[
"zone05",
1
],
[
"zone06",
1
],
[
"zone07",
1
],
[
"zone08",
1
],
[
"zone09",
1
],
[
"zone10",
1
],
[
"zone11",
1
],
[
"zone12",
1
],
[
"zone13",
1
],
[
"zone14",
1
],
[
"zone15",
1
],
[
"zone16",
1
],
[
"zone17",
1
],
[
"zone18",
1
],
[
"zone19",
1
],
[
"zone20",
1
]
]
}
下面的 expected.json 是按合同逐项写定的预期结果，测试核对它与手写断言一致，变异检查再用它直接对照新增入口，不能用刚跑出的结果覆盖它。原查询的关键反例则由测试中的独立断言固定；本次完整观察值另存为 observed-exact.json，便于逐项对照。
保存为 expected.json：
{
"empty": {
"status": "ok",
"rows": []
},
"zero": {
"status": "ok",
"rows": [
{
"zone": "east",
"parcels": 1,
"units_decimal": "0"
}
]
},
"missing": {
"status": "rejected",
"reason": "missing_units"
},
"text": {
"status": "rejected",
"reason": "invalid_units_type"
},
"numeric_prefix": {
"status": "rejected",
"reason": "invalid_units_type"
},
"fraction": {
"status": "rejected",
"reason": "invalid_units_type"
},
"blob": {
"status": "rejected",
"reason": "invalid_units_type"
},
"invalid_zone": {
"status": "rejected",
"reason": "invalid_zone"
},
"negative": {
"status": "rejected",
"reason": "invalid_units_range"
},
"numeric_text_affinity": {
"status": "ok",
"rows": [
{
"zone": "east",
"parcels": 1,
"units_decimal": "7"
}
]
},
"large_exact": {
"status": "ok",
"rows": [
{
"zone": "east",
"parcels": 1,
"units_decimal": "9007199254740993"
}
]
},
"sum_overflow": {
"status": "rejected",
"reason": "sum_out_of_range"
},
"late_bad_group": {
"status": "rejected",
"reason": "invalid_units_type"
},
"source_limit": {
"status": "rejected",
"reason": "source_limit"
},
"group_limit": {
"status": "rejected",
"reason": "group_limit"
}
}
展开本次原查询、TOTAL 探针与新增合同的完整观察输出
保存为 observed-exact.json：
{
"python": "3.12.14",
"sqlite": "3.53.1",
"scope": "local synthetic fixtures; no model, network or production database",
"cases": {
"empty": {
"storage_classes": [],
"original_report": {
"status": "ok",
"rows": []
},
"total_probe": 0.0,
"exact_contract": {
"status": "ok",
"rows": []
}
},
"zero": {
"storage_classes": [
[
"integer",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1,
0
]
]
},
"total_probe": 0.0,
"exact_contract": {
"status": "ok",
"rows": [
{
"zone": "east",
"parcels": 1,
"units_decimal": "0"
}
]
}
},
"missing": {
"storage_classes": [
[
"null",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1,
null
]
]
},
"total_probe": 0.0,
"exact_contract": {
"status": "rejected",
"reason": "missing_units"
}
},
"text": {
"storage_classes": [
[
"text",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1,
0.0
]
]
},
"total_probe": 0.0,
"exact_contract": {
"status": "rejected",
"reason": "invalid_units_type"
}
},
"numeric_prefix": {
"storage_classes": [
[
"text",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1,
12.0
]
]
},
"total_probe": 12.0,
"exact_contract": {
"status": "rejected",
"reason": "invalid_units_type"
}
},
"fraction": {
"storage_classes": [
[
"real",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1,
0.5
]
]
},
"total_probe": 0.5,
"exact_contract": {
"status": "rejected",
"reason": "invalid_units_type"
}
},
"blob": {
"storage_classes": [
[
"blob",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1,
12.0
]
]
},
"total_probe": 12.0,
"exact_contract": {
"status": "rejected",
"reason": "invalid_units_type"
}
},
"invalid_zone": {
"storage_classes": [
[
"integer",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
null,
1,
1
]
]
},
"total_probe": 1.0,
"exact_contract": {
"status": "rejected",
"reason": "invalid_zone"
}
},
"negative": {
"storage_classes": [
[
"integer",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1,
-1
]
]
},
"total_probe": -1.0,
"exact_contract": {
"status": "rejected",
"reason": "invalid_units_range"
}
},
"numeric_text_affinity": {
"storage_classes": [
[
"integer",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1,
7
]
]
},
"total_probe": 7.0,
"exact_contract": {
"status": "ok",
"rows": [
{
"zone": "east",
"parcels": 1,
"units_decimal": "7"
}
]
}
},
"large_exact": {
"storage_classes": [
[
"integer",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1,
9007199254740993
]
]
},
"total_probe": 9007199254740992.0,
"exact_contract": {
"status": "ok",
"rows": [
{
"zone": "east",
"parcels": 1,
"units_decimal": "9007199254740993"
}
]
}
},
"sum_overflow": {
"storage_classes": [
[
"integer",
2
]
],
"original_report": {
"status": "database_error",
"class": "OperationalError",
"message": "integer overflow"
},
"total_probe": 9.223372036854776e+18,
"exact_contract": {
"status": "rejected",
"reason": "sum_out_of_range"
}
},
"late_bad_group": {
"storage_classes": [
[
"integer",
1
],
[
"text",
1
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1,
2
],
[
"west",
1,
0.0
]
]
},
"total_probe": 2.0,
"exact_contract": {
"status": "rejected",
"reason": "invalid_units_type"
}
},
"source_limit": {
"storage_classes": [
[
"integer",
1001
]
],
"original_report": {
"status": "ok",
"rows": [
[
"east",
1001,
1001
]
]
},
"total_probe": 1001.0,
"exact_contract": {
"status": "rejected",
"reason": "source_limit"
}
},
"group_limit": {
"storage_classes": [
[
"integer",
21
]
],
"original_report": {
"status": "ok",
"rows": [
[
"zone00",
1,
1
],
[
"zone01",
1,
1
],
[
"zone02",
1,
1
],
[
"zone03",
1,
1
],
[
"zone04",
1,
1
],
[
"zone05",
1,
1
],
[
"zone06",
1,
1
],
[
"zone07",
1,
1
],
[
"zone08",
1,
1
],
[
"zone09",
1,
1
]
]
},
"total_probe": 21.0,
"exact_contract": {
"status": "rejected",
"reason": "group_limit"
}
}
}
}
13. 验证拒绝后不会交付半份报告
测试同时检查数据合同和失败传播。east=2、west=pending 这一组先有可累计的值，整份响应仍只有 rejected 与 invalid_units_type。拒绝前后自建库主文件的 SHA-256 相同。恢复测试在单独的测试写连接中把 synthetic west 修正为整数 3，关闭后重新调用，才得到完整的两组结果：
{"status":"ok","rows":[{"zone":"east","parcels":1,"units_decimal":"2"},{"zone":"west","parcels":1,"units_decimal":"3"}]}这次修正只为演示自有 fixture 的恢复步骤，不属于只读工具能力。真实业务中应由有权限的数据负责人核对并修正来源，再从一次新的完整调用开始，不能沿用上次计算了一半的小计。缺失数据库、schema 不符及查询中断须保留数据库执行失败，宿主可以把异常转换为明确的失败响应，不能让它落入空集成功分支。中断测试刻意把回调间隔设为 1 并立即中断，用来稳定检查 SQLITE_INTERRUPT 与连接清理；它没有测量正常阈值的耗时。
保存为 test_exact_report.py：
"""Hand-written business oracles, native local SQLite. No third-party dependencies."""
import hashlib
import json
from pathlib import Path
import sqlite3
import tempfile
import unittest
from unittest.mock import patch
import exact_report
from probe import fixture, run
def ok(zone, count, units):
return {'status': 'ok', 'rows': [
{'zone': zone, 'parcels': count, 'units_decimal': units}]}
EXPECTED = {
'empty': {'status': 'ok', 'rows': []},
'zero': ok('east', 1, '0'),
'missing': {'status': 'rejected', 'reason': 'missing_units'},
'text': {'status': 'rejected', 'reason': 'invalid_units_type'},
'numeric_prefix': {'status': 'rejected', 'reason': 'invalid_units_type'},
'fraction': {'status': 'rejected', 'reason': 'invalid_units_type'},
'blob': {'status': 'rejected', 'reason': 'invalid_units_type'},
'invalid_zone': {'status': 'rejected', 'reason': 'invalid_zone'},
'negative': {'status': 'rejected', 'reason': 'invalid_units_range'},
'numeric_text_affinity': ok('east', 1, '7'),
'large_exact': ok('east', 1, '9007199254740993'),
'sum_overflow': {'status': 'rejected', 'reason': 'sum_out_of_range'},
'late_bad_group': {'status': 'rejected', 'reason': 'invalid_units_type'},
'source_limit': {'status': 'rejected', 'reason': 'source_limit'},
'group_limit': {'status': 'rejected', 'reason': 'group_limit'},
}
class ExactReportTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.observed = run()['cases']
def test_complete_hand_written_contract_oracles(self):
self.assertEqual(set(self.observed), set(EXPECTED))
for name, expected in EXPECTED.items():
with self.subTest(name=name):
self.assertEqual(self.observed[name]['exact_contract'], expected)
def test_integer_affinity_does_not_reject_text_or_fraction(self):
self.assertEqual(self.observed['text']['storage_classes'], [('text', 1)])
self.assertEqual(self.observed['fraction']['storage_classes'], [('real', 1)])
self.assertEqual(self.observed['numeric_text_affinity']['storage_classes'], [('integer', 1)])
def test_native_counterexamples(self):
self.assertEqual(self.observed['text']['original_report']['rows'], [('east', 1, 0.0)])
self.assertEqual(self.observed['numeric_prefix']['original_report']['rows'], [('east', 1, 12.0)])
self.assertEqual(self.observed['missing']['original_report']['rows'], [('east', 1, None)])
def test_total_loses_exact_integer(self):
seen = self.observed['large_exact']
self.assertEqual(seen['original_report']['rows'], [('east', 1, 9007199254740993)])
self.assertEqual(seen['total_probe'], 9007199254740992.0)
self.assertNotEqual(int(seen['total_probe']), 9007199254740993)
def test_native_sum_overflow_is_database_failure(self):
self.assertEqual(self.observed['sum_overflow']['original_report']['status'], 'database_error')
self.assertEqual(self.observed['sum_overflow']['original_report']['class'], 'OperationalError')
def test_rejected_result_cannot_leak_partial_rows(self):
for name, case in self.observed.items():
result = case['exact_contract']
if result['status'] == 'rejected':
self.assertEqual(set(result), {'status', 'reason'}, name)
def test_read_only_and_recovery_after_separate_synthetic_repair(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d) / 'synthetic.sqlite'
fixture(path, [('east', 2), ('west', 'pending')])
before = hashlib.sha256(path.read_bytes()).hexdigest()
self.assertEqual(exact_report.tool_result(path, 'sent'), EXPECTED['late_bad_group'])
self.assertEqual(hashlib.sha256(path.read_bytes()).hexdigest(), before)
writer = sqlite3.connect(path)
try:
writer.execute('UPDATE parcels SET units = ? WHERE zone = ?', (3, 'west'))
writer.commit()
finally:
writer.close()
# A fresh invocation starts from scratch, never joins a prior partial result.
self.assertEqual(exact_report.tool_result(path, 'sent'), {'status': 'ok', 'rows': [
{'zone': 'east', 'parcels': 1, 'units_decimal': '2'},
{'zone': 'west', 'parcels': 1, 'units_decimal': '3'}]})
def test_readonly_open_failure_propagates_and_creates_no_file(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d) / 'missing.sqlite'
with self.assertRaises(sqlite3.DatabaseError):
exact_report.tool_result(path, 'sent')
self.assertFalse(path.exists())
def test_source_limit_exact_boundary(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d) / 'synthetic.sqlite'
fixture(path, [('east', 1)] * 1000)
self.assertEqual(exact_report.tool_result(path, 'sent'), ok('east', 1000, '1000'))
def test_signed_64_bit_total_exact_boundary(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d) / 'synthetic.sqlite'
fixture(path, [('east', 9223372036854775806), ('east', 1)])
self.assertEqual(exact_report.tool_result(path, 'sent'), ok('east', 2, '9223372036854775807'))
def test_state_filter_and_invalid_options(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d) / 'synthetic.sqlite'
fixture(path, [('east', 1)])
self.assertEqual(exact_report.tool_result(path, 'queued'), {'status': 'ok', 'rows': []})
for state in (True, None, [], 'all', "sent' OR 1=1 --"):
self.assertEqual(exact_report.tool_result(path, state),
{'status': 'rejected', 'reason': 'invalid_state'})
def test_closes_connection_on_rejection(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d) / 'synthetic.sqlite'
fixture(path, [('east', None)])
real = exact_report.connection(path, guarded=True)
closed = []
class Spy:
def __getattr__(self, name):
return getattr(real, name)
def close(self):
real.close()
closed.append(True)
with patch.object(exact_report, 'connection', return_value=Spy()):
self.assertEqual(exact_report.tool_result(path, 'sent'), EXPECTED['missing'])
self.assertEqual(closed, [True])
with self.assertRaises(sqlite3.ProgrammingError):
real.execute('SELECT 1')
def test_dependency_matches_cleanup_revision(self):
root = Path(__file__).parent
# This is artifact identity, not an oracle of correct program behavior.
expected = '69d39925eb1f6ab183a63cf333516663bef36f6ba93bda555127eac9e8132873'
self.assertEqual(hashlib.sha256((root / 'readonly_tool.py').read_bytes()).hexdigest(), expected)
def test_standalone_expected_file_matches_handwritten_oracle(self):
root = Path(__file__).parent
self.assertEqual(json.loads((root / 'expected.json').read_text()), EXPECTED)
def test_standalone_fixture_manifest_matches_inputs(self):
from probe import CASES
root = Path(__file__).parent
def decode(value):
if isinstance(value, dict) and set(value) == {'__bytes_hex__'}:
return bytes.fromhex(value['__bytes_hex__'])
if isinstance(value, dict) and set(value) == {'repeat', 'row'}:
if type(value['repeat']) is not int or not 0 <= value['repeat'] <= 1001:
raise ValueError('invalid synthetic repeat count')
if not isinstance(value['row'], list) or len(value['row']) != 2:
raise ValueError('invalid synthetic repeated row')
return [value['row'].copy() for _ in range(value['repeat'])]
return value
serialized = json.loads((root / 'fixtures.json').read_text(), object_hook=decode)
self.assertEqual({name: [tuple(row) for row in rows] for name, rows in serialized.items()}, CASES)
def test_twenty_groups_is_complete_success(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d) / 'synthetic.sqlite'
fixture(path, [('zone%02d' % n, 1) for n in range(20)])
expected = {'status': 'ok', 'rows': [
{'zone': 'zone%02d' % n, 'parcels': 1, 'units_decimal': '1'}
for n in range(20)]}
self.assertEqual(exact_report.tool_result(path, 'sent'), expected)
def test_empty_and_overlong_zone_rejected(self):
with tempfile.TemporaryDirectory() as d:
for i, zone in enumerate(('', 'x' * 81)):
path = Path(d) / ('zone%d.sqlite' % i)
fixture(path, [(zone, 1)])
self.assertEqual(exact_report.tool_result(path, 'sent'), EXPECTED['invalid_zone'])
def test_missing_schema_failure_is_not_empty_success(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d) / 'synthetic.sqlite'
db = sqlite3.connect(path)
db.execute('CREATE TABLE unrelated(x)')
db.close()
with self.assertRaises(sqlite3.DatabaseError):
exact_report.tool_result(path, 'sent')
def test_progress_interrupt_propagates_and_closes(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d) / 'synthetic.sqlite'
fixture(path, [('east', 1)] * 200)
real = exact_report.connection(path, guarded=True)
closed = []
class InterruptImmediately:
def __getattr__(self, name):
return getattr(real, name)
def set_progress_handler(self, callback, interval):
# Deterministic injection, not a performance/budget measurement.
real.set_progress_handler(lambda: 1, 1)
def close(self):
real.close()
closed.append(True)
with patch.object(exact_report, 'connection', return_value=InterruptImmediately()):
with self.assertRaises(sqlite3.OperationalError) as caught:
exact_report.tool_result(path, 'sent')
self.assertEqual(caught.exception.sqlite_errorcode, sqlite3.SQLITE_INTERRUPT)
self.assertEqual(closed, [True])
if __name__ == '__main__':
unittest.main(verbosity=2)
补充的 3 个测试方法覆盖 15 个查询主体与关闭组合：progress 注册、execute、fetchall、业务拒绝及成功结果，每个分别搭配正常关闭、RuntimeError 和 SystemExit。测试使用自有临时库与代理注入，检查原异常对象、只关闭一次、拒绝无部分结果，以及成功路径关闭失败继续传播。修正前 15 个组合中 8 个不满足这些断言，修正后全部通过；原 23 项、19 个整数报表方法、5 个初始化清理方法及 8 个变异检查仍通过。这不是原有初始化检查的重复，也不证明生产 SQLite 会自然出现这些关闭故障。
保存为 test_exact_cleanup.py：
"""Synthetic exact-report body/cleanup failures; no real user database."""
import sqlite3
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
import exact_report
from fixture import create_fixture
class ExactCleanupTests(unittest.TestCase):
def exercise(self, stages):
results=[]
with tempfile.TemporaryDirectory(prefix='owned-') as folder:
p=Path(folder)/'synthetic.sqlite';create_fixture(p)
for stage in stages:
for close_type in [None,RuntimeError,SystemExit]:
db=exact_report.connection(p,guarded=True)
original=KeyboardInterrupt('injected '+stage)
cleanup=close_type('injected cleanup') if close_type else None
calls=[]
class Proxy:
def set_progress_handler(self,*args):
if stage=='progress':raise original
return db.set_progress_handler(*args)
def execute(self,*args):
if stage=='execute':raise original
if stage=='fetchall':
class Cursor:
def fetchall(self):raise original
return Cursor()
if stage=='validation':
class Cursor:
def fetchall(self):return [('east',None)]
return Cursor()
return db.execute(*args)
def close(self):
calls.append('close')
if cleanup:raise cleanup
db.close()
try:
with patch.object(exact_report,'connection',return_value=Proxy()):
try:result=exact_report.tool_result(p,'sent');caught=None
except BaseException as e:caught=e;result=None
if stage in ['progress','execute','fetchall']:passed=caught is original
elif stage=='validation':passed=caught is None and result=={'status':'rejected','reason':'missing_units'}
elif cleanup:passed=caught is cleanup
else:passed=caught is None and result=={'status':'ok','rows':[{'zone':'east','parcels':1,'units_decimal':'2'},{'zone':'west','parcels':1,'units_decimal':'3'}]}
closed=False
try:db.execute('SELECT 1')
except sqlite3.ProgrammingError:closed=True
passed=passed and len(calls)==1 and (cleanup is not None or closed)
results.append({'stage':stage,'close_error':close_type.__name__ if close_type else None,'pass':passed,'raised':type(caught).__name__ if caught else None,'returned':result,'close_attempts':len(calls),'handle_closed':closed})
finally:db.close()
self.assertEqual(len(results),len(stages)*3)
self.assertTrue(all(r['pass'] for r in results),results)
def test_primary_body_failure_preserved(self):
self.exercise(['progress', 'execute', 'fetchall'])
def test_validation_rejection_preserved(self):
self.exercise(['validation'])
def test_success_close_failure_propagates(self):
self.exercise(['success'])
if __name__ == '__main__':
unittest.main(verbosity=2)
运行主体失败与关闭组合检查：
python -B -m unittest -v test_exact_cleanup.py14. 把错误改法放回去，检查测试能否拦住
只检查正确实现会通过，容易遗漏测试自身的问题。下面的脚本在独立命名空间中编译每个错误版本，只把该版本对应的合成库放进临时目录；磁盘上的原实现不变。每次只改一处，实际调用后与 expected.json 的固定结果比较。错误版本若仍返回预期值，脚本会报 mutant survived；语法错误或意外异常也会使脚本失败，不能被计为成功检出。输出列出错误结果和预期结果。
保存为 check_mutants.py：
"""Deliberately wrong in-memory variants must disagree with fixed expected results.
This script never modifies exact_report.py or connects to external databases.
"""
import json
from pathlib import Path
import tempfile
from probe import CASES, fixture
ROOT = Path(__file__).parent
SOURCE = (ROOT / 'exact_report.py').read_text()
EXPECTED = json.loads((ROOT / 'expected.json').read_text())
MUTANTS = [
('null_as_zero', "raise ReportRejected('missing_units')", 'units = 0', 'missing'),
('coerce_float', "raise ReportRejected('invalid_units_type')", 'units = int(units)', 'fraction'),
('truncate_source', "raise ReportRejected('source_limit')", 'rows = rows[:SOURCE_LIMIT]', 'source_limit'),
('ignore_group_limit', "raise ReportRejected('group_limit')", 'pass', 'group_limit'),
('ignore_total_range', "raise ReportRejected('sum_out_of_range')", 'pass', 'sum_overflow'),
('float_total', "'units_decimal': str(total)", "'units_decimal': str(int(float(total)))", 'large_exact'),
('rejection_as_empty', "return {'status': 'rejected', 'reason': str(error)}",
"return {'status': 'ok', 'rows': []}", 'late_bad_group'),
('allow_negative', 'if not 0 <= units <= MAX_I64:', 'if not -1 <= units <= MAX_I64:', 'negative'),
]
def main():
results = []
with tempfile.TemporaryDirectory(prefix='synthetic-mutants-') as folder:
for name, old, new, case in MUTANTS:
if SOURCE.count(old) != 1:
raise AssertionError('mutation anchor mismatch: ' + name)
namespace = {'__name__': 'synthetic_' + name}
exec(compile(SOURCE.replace(old, new), name + '.py', 'exec'), namespace)
path = Path(folder) / (name + '.sqlite')
fixture(path, CASES[case])
actual = namespace['tool_result'](path, 'sent')
if actual == EXPECTED[case]:
raise AssertionError('mutant survived: ' + name)
results.append({'mutant': name, 'fixture': case, 'caught': True,
'wrong_result': actual, 'expected': EXPECTED[case]})
print(json.dumps({'scope': 'local in-memory deliberate bugs', 'mutants': results}, indent=2))
if __name__ == '__main__':
main()
本次本地复跑的新增测试为 19 个 unittest 测试方法，全部通过，其中逐项核对 15 组输入。8 个故意错误的版本全部产生与固定预期不同的结果。另保留原实验的 23 项检查；这些数字分别描述各自的检查集合，不代表生产覆盖率。
TOTAL 探针本身也是一个负面对照：不产生整数溢出异常，仍会让 9007199254740993 丢掉最后一位。验收时要固定期待的整数或拒绝状态，不能只断言查询没有报错。把超量输入截为前 1000 行、把 NULL 当零、把总量转 float，以及取消总量业务上限，都会与相应固定预期不符。
15. 收到拒绝或执行失败后怎样继续
- missing_units、invalid_units_type、invalid_units_range：核对导入来源、缺失标记与单位定义。保留失败原因，别把坏行丢掉后仍称为完整统计。
- invalid_zone：检查地区字段的缺失、类型及长度。当前合同不做清理、合并或别名归一化，来源修正后再重新生成。
- sum_out_of_range：先检查单位是否用错、是否重复导入，或业务是否真的需要更大范围。若要扩大范围，应连同接收端字段合同和边界测试一起修改。
- source_limit、group_limit：本次完整报告没有形成。重新明确筛选范围，或设计经验证的全量处理方式；只提高数字不能自动解决扫描耗时与内存开销。
- DatabaseError：调查文件路径、schema、权限和中断原因。保留执行失败，关闭连接；问题解决后重新读取整个目标集合。
补充实验只覆盖这里的合成普通表、默认类型映射与有界整数统计。没有验证生产数据库、恶意文件、虚拟表、自定义 converter、并发写入、多版本兼容、真实模型调用、总响应字节上限或硬 deadline。既有只读边界和身份授权要求继续适用。SVG 已做本地图像检查，HTML 做结构与源码提取校验，未做浏览器像素验收。
本节关注数据已经存入 SQLite 之后，哪些值可以参与精确汇总及怎样交付完整结果。备份时点与跨页一致性仍分别需要备份、事务和分页实验来验证；这里没有新增这些结论。
可以把模型生成的任意 SQL 传进这个工具吗？
本例仅开放 state 和 limit。SQL、路径及允许对象由可信宿主固定，没有通用 SQL 参数。
mode=ro 是否会隐藏内部备注？
不会。本实验能读取合成 private_note；表列权限由额外的 authorizer 规则限制，业务用户授权还需独立实现。
progress handler 能确保查询在指定毫秒内结束吗？
不能。回调依据近似 VM 指令推进，本文的约 30ms 函数在零次回调时结束。硬墙钟限制需另行设计。
COUNT(*) 被拒绝时能允许所有 database=None 吗？
不应如此放宽。先观察目标版本与具体 SQL 的回调参数，保持失败关闭；显式列计数还需检查 NULL 语义。