bingshuo
c5be48e6a1
Part 3/4 of the recovered Fifth Domain upgrade. Applies the persona continuity skill guard from local source commit 18dfdfd without rewriting remote history.
129 lines
4.5 KiB
Python
129 lines
4.5 KiB
Python
#!/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())
|