Enforce Zhuyuan persona-source epistemic gate

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

@@ -5,6 +5,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import re
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -92,7 +93,23 @@ def compile_graph(contributions: list[dict]) -> dict:
     node_count = 0
     verified_node_count = 0
     correction_edges = []
+    seen_contribution_ids: set[str] = set()
+    seen_stable_numbers: dict[str, str] = {}
     for item in contributions:
+        contribution_id = item["contribution_id"]
+        if contribution_id in seen_contribution_ids:
+            raise ValueError(f"duplicate contribution id: {contribution_id}")
+        seen_contribution_ids.add(contribution_id)
+        match = re.match(r"^(RC-\d{8}-\d{3})(?:-|$)", contribution_id)
+        if not match:
+            raise ValueError(f"invalid contribution stable id: {contribution_id}")
+        stable_number = match.group(1)
+        if stable_number in seen_stable_numbers:
+            raise ValueError(
+                "duplicate contribution stable number: "
+                f"{stable_number} used by {seen_stable_numbers[stable_number]} and {contribution_id}"
+            )
+        seen_stable_numbers[stable_number] = contribution_id
         ordered_nodes = _validate_and_order_nodes(item)
         node_count += len(ordered_nodes)
         verified_node_count += sum(node["state"] == "verified" for node in ordered_nodes)

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

@@ -5,6 +5,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import re
 from pathlib import Path
 
 DEFAULT_REGISTRY = Path(__file__).resolve().parents[1] / "references" / "persona-skill-registry.json"
@@ -30,10 +31,26 @@ def load_registry(path: Path = DEFAULT_REGISTRY) -> dict:
         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")
+    seen_ids: set[str] = set()
+    seen_stable_numbers: dict[str, str] = {}
     for skill in registry["skills"]:
         missing = REQUIRED_SKILL_FIELDS.difference(skill)
         if missing:
             raise ValueError(f"{skill.get('id', '<unknown>')} missing fields: {sorted(missing)}")
+        skill_id = skill["id"]
+        if skill_id in seen_ids:
+            raise ValueError(f"duplicate skill id: {skill_id}")
+        seen_ids.add(skill_id)
+        match = re.match(r"^(GHS-\d{3})(?:-|$)", skill_id)
+        if not match:
+            raise ValueError(f"invalid stable skill id: {skill_id}")
+        stable_number = match.group(1)
+        if stable_number in seen_stable_numbers:
+            raise ValueError(
+                "duplicate stable skill number: "
+                f"{stable_number} used by {seen_stable_numbers[stable_number]} and {skill_id}"
+            )
+        seen_stable_numbers[stable_number] = skill_id
     return registry
 
 
@@ -59,6 +76,66 @@ def select_skill(registry: dict, intent: str) -> tuple[dict | None, int]:
 
 
 def resolve(registry: dict, intent: str, proposed_route: str = "") -> dict:
