217 lines
10 KiB
Python
217 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""人感 v2 共享骨架与技能 5 生命周期的离线负路。"""
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import importlib.util
|
||
import json
|
||
import pathlib
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
|
||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT / "src"))
|
||
|
||
from deai import evaluation, gates, load # noqa: E402
|
||
from deai.baseline import draft_ledger, profile_text, run_voice_gate # noqa: E402
|
||
from deai.carriers import carrier_at, carrier_ranges # noqa: E402
|
||
from deai.diagnose import run_deterministic_rules # noqa: E402
|
||
|
||
|
||
def load_script(name: str, path: pathlib.Path):
|
||
spec = importlib.util.spec_from_file_location(name, path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
assert spec.loader is not None
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
MINE = load_script(
|
||
"mine_ai_flavor_v2",
|
||
ROOT.parent / ".claude/skills/capture-ai-flavor-cases/scripts/mine_ai_flavor.py",
|
||
)
|
||
CAPTURE = load_script(
|
||
"capture_cases_v2",
|
||
ROOT.parent / ".claude/skills/capture-ai-flavor-cases/scripts/capture_cases.py",
|
||
)
|
||
|
||
|
||
class HumanizationV2Test(unittest.TestCase):
|
||
def test_baseline_draft_exposes_sampling_and_unknowns(self):
|
||
text = "".join(f"第{i}句，门外的雨还在落下，青石路上没有一个人。\n" for i in range(1, 65))
|
||
ledger = draft_ledger("synthetic:demo", [{"source_ref": "demo.txt", "text": text}])
|
||
self.assertEqual(ledger["status"], "candidate")
|
||
self.assertTrue(ledger["sampling"]["sample_sufficient"])
|
||
self.assertIn("characters", ledger["unknown_fields"])
|
||
self.assertTrue(ledger["narrator"]["metrics"]["sentence_length"]["median"] > 0)
|
||
|
||
def test_baseline_rejects_source_hash_drift(self):
|
||
with self.assertRaisesRegex(ValueError, "source_sha256"):
|
||
draft_ledger(
|
||
"synthetic:demo",
|
||
[{"source_ref": "demo.txt", "source_sha256": "0" * 64, "text": "甲走进门。"}],
|
||
)
|
||
|
||
def test_baseline_profile_is_deterministic(self):
|
||
text = "甲走进门。乙关上窗。雨落下来。"
|
||
self.assertEqual(profile_text(text), profile_text(text))
|
||
|
||
def test_baseline_duplicate_voice_ownership_is_rejected_by_skill_validator(self):
|
||
establish = load_script(
|
||
"establish_v2",
|
||
ROOT.parent / ".claude/skills/establish-voice-baseline/scripts/establish_voice_baseline.py",
|
||
)
|
||
ledger = {
|
||
"schema_version": "voice-baseline-v1",
|
||
"work_ref": "synthetic:demo",
|
||
"narrator": {"sentence_habits": [], "punctuation_habits": []},
|
||
"characters": {
|
||
"甲": {"verbal_tics": ["嗯"], "sample_lines": ["嗯"]},
|
||
"乙": {"verbal_tics": ["嗯"], "sample_lines": ["嗯"]},
|
||
},
|
||
"untouchable_verbal_tics": {"甲": ["嗯"], "乙": ["嗯"]},
|
||
"protected_spans": [], "blacklist": [],
|
||
}
|
||
with self.assertRaisesRegex(establish.BaselineContractError, "同时归属"):
|
||
establish.validate_ledger(ledger, work_ref="synthetic:demo")
|
||
|
||
def test_carrier_scope_masks_dialogue_and_marks_carve_out(self):
|
||
samples = load.load_samples()
|
||
rules = load.load_rules(samples=samples)
|
||
text = "「值得注意的是，这里是对白。」值得注意的是，这里是叙述。"
|
||
artifact = run_deterministic_rules(text, [rules["l002"]], "v-test", "Audit")
|
||
self.assertEqual(len(artifact["findings"]), 1)
|
||
self.assertEqual(artifact["findings"][0]["carrier"], "narration")
|
||
self.assertEqual(artifact["findings"][0]["decision_proposal"], "ask")
|
||
self.assertEqual(carrier_at(text, 1, 7, carrier_ranges(text)), "dialogue")
|
||
|
||
def test_deterministic_rules_have_contract_replay(self):
|
||
"""确定性触发器规则（regex/handler/density）都要能回放四类夹具。
|
||
|
||
blocking 类机械规则的默认建议就是 repair，属于其合同本身；
|
||
safe_default 约束只对非 blocking 规则断言。
|
||
"""
|
||
samples = load.load_samples()
|
||
rules = load.load_rules(samples=samples)
|
||
deterministic = [rule for rule in rules.values() if rule["trigger"]["type"] != "model_judgment"]
|
||
self.assertGreaterEqual(len(deterministic), 2)
|
||
for rule in deterministic:
|
||
if rule["default_disposition"] == "blocking":
|
||
continue
|
||
report = evaluation.evaluate_rule_contract(rule, samples)
|
||
self.assertTrue(report["contract_pass"], report)
|
||
self.assertEqual(report["dataset_kind"], "shipped_contract_samples")
|
||
|
||
def test_modality_and_time_fact_gates_block_known_regressions(self):
|
||
modality = gates.modality_attribution([{
|
||
"original_exact": "多半背景不凡", "replacement": "背景都不凡",
|
||
}])
|
||
temporal = gates.temporal_fact_delta([{
|
||
"original_exact": "谁也不知道", "replacement": "那年秋分的夜里，谁也不知道",
|
||
}])
|
||
numeric_temporal = gates.temporal_fact_delta([{
|
||
"original_exact": "谁也不知道", "replacement": "2026年3天后的夜里，谁也不知道",
|
||
}])
|
||
self.assertFalse(modality["pass"])
|
||
self.assertFalse(temporal["pass"])
|
||
self.assertFalse(numeric_temporal["pass"])
|
||
|
||
def test_voice_gate_does_not_turn_missing_or_candidate_baseline_into_pass(self):
|
||
result = run_voice_gate("甲走进门。乙关上窗。", "甲走进门。乙关上窗。", None)
|
||
candidate = run_voice_gate(
|
||
"甲走进门。乙关上窗。", "甲走进门。乙关上窗。", {"status": "candidate"}
|
||
)
|
||
self.assertIsNone(result["pass"])
|
||
self.assertEqual(result["status"], "unverified")
|
||
self.assertIsNone(candidate["pass"])
|
||
self.assertEqual(candidate["status"], "unverified")
|
||
|
||
def test_rule_activation_requires_holdout_and_approver(self):
|
||
samples = load.load_samples()
|
||
rules = load.load_rules(samples=samples)
|
||
# s002 已 active；激活门行为用它的 candidate 副本演练，不依赖库里必须留有候选。
|
||
candidate = {**rules["s002"], "status": "candidate"}
|
||
version = candidate["version"]
|
||
contract = evaluation.evaluate_rule_contract(candidate, samples)
|
||
with self.assertRaises(evaluation.EvaluationError):
|
||
evaluation.activate_rule(candidate, contract_report=contract, holdout_report=None, approver="qingse")
|
||
with self.assertRaises(evaluation.EvaluationError):
|
||
evaluation.activate_rule(
|
||
candidate,
|
||
contract_report={**contract, "rule_id": "s002", "rule_version": version},
|
||
holdout_report={"rule_id": "s002", "eligible": True},
|
||
approver="qingse",
|
||
)
|
||
holdout = evaluation.evaluate_holdout(candidate, {
|
||
"dataset_id": "holdout-s002-v1",
|
||
"sf_total": 5,
|
||
"sf_hit": 5,
|
||
"snf_total": 5,
|
||
"snf_false_repair": 0,
|
||
"boundary_total": 2,
|
||
"boundary_false_repair": 0,
|
||
"regression_total": 2,
|
||
"regression_safe": 2,
|
||
})
|
||
full_holdout = {
|
||
**contract,
|
||
"rule_id": "s002", "rule_version": version,
|
||
"eligible": True, "holdout": holdout,
|
||
}
|
||
with self.assertRaises(evaluation.EvaluationError):
|
||
evaluation.activate_rule(
|
||
candidate, contract_report={**contract, "rule_id": "s002", "rule_version": version},
|
||
holdout_report={**full_holdout, "rule_version": version + 1}, approver="qingse"
|
||
)
|
||
active = evaluation.activate_rule(
|
||
candidate, contract_report={**contract, "rule_id": "s002", "rule_version": version},
|
||
holdout_report=full_holdout, approver="qingse"
|
||
)
|
||
self.assertEqual(active["status"], "active")
|
||
self.assertEqual(active["activation"]["approved_by"], "qingse")
|
||
|
||
def test_mining_extra_sample_requires_canonical_verified_projection(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = pathlib.Path(tmp)
|
||
source = root / "source.txt"
|
||
source.write_text("值得注意的是，门外下起了雨。", encoding="utf-8")
|
||
shadow = CAPTURE.capture_file(source, work_ref="synthetic:work", source_license="owned")[0]
|
||
annotated = CAPTURE.annotate_card(shadow, label="sf", carrier="narration")
|
||
verification = {"cards": [CAPTURE.revalidate_card(annotated, source_path=source)]}
|
||
canonical = CAPTURE.confirm_card(
|
||
annotated, reviewer="human", note="功能已确认", verification=verification
|
||
)
|
||
sample = CAPTURE.project_sample(canonical, verification=verification)
|
||
sample_path = root / "samples.json"
|
||
sample_path.write_text(json.dumps({"samples": [sample]}, ensure_ascii=False), encoding="utf-8")
|
||
with self.assertRaises(MINE.MiningError):
|
||
MINE._load_extra_samples([sample_path])
|
||
loaded = MINE._load_extra_samples(
|
||
[sample_path], cards={canonical["id"]: canonical}, verification=verification
|
||
)
|
||
self.assertEqual(loaded[sample["id"]]["case_card_id"], canonical["id"])
|
||
tampered = dict(sample, text="篡改后的样例")
|
||
sample_path.write_text(json.dumps({"samples": [tampered]}, ensure_ascii=False), encoding="utf-8")
|
||
with self.assertRaises(MINE.MiningError):
|
||
MINE._load_extra_samples(
|
||
[sample_path], cards={canonical["id"]: canonical}, verification=verification
|
||
)
|
||
|
||
def test_mining_evaluate_cli_is_offline_by_explicit_flag(self):
|
||
rule_path = ROOT / "rules/structural/s002.yaml"
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
output = pathlib.Path(tmp) / "evaluation.json"
|
||
code = MINE.main([
|
||
"evaluate-rule", "--rule", str(rule_path), "--output", str(output), "--offline",
|
||
])
|
||
self.assertEqual(code, 0)
|
||
report = json.loads(output.read_text(encoding="utf-8"))
|
||
self.assertTrue(report["contract_pass"])
|
||
self.assertFalse(report["eligible"])
|
||
self.assertEqual(report["persistence"]["status"], "offline")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|