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

#!/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"],
        "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"],
    }


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())