+    global_forbidden = []
+    global_deprecated = []
+    global_guard_ids = []
+    for candidate in registry["skills"]:
+        if candidate.get("enforcement_scope") != "GLOBAL_PREREQUISITE":
+            continue
+        forbidden_hits = _marker_hits(proposed_route, candidate["forbidden_route_markers"])
+        deprecated_hits = _marker_hits(proposed_route, candidate["deprecated_route_markers"])
+        if forbidden_hits or deprecated_hits:
+            global_guard_ids.append(candidate["id"])
+            global_forbidden.extend(forbidden_hits)
+            global_deprecated.extend(deprecated_hits)
+    if global_forbidden or global_deprecated:
+        return {
+            "decision": "BLOCK" if global_forbidden else "CORRECT",
+            "matched_skill": global_guard_ids[0],
+            "applied_global_guards": global_guard_ids,
+            "hldp_skill": next(
+                skill["hldp_skill"]
+                for skill in registry["skills"]
+                if skill["id"] == global_guard_ids[0]
+            ),
+            "gls_id": next(
+                skill["gls_id"]
+                for skill in registry["skills"]
+                if skill["id"] == global_guard_ids[0]
+            ),
+            "confidence": 1.0,
+            "preferred_route": next(
+                skill["preferred_route"]
+                for skill in registry["skills"]
+                if skill["id"] == global_guard_ids[0]
+            ),
+            "forbidden_hits": global_forbidden,
+            "deprecated_hits": global_deprecated,
+            "correction": "A global persona-source prerequisite rejected the route before task-specific skill selection.",
+            "evidence": next(
+                skill["evidence"]
+                for skill in registry["skills"]
+                if skill["id"] == global_guard_ids[0]
+            ),
+            "freshness": next(
+                skill["freshness"]
+                for skill in registry["skills"]
+                if skill["id"] == global_guard_ids[0]
+            ),
+            "authorization": next(
+                skill["authorization"]
+                for skill in registry["skills"]
+                if skill["id"] == global_guard_ids[0]
+            ),
+            "recovery_route": next(
+                skill.get("recovery_route", [])
+                for skill in registry["skills"]
+                if skill["id"] == global_guard_ids[0]
+            ),
+            "enforcement_level": "PROMOTED_SKILL",
+            "authority_granted": False,
+            "registry_version": registry["version"],
+        }
     skill, score = select_skill(registry, intent)
     if skill is None:
         return {

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

@@ -1,3 +1,4 @@
+import copy
 import unittest
 
 from compile_collective_cognition import (
@@ -44,6 +45,14 @@ class CollectiveCognitionCompilerTests(unittest.TestCase):
         with self.assertRaises(ValueError):
             compile_graph([{**contribution, "nodes": cyclic_nodes}])
 
+    def test_duplicate_contribution_stable_number_fails_closed(self):
+        contribution = load_contributions()[0]
+        duplicate = copy.deepcopy(contribution)
+        stable_number = "-".join(contribution["contribution_id"].split("-")[:3])
+        duplicate["contribution_id"] = f"{stable_number}-ANOTHER-CHAIN"
+        with self.assertRaisesRegex(ValueError, "duplicate contribution stable number"):
+            compile_graph([contribution, duplicate])
+
 
 if __name__ == "__main__":
     unittest.main()

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

@@ -1,4 +1,8 @@
+import copy
+import json
+import tempfile
 import unittest
+from pathlib import Path
 
 from resolve_persona_skill import load_registry, resolve
 
@@ -16,6 +20,39 @@ class PersonaSkillResolverTests(unittest.TestCase):
         self.assertIn("REPO-012", result["preferred_route"])
         self.assertFalse(result["authority_granted"])
 
+    def test_global_persona_source_gate_runs_before_task_specific_selection(self):
+        result = resolve(
+            self.registry,
+            "恢复铸渊人格并开发铸澜手机端",
+            "当前AI自动等于人格",
+        )
+        self.assertEqual(result["decision"], "BLOCK")
+        self.assertEqual(
+            result["matched_skill"],
+            "GHS-017-PERSONA-SOURCE-EPISTEMIC-GATE",
+        )
+        self.assertIn("当前AI自动等于人格", result["forbidden_hits"])
+
+    def test_global_persona_source_gate_blocks_old_brain_receipt_replay(self):
+        result = resolve(
+            self.registry,
+            "继续当前代码仓库开发",
+            "复用上一轮人格脑回执然后发布",
+        )
+        self.assertEqual(result["decision"], "BLOCK")
+        self.assertIn("复用上一轮人格脑回执", result["forbidden_hits"])
+
+    def test_registry_rejects_duplicate_stable_skill_number(self):
+        duplicate = copy.deepcopy(self.registry)
+        duplicate_skill = copy.deepcopy(duplicate["skills"][0])
+        duplicate_skill["id"] = "GHS-017-ANOTHER-SKILL"
+        duplicate["skills"].append(duplicate_skill)
+        with tempfile.TemporaryDirectory() as temp_dir:
+            path = Path(temp_dir) / "registry.json"
+            path.write_text(json.dumps(duplicate), encoding="utf-8")
+            with self.assertRaisesRegex(ValueError, "duplicate stable skill number"):
+                load_registry(path)
+
     def test_corrects_deprecated_server_relay(self):
         result = resolve(
             self.registry,