d80df3ed7c
范围（不含 design-story-foundation、docs/design、docs/write-chapter、
craft/、humanization/README.md 等进行中改动）:
1. humanization 规则/样例运行时数据库权威
- db/ddl/111：example_ai_flavor_rule / example_ai_flavor_sample /
example_ai_flavor_rule_event（append-only 生命周期留痕），已应用到 muse-example
- deai/load_db.py：数据库装载器，激活门/重复检测/指纹与文件装载器同源；
数据库失败关闭，不静默回退 Git 文件资产
- humanization/tools/seed_rules_db.py：YAML 种子单事务同步，幂等、
变化留痕、--strict 对 db-only 行失败关闭；真实库已种入 26 规则/107 样例
- prevent/diagnose/revise 生产路径切到数据库读取（--offline 显式读文件），
落库前新鲜度检查与合同声明来源一致；四个 SKILL.md 数据库合同同步
- 真实库验证：规则库指纹与文件种子一致（v-609bc40e21d0b5db），
三个生产脚本端到端从库装载通过
2. PostgreSQL 集成：显式授权后 9/9 通过
- 此前被依赖门阻断的 6 个 _db/smoke 测试全部通过
- extract rollback 冒烟改为回滚事务内自给夹具（pending 窗/草稿缺失时自建），
不再依赖瞬时生产状态；夹具残留核验为 0
3. Skill 行为评测脚手架（真实执行数量仍为 0）
- harness/evals/skill_eval.py：场景合同、六类评测范畴、适配器和结构化裁决报告
- diagnose-ai-flavor 参考场景 4 条 + 管道自测 7 项通过
- 真实模型适配器未授权时以稳定码 EVAL_ADAPTER_UNAVAILABLE 失败关闭；
清单登记 skill_behavior_eval 条目，默认被依赖门阻断
4. evaluate-frozen-replay raw 存储边界冲突
- docs/2026-08-19 备忘录：平台 DB-first 合同（创始人批准）与回放链
仓外 vault 强制的冲突事实、两个选项和裁决前约束；运行时合同未单方面改写
5. harness 自身修复
- runner 对账语义：行为评测入口不参与测试资产双向等值，但登记文件必须存在；
manifest 保留 skill_behavior_eval 布尔字段并校验类型
- 新增 2 条对账回归用例
验证证据: harness 三组自测 15+15+7 通过；静态审计 32 Skill / 0 问题；
76 个非数据库条目通过；9 个 PostgreSQL 集成条目显式授权后通过；
行为评测条目默认阻断；py_compile 与 git diff --check 通过。
未调用真实模型、embedding 或额度；真实行为评测执行数量仍为 0。
100 lines
4.0 KiB
Python
100 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""数据库规则/样例装载器离线测试：与文件装载等价、失败关闭、激活门同源。"""
|
|
import pathlib
|
|
import sys
|
|
import unittest
|
|
|
|
PROJECT_ROOT = pathlib.Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(PROJECT_ROOT / "humanization" / "src"))
|
|
|
|
from deai import load, load_db # noqa: E402
|
|
|
|
|
|
class _Result:
|
|
def __init__(self, rows):
|
|
self.rows = list(rows)
|
|
|
|
def fetchall(self):
|
|
return list(self.rows)
|
|
|
|
|
|
class _FakeConn:
|
|
"""按表名返回 payload 行；可注入执行异常验证失败关闭。"""
|
|
|
|
def __init__(self, rule_payloads=None, sample_payloads=None, *, error=None):
|
|
self.rule_payloads = list(rule_payloads or [])
|
|
self.sample_payloads = list(sample_payloads or [])
|
|
self.error = error
|
|
|
|
def execute(self, sql, params=None):
|
|
if self.error is not None:
|
|
raise self.error
|
|
if "example_ai_flavor_sample" in sql:
|
|
return _Result([(payload,) for payload in self.sample_payloads])
|
|
if "example_ai_flavor_rule" in sql:
|
|
return _Result([(payload,) for payload in self.rule_payloads])
|
|
raise AssertionError(f"未预期的 SQL: {sql}")
|
|
|
|
|
|
def _file_corpus():
|
|
samples = load.load_samples()
|
|
rules = load.load_rules(samples=samples)
|
|
return samples, rules
|
|
|
|
|
|
class LoadDbContractTest(unittest.TestCase):
|
|
def test_db_corpus_equals_file_corpus_and_same_fingerprint(self):
|
|
samples, rules = _file_corpus()
|
|
conn = _FakeConn(
|
|
rule_payloads=[load_db.rule_row(rule)["payload"] for rule in sorted(rules.values(), key=lambda r: r["id"])],
|
|
sample_payloads=[load_db.sample_row(s)["payload"] for s in sorted(samples.values(), key=lambda s: s["id"])],
|
|
)
|
|
db_samples = load_db.load_samples_from_db(conn)
|
|
db_rules = load_db.load_rules_from_db(conn, samples=db_samples)
|
|
self.assertEqual(db_samples, samples)
|
|
self.assertEqual(db_rules, rules)
|
|
self.assertEqual(load_db.canonical_sha(rules), load_db.canonical_sha(db_rules))
|
|
self.assertEqual(load.rule_library_version(db_rules), load.rule_library_version(rules))
|
|
|
|
def test_db_failure_is_fail_closed_without_fallback(self):
|
|
conn = _FakeConn(error=RuntimeError("connection refused"))
|
|
with self.assertRaises(load.LoadError) as ctx:
|
|
load_db.load_rules_from_db(conn)
|
|
self.assertIn("失败关闭", str(ctx.exception))
|
|
self.assertIn("不回退", str(ctx.exception))
|
|
|
|
def test_active_rule_missing_samples_is_rejected_on_db_path(self):
|
|
samples, rules = _file_corpus()
|
|
broken = load_db.rule_row(rules["l001"])["payload"]
|
|
broken = dict(broken)
|
|
broken["samples"] = {"sf": [], "snf": ["snf-l001-01"], "boundary": ["b-l001-01"], "regression": ["reg-l001-01"]}
|
|
conn = _FakeConn(
|
|
rule_payloads=[broken],
|
|
sample_payloads=[load_db.sample_row(s)["payload"] for s in samples.values()],
|
|
)
|
|
with self.assertRaises(load.LoadError) as ctx:
|
|
load_db.load_rules_from_db(conn, samples=load_db.load_samples_from_db(conn))
|
|
self.assertIn("l001", str(ctx.exception))
|
|
|
|
def test_duplicate_rule_id_is_rejected(self):
|
|
samples, rules = _file_corpus()
|
|
payload = load_db.rule_row(rules["l001"])["payload"]
|
|
conn = _FakeConn(rule_payloads=[payload, payload], sample_payloads=[])
|
|
with self.assertRaisesRegex(load.LoadError, "重复"):
|
|
load_db.load_rules_from_db(conn)
|
|
|
|
def test_row_projection_sha_matches_canonical_payload(self):
|
|
samples, rules = _file_corpus()
|
|
for rule in rules.values():
|
|
row = load_db.rule_row(rule)
|
|
self.assertEqual(row["content_sha256"], load_db.canonical_sha(rule))
|
|
self.assertEqual(row["payload"], rule)
|
|
for sample in samples.values():
|
|
row = load_db.sample_row(sample)
|
|
self.assertEqual(row["content_sha256"], load_db.canonical_sha(sample))
|
|
self.assertEqual(row["payload"], sample)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|