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

#!/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())