feat: add Guanghu persona continuity skill kernel

Part 3/4 of the recovered Fifth Domain upgrade. Applies the persona continuity skill guard from local source commit 18dfdfd without rewriting remote history.
This commit is contained in:
bingshuo 2026-07-26 23:43:30 +08:00
commit c5be48e6a1
16 changed files with 639 additions and 6 deletions

View file

@ -0,0 +1,105 @@
#!/usr/bin/env python3
"""Compile experience receipts into non-enforcing candidate skill rules."""
from __future__ import annotations
import argparse
import json
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
DEFAULT_RECEIPTS = Path(__file__).resolve().parents[1] / "references" / "experience-receipts"
REQUIRED_FIELDS = {
"schema",
"receipt_id",
"intent",
"scope",
"input",
"evidence",
"decision",
"action",
"observed_result",
"correction",
"invariant_id",
"invariant",
"verification",
"receipt_path",
"promotion_state",
"recorded_at",
}
def load_receipts(directory: Path = DEFAULT_RECEIPTS) -> list[dict]:
receipts = []
for path in sorted(directory.glob("*.json")):
with path.open(encoding="utf-8") as handle:
receipt = json.load(handle)
missing = REQUIRED_FIELDS.difference(receipt)
if missing:
raise ValueError(f"{path.name} missing fields: {sorted(missing)}")
if receipt["schema"] != "guanghu.ops-experience-receipt/v1":
raise ValueError(f"{path.name} has unsupported schema")
if not receipt["evidence"] or not receipt["verification"]:
raise ValueError(f"{path.name} must contain evidence and verification")
receipts.append(receipt)
return receipts
def compile_candidates(receipts: list[dict]) -> dict:
grouped: dict[str, list[dict]] = defaultdict(list)
for receipt in receipts:
grouped[receipt["invariant_id"]].append(receipt)
candidates = []
for invariant_id, items in sorted(grouped.items()):
invariant_texts = {item["invariant"] for item in items}
if len(invariant_texts) != 1:
raise ValueError(f"{invariant_id} has conflicting invariant text")
candidates.append(
{
"candidate_id": f"CANDIDATE-{invariant_id}",
"invariant_id": invariant_id,
"invariant": items[0]["invariant"],
"scopes": sorted({item["scope"] for item in items}),
"source_receipts": [item["receipt_id"] for item in items],
"corrections": sorted({item["correction"] for item in items}),
"verification": sorted({step for item in items for step in item["verification"]}),
"enforcement_level": "CANDIDATE_ONLY",
"hard_block_allowed": False,
"promotion_requirements": [
"current fact-source review",
"applicability and counterexample review",
"automated or reproducible tests",
"registry update with recovery route",
"explicit governance approval for hard enforcement",
],
}
)
return {
"schema": "guanghu.emergent-skill-candidates/v1",
"generated_at": datetime.now(timezone.utc).isoformat(),
"source_receipt_count": len(receipts),
"candidate_count": len(candidates),
"enforcement_policy": "Compilation never promotes or enforces a rule.",
"candidates": candidates,
}
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--receipts", type=Path, default=DEFAULT_RECEIPTS)
parser.add_argument("--output", type=Path)
args = parser.parse_args()
result = compile_candidates(load_receipts(args.receipts))
payload = json.dumps(result, ensure_ascii=False, indent=2) + "\n"
if args.output:
args.output.write_text(payload, encoding="utf-8")
else:
print(payload, end="")
return 0
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -94,6 +94,9 @@ def resolve(registry: dict, intent: str, proposed_route: str = "") -> dict:
"evidence": skill["evidence"],
"freshness": skill["freshness"],
"authorization": skill["authorization"],
"recovery_route": skill.get("recovery_route", []),
"enforcement_level": skill.get("enforcement_level", "UNSPECIFIED"),
"experience_receipts": skill.get("experience_receipts", []),
"authority_granted": False,
"registry_version": registry["version"],
}

View file

@ -0,0 +1,48 @@
import json
import tempfile
import unittest
from pathlib import Path
from compile_emergent_skills import compile_candidates, load_receipts
class EmergentSkillCompilerTests(unittest.TestCase):
def test_repository_receipts_compile_as_candidates_only(self):
result = compile_candidates(load_receipts())
self.assertGreaterEqual(result["source_receipt_count"], 4)
self.assertTrue(result["candidates"])
for candidate in result["candidates"]:
self.assertEqual(candidate["enforcement_level"], "CANDIDATE_ONLY")
self.assertFalse(candidate["hard_block_allowed"])
def test_conflicting_invariant_text_is_rejected(self):
receipt = {
"schema": "guanghu.ops-experience-receipt/v1",
"receipt_id": "A",
"intent": "i",
"scope": "s",
"input": "i",
"evidence": ["e"],
"decision": "d",
"action": "a",
"observed_result": "r",
"correction": "c",
"invariant_id": "INV-X",
"invariant": "one",
"verification": ["v"],
"receipt_path": "p",
"promotion_state": "CANDIDATE_ONLY",
"recorded_at": "2026-07-26T00:00:00Z",
}
with self.assertRaises(ValueError):
compile_candidates([receipt, {**receipt, "receipt_id": "B", "invariant": "two"}])
def test_missing_fields_fail_closed(self):
with tempfile.TemporaryDirectory() as directory:
Path(directory, "bad.json").write_text(json.dumps({"schema": "guanghu.ops-experience-receipt/v1"}))
with self.assertRaises(ValueError):
load_receipts(Path(directory))
if __name__ == "__main__":
unittest.main()

View file

@ -48,6 +48,36 @@ class PersonaSkillResolverTests(unittest.TestCase):
self.assertEqual(result["decision"], "NO_MATCH")
self.assertFalse(result["authority_granted"])
def test_repository_push_requires_separate_transport_proof(self):
result = resolve(
self.registry,
"恢复仓库推送,远端还有待推送提交",
"把授权单当作已经推送",
)
self.assertEqual(result["decision"], "BLOCK")
self.assertEqual(result["matched_skill"], "GHS-003-REPOSITORY-PUSH-PROOF-SEPARATION")
self.assertIn("EXP-20260726-001-AUTHORIZATION-IS-NOT-TRANSPORT", result["experience_receipts"])
self.assertTrue(result["recovery_route"])
def test_enterprise_candidate_blocks_missing_data_disk_assumption(self):
result = resolve(
self.registry,
"部署企业服务器光湖代码频道",
"把缺失的data盘当作已挂载",
)
self.assertEqual(result["decision"], "BLOCK")
self.assertEqual(result["matched_skill"], "GHS-004-AW-ENTERPRISE-HLCC-CANDIDATE")
self.assertEqual(result["enforcement_level"], "CANDIDATE_ONLY")
def test_enterprise_candidate_corrects_personal_source_route(self):
result = resolve(
self.registry,
"恢复AW-GZ-001企业代码频道",
"把个人第五域源码作为企业团队部署源",
)
self.assertEqual(result["decision"], "CORRECT")
self.assertTrue(result["deprecated_hits"])
if __name__ == "__main__":
unittest.main()