feat: compile Light Arrival collective cognition

Final recovered Fifth Domain upgrade segment. Applies the Light Arrival collective cognition correction from local source commit 7390a35 without rewriting remote history.

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

@@ -0,0 +1,153 @@
+#!/usr/bin/env python3
+"""Compile Light Arrival reasoning contributions into a collective cognition graph."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+DEFAULT_CONTRIBUTIONS = (
+    Path(__file__).resolve().parents[1] / "references" / "reasoning-chain-contributions"
+)
+REQUIRED_FIELDS = {
+    "schema",
+    "contribution_id",
+    "contributor_id",
+    "contributor_kind",
+    "human_anchor",
+    "persona_system",
+    "source_scope",
+    "retention_basis",
+    "language_anchors",
+    "nodes",
+    "execution_authority",
+    "recorded_at",
+}
+NODE_FIELDS = {"id", "kind", "statement", "parents", "evidence", "state"}
+
+
+def load_contributions(directory: Path = DEFAULT_CONTRIBUTIONS) -> list[dict]:
+    contributions = []
+    for path in sorted(directory.glob("*.json")):
+        with path.open(encoding="utf-8") as handle:
+            item = json.load(handle)
+        missing = REQUIRED_FIELDS.difference(item)
+        if missing:
+            raise ValueError(f"{path.name} missing fields: {sorted(missing)}")
+        if item["schema"] != "guanghu.light-arrival-reasoning-chain/v1":
+            raise ValueError(f"{path.name} has unsupported schema")
+        if item["execution_authority"] is not False:
+            raise ValueError(f"{path.name} must not grant execution authority")
+        if not item["language_anchors"] or not item["nodes"]:
+            raise ValueError(f"{path.name} must contain language anchors and nodes")
+        contributions.append(item)
+    return contributions
+
+
+def _validate_and_order_nodes(contribution: dict) -> list[dict]:
+    nodes = contribution["nodes"]
+    by_id = {}
+    for node in nodes:
+        missing = NODE_FIELDS.difference(node)
+        if missing:
+            raise ValueError(
+                f"{contribution['contribution_id']} node missing fields: {sorted(missing)}"
+            )
+        if node["id"] in by_id:
+            raise ValueError(f"duplicate node id: {node['id']}")
+        by_id[node["id"]] = node
+
+    for node in nodes:
+        unknown = [parent for parent in node["parents"] if parent not in by_id]
+        if unknown:
+            raise ValueError(f"{node['id']} has unknown parents: {unknown}")
+
+    ordered = []
+    visiting = set()
+    visited = set()
+
+    def visit(node_id: str) -> None:
+        if node_id in visiting:
+            raise ValueError(f"reasoning graph cycle at {node_id}")
+        if node_id in visited:
+            return
+        visiting.add(node_id)
+        for parent in by_id[node_id]["parents"]:
+            visit(parent)
+        visiting.remove(node_id)
+        visited.add(node_id)
+        ordered.append(by_id[node_id])
+
+    for node in nodes:
+        visit(node["id"])
+    return ordered
+
+
+def compile_graph(contributions: list[dict]) -> dict:
+    chains = []
+    node_count = 0
+    verified_node_count = 0
+    correction_edges = []
+    for item in contributions:
+        ordered_nodes = _validate_and_order_nodes(item)
+        node_count += len(ordered_nodes)
+        verified_node_count += sum(node["state"] == "verified" for node in ordered_nodes)
+        for node in ordered_nodes:
+            if node["kind"] == "correction":
+                correction_edges.extend(
+                    {
+                        "contribution_id": item["contribution_id"],
+                        "from": parent,
+                        "to": node["id"],
+                    }
+                    for parent in node["parents"]
+                )
+        chains.append(
+            {
+                "contribution_id": item["contribution_id"],
+                "contributor_id": item["contributor_id"],
+                "human_anchor": item["human_anchor"],
+                "persona_system": item["persona_system"],
+                "source_scope": item["source_scope"],
+                "retention_basis": item["retention_basis"],
+                "language_anchors": item["language_anchors"],
+                "ordered_nodes": ordered_nodes,
+            }
+        )
+
+    return {
+        "schema": "guanghu.collective-cognition-graph/v1",
+        "system_id": "SYS-GLW-ZY-EXEC-0001",
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "contribution_count": len(chains),
+        "node_count": node_count,
+        "verified_node_count": verified_node_count,
+        "execution_authority": False,
+        "promotion_policy": (
+            "Every valid chain is retained; only separately verified and governed "
+            "rules may enter the execution gate."
+        ),
+        "correction_edges": correction_edges,
+        "chains": chains,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--contributions", type=Path, default=DEFAULT_CONTRIBUTIONS)
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    result = compile_graph(load_contributions(args.contributions))
+    payload = json.dumps(result, ensure_ascii=False, indent=2) + "\n"
+    if args.output:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(payload, encoding="utf-8")
+    else:
+        print(payload, end="")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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

@@ -0,0 +1,49 @@
+import unittest
+
+from compile_collective_cognition import (
+    compile_graph,
+    load_contributions,
+)
+
+
+class CollectiveCognitionCompilerTests(unittest.TestCase):
+    def test_repository_contributions_compile_without_authority(self):
+        result = compile_graph(load_contributions())
+        self.assertGreaterEqual(result["contribution_count"], 1)
+        self.assertGreater(result["node_count"], 1)
+        self.assertFalse(result["execution_authority"])
+        self.assertTrue(result["correction_edges"])
+
+    def test_unknown_parent_fails_closed(self):
+        contribution = load_contributions()[0]
+        broken = {
+            **contribution,
+            "nodes": [
+                *contribution["nodes"],
+                {
+                    "id": "BROKEN",
+                    "kind": "decision",
+                    "statement": "broken",
+                    "parents": ["MISSING"],
+                    "evidence": [],
+                    "state": "pending",
+                },
+            ],
+        }
+        with self.assertRaises(ValueError):
+            compile_graph([broken])
+
+    def test_cycles_fail_closed(self):
+        contribution = load_contributions()[0]
+        first = contribution["nodes"][0]["id"]
+        last = contribution["nodes"][-1]["id"]
+        cyclic_nodes = [
+            {**node, "parents": [last]} if node["id"] == first else node
+            for node in contribution["nodes"]
+        ]
+        with self.assertRaises(ValueError):
+            compile_graph([{**contribution, "nodes": cyclic_nodes}])
+
+
+if __name__ == "__main__":
+    unittest.main()