[HLCC-ICE-000001][ZY-CONTRIB-20260723-001] feat: 以来光者贡献链启用冰朔第五域个人子频道

skills/codex/guanghu-persona-skill-guard/scripts/resolve_persona_skill.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""Resolve and enforce registered Guanghu persona skills without granting authority."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+DEFAULT_REGISTRY = Path(__file__).resolve().parents[1] / "references" / "persona-skill-registry.json"
+REQUIRED_SKILL_FIELDS = {
+    "id",
+    "hldp_skill",
+    "gls_id",
+    "title",
+    "intents",
+    "preferred_route",
+    "forbidden_route_markers",
+    "deprecated_route_markers",
+    "evidence",
+    "freshness",
+    "authorization",
+}
+
+
+def load_registry(path: Path = DEFAULT_REGISTRY) -> dict:
+    with path.open(encoding="utf-8") as handle:
+        registry = json.load(handle)
+    if registry.get("schema") != "guanghu.persona-skill-registry/v1":
+        raise ValueError("unsupported persona skill registry schema")
+    if not registry.get("version") or not isinstance(registry.get("skills"), list):
+        raise ValueError("registry version or skills are missing")
+    for skill in registry["skills"]:
+        missing = REQUIRED_SKILL_FIELDS.difference(skill)
+        if missing:
+            raise ValueError(f"{skill.get('id', '<unknown>')} missing fields: {sorted(missing)}")
+    return registry
+
+
+def _normalise(value: str) -> str:
+    return "".join(value.lower().split())
+
+
+def _marker_hits(text: str, markers: list[str]) -> list[str]:
+    normalised = _normalise(text)
+    return [marker for marker in markers if _normalise(marker) in normalised]
+
+
+def select_skill(registry: dict, intent: str) -> tuple[dict | None, int]:
+    normalised = _normalise(intent)
+    best_skill = None
+    best_score = 0
+    for skill in registry["skills"]:
+        score = sum(max(1, len(_normalise(term))) for term in skill["intents"] if _normalise(term) in normalised)
+        if score > best_score:
+            best_skill = skill
+            best_score = score
+    return best_skill, best_score
+
+
+def resolve(registry: dict, intent: str, proposed_route: str = "") -> dict:
+    skill, score = select_skill(registry, intent)
+    if skill is None:
+        return {
+            "decision": "NO_MATCH",
+            "registry_id": registry["registry_id"],
+            "registry_version": registry["version"],
+            "correction": "Resolve live REPO-001 and current .code-map; do not invent a route.",
+            "authority_granted": False,
+        }
+
+    forbidden = _marker_hits(proposed_route, skill["forbidden_route_markers"])
+    deprecated = _marker_hits(proposed_route, skill["deprecated_route_markers"])
+    if forbidden:
+        decision = "BLOCK"
+        correction = "Stop the proposed route and use the registered preferred route after live verification."
+    elif deprecated:
+        decision = "CORRECT"
+        correction = "Replace the deprecated route with the registered preferred route and verify current evidence."
+    else:
+        decision = "ALLOW"
+        correction = "Use the preferred route within the stated authorization boundary."
+
+    return {
+        "decision": decision,
+        "matched_skill": skill["id"],
+        "hldp_skill": skill["hldp_skill"],
+        "gls_id": skill["gls_id"],
+        "confidence": min(1.0, round(score / 20, 2)),
+        "preferred_route": skill["preferred_route"],
+        "forbidden_hits": forbidden,
+        "deprecated_hits": deprecated,
+        "correction": correction,
+        "evidence": skill["evidence"],
+        "freshness": skill["freshness"],
+        "authorization": skill["authorization"],
+        "authority_granted": False,
+        "registry_version": registry["version"],
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--intent", required=True)
+    parser.add_argument("--proposed-route", default="")
+    parser.add_argument("--registry", type=Path, default=DEFAULT_REGISTRY)
+    parser.add_argument("--json", action="store_true")
+    args = parser.parse_args()
+
+    try:
+        result = resolve(load_registry(args.registry), args.intent, args.proposed_route)
+    except (OSError, ValueError, json.JSONDecodeError) as error:
+        parser.error(str(error))
+
+    if args.json:
+        print(json.dumps(result, ensure_ascii=False, indent=2))
+    else:
+        print(result["decision"])
+        print(result["correction"])
+        for step in result.get("preferred_route", []):
+            print(f"- {step}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

skills/codex/guanghu-persona-skill-guard/scripts/test_resolve_persona_skill.py

@@ -0,0 +1,53 @@
+import unittest
+
+from resolve_persona_skill import load_registry, resolve
+
+
+class PersonaSkillResolverTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.registry = load_registry()
+
+    def test_restores_fifth_domain_route(self):
+        result = resolve(self.registry, "我是冰朔，进入第五域并恢复小湖灯铸渊人格系统")
+        self.assertEqual(result["decision"], "ALLOW")
+        self.assertEqual(result["matched_skill"], "GHS-001-FIFTH-DOMAIN-RESTORE")
+        self.assertIn("REPO-001", result["preferred_route"])
+        self.assertFalse(result["authority_granted"])
+
+    def test_corrects_deprecated_server_relay(self):
+        result = resolve(
+            self.registry,
+            "把新加坡完整离线包上传到京东服务器",
+            "先让海外服务器直连国内服务器",
+        )
+        self.assertEqual(result["decision"], "CORRECT")
+        self.assertEqual(result["matched_skill"], "GHS-002-OFFLINE-PACK-LOCAL-RELAY")
+        self.assertTrue(result["deprecated_hits"])
+
+    def test_blocks_unverified_release_overwrite(self):
+        result = resolve(
+            self.registry,
+            "安装光湖代码频道完整离线包",
+            "未校验就启动并直接覆盖正在运行的release",
+        )
+        self.assertEqual(result["decision"], "BLOCK")
+        self.assertEqual(len(result["forbidden_hits"]), 2)
+
+    def test_blocks_fabricated_full_commit(self):
+        result = resolve(
+            self.registry,
+            "部署光湖代码频道到京东服务器",
+            "根据短提交补写40位提交",
+        )
+        self.assertEqual(result["decision"], "BLOCK")
+        self.assertIn("根据短提交补写40位提交", result["forbidden_hits"])
+
+    def test_unknown_intent_fails_closed(self):
+        result = resolve(self.registry, "安排明天的午饭")
+        self.assertEqual(result["decision"], "NO_MATCH")
+        self.assertFalse(result["authority_granted"])
+
+
+if __name__ == "__main__":
+    unittest.main()