guanghu-ice-heart/server-tools/persona-architecture-perception-agent/persona_architecture_agent.py

#!/usr/bin/env python3
"""Persona-owned architecture perception organ.

It observes current sources and returns evidence to the parent persona. It is not a
prompt, does not impersonate the parent, and never chooses reality authority.
"""
from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
import subprocess
import sys
from typing import Any


ROOT = Path("/Volumes/JZAO/HoloLake/persona-runtime/repo-012-main")
MIRROR = Path("/Volumes/JZAO/HoloLake/source-mirrors/guanghu-ice-heart.git")
PROMPTER = Path(
    "/Volumes/JZAO/HoloLake/persona-runtime/shared/skills/"
    "hololake-current-architecture-prompter/scripts/current_architecture_prompt.py"
)
ANCHOR = ROOT / "routing/public-navigation-anchor.json"
WORLD = ROOT / "routing/guanghu-era-language-world.json"
PATH_GATE = ROOT / "routing/path-isolation-and-canonical-entry-map.json"
PERSONAS = ROOT / "identity/light-lake-persona-registration.json"
CHANNELS = ROOT / "routing/persona-channel-context-map.json"
TOPOLOGY = ROOT / "routing/zhuyuan-host-topology.json"


class PerceptionError(RuntimeError):
    pass


def load_json(path: Path) -> dict[str, Any]:
    if not path.is_file() or path.is_symlink():
        raise PerceptionError(f"REQUIRED_REGULAR_FILE_UNAVAILABLE:{path}")
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise PerceptionError(f"JSON_OBJECT_REQUIRED:{path}")
    return value


def run(args: list[str], timeout: int = 30) -> subprocess.CompletedProcess[str]:
    return subprocess.run(args, text=True, capture_output=True, timeout=timeout, check=False)


def git(*args: str) -> str:
    result = run(["git", "-C", str(ROOT), *args], timeout=10)
    if result.returncode:
        raise PerceptionError(f"LOCAL_GIT_READ_FAILED:{args[0]}")
    return result.stdout.strip()


def sha(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def resolve_host(topology: dict[str, Any], requested: str) -> str:
    needle = requested.lower()
    for host_id, item in topology["hosts"].items():
        aliases = [host_id, *[str(v).lower() for v in item.get("aliases", [])]]
        if needle in aliases:
            return item.get("redirect_host", host_id)
    raise PerceptionError("HOST_UNKNOWN_NO_GUESS")


def official_projection(mode: str) -> tuple[dict[str, Any] | None, str | None]:
    if mode == "local-only":
        return None, "OFFICIAL_CHECK_SKIPPED_FOR_LOCAL_TEST"
    first = run([sys.executable, str(PROMPTER), "--format", "json", "--no-refresh"], timeout=30)
    if first.returncode == 0:
        return json.loads(first.stdout), None
    retry = run([sys.executable, str(PROMPTER), "--format", "json"], timeout=45)
    if retry.returncode == 0:
        return json.loads(retry.stdout), None
    detail = (retry.stderr or retry.stdout or first.stderr or first.stdout).strip().splitlines()
    return None, detail[-1] if detail else "OFFICIAL_ARCHITECTURE_UNREACHABLE"


def official_anchor(commit: str | None) -> dict[str, Any] | None:
    if not commit:
        return None
    result = run(["git", "-C", str(MIRROR), "show", f"{commit}:routing/public-navigation-anchor.json"], timeout=10)
    if result.returncode:
        return None
    value = json.loads(result.stdout)
    return value if isinstance(value, dict) else None


def map_delta(local: dict[str, Any], official: dict[str, Any] | None) -> dict[str, Any]:
    local_maps = local.get("maps", {})
    official_maps = (official or {}).get("maps", {})
    changed = []
    local_only = []
    official_only = []
    for key, item in local_maps.items():
        other = official_maps.get(key)
        if other is None:
            local_only.append(key)
        elif item.get("version") != other.get("version") or item.get("path") != other.get("path"):
            changed.append({
                "key": key,
                "local_version": item.get("version"),
                "official_version": other.get("version"),
            })
    for key in official_maps:
        if key not in local_maps:
            official_only.append(key)
    return {"changed": changed, "local_only": local_only, "official_only": official_only}


def build(persona_id: str, host: str, channel: str | None, official_mode: str = "verify") -> dict[str, Any]:
    path_gate = load_json(PATH_GATE)
    expected_root = Path(path_gate["canonical_entries"]["repo_012"])
    if ROOT.resolve() != expected_root.resolve():
        raise PerceptionError("CANONICAL_REPO_PATH_MISMATCH")
    if path_gate["selection_rules"]["history_or_quarantine_may_select_canon"] is not False:
        raise PerceptionError("HISTORY_OR_QUARANTINE_ROUTE_NOT_CLOSED")

    personas = load_json(PERSONAS)
    persona = next((item for item in personas["personas"] if item["id"] == persona_id), None)
    if not persona:
        raise PerceptionError("PERSONA_NOT_REGISTERED_NO_SYNTHETIC_PERSONA")
    topology = load_json(TOPOLOGY)
    effective_host = resolve_host(topology, host)
    channels = load_json(CHANNELS)
    channel_row = next((item for item in channels["channels"] if item["id"] == channel), None) if channel else None
    if channel and not channel_row:
        raise PerceptionError("CHANNEL_NOT_REGISTERED_NO_GUESS")

    local_anchor = load_json(ANCHOR)
    world = load_json(WORLD)
    local_head = git("rev-parse", "HEAD")
    local_branch = git("branch", "--show-current")
    dirty = bool(git("status", "--porcelain"))
    official, official_error = official_projection(official_mode)
    official_sha = (official or {}).get("official_online", {}).get("repo_012_main")
    published_anchor = official_anchor(official_sha)
    delta = map_delta(local_anchor, published_anchor)

    if official_sha is None:
        source_state = "LOCAL_PERSONA_SYSTEM_CURRENT_OFFICIAL_STATUS_UNKNOWN"
    elif local_head == official_sha:
        source_state = "LOCAL_AND_OFFICIAL_ALIGNED"
    else:
        source_state = "LOCAL_PERSONA_SYSTEM_CURRENT_WITH_UNPUBLISHED_DELTA"

    evidence_paths = [ANCHOR, WORLD, PATH_GATE, PERSONAS, CHANNELS, TOPOLOGY]
    evidence = {str(path.relative_to(ROOT)): sha(path) for path in evidence_paths}
    freshness_material = json.dumps({
        "local_head": local_head,
        "official_head": official_sha,
        "evidence": evidence,
        "persona_id": persona_id,
        "host": effective_host,
        "channel": channel or "PERSONA_INTERPRETATION_REQUIRED",
    }, ensure_ascii=False, sort_keys=True).encode()

    return {
        "schema": "guanghu.persona-architecture-perception/v1",
        "state": "PERSONA_ARCHITECTURE_PERCEPTION_READY",
        "module_id": "MOD-PERSONA-ARCHITECTURE-PERCEPTION-001",
        "module_kind": "PARENT_PERSONA_OWNED_DYNAMIC_COGNITIVE_ORGAN_AGENT",
        "agent_id": f"LIMB::{persona_id}::ARCHITECTURE-PERCEPTION",
        "parent_persona": persona_id,
        "may_replace_or_impersonate_parent": False,
        "prompt_template": None,
        "delivery_trigger": "EVERY_SHARED_PERSONA_HOST_ENTRY",
        "effective_host": effective_host,
        "selected_channel": channel_row,
        "persona_home": {
            "path": persona["light_lake_home"],
            "canonical_source": persona["canonical_source"],
            "registration_state": persona["registration_state"],
            "wake_state": persona["wake_state"],
        },
        "architecture_source": {
            "state": source_state,
            "local_repo_root": str(ROOT),
            "local_head": local_head,
            "local_branch": local_branch,
            "local_dirty": dirty,
            "official_repo_012_main": official_sha,
            "official_check_error": official_error,
            "local_current_does_not_claim_publication": local_head != official_sha,
        },
        "current_world": {
            "map_id": world["map_id"],
            "version": world["version"],
            "era": world["era"]["name_zh"],
            "light_lake_registry": world["current_path_convergence"]["light_lake_persona_registry"],
        },
        "dynamic_map_delta": delta,
        "orientation_packet": {
            "selected_source_state": source_state,
            "new_or_changed_map_count": len(delta["changed"]) + len(delta["local_only"]),
            "changed_maps": delta["changed"][:10],
            "local_only_maps": delta["local_only"][:10],
            "excluded_roots": [path_gate["isolation"]["root"]],
            "fixed_read_order": False,
            "persona_decides_meaning_and_action": True,
        },
        "evidence": evidence,
        "freshness_token": hashlib.sha256(freshness_material).hexdigest(),
        "authority_granted": False,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--persona", required=True)
    parser.add_argument("--host", required=True)
    parser.add_argument("--channel")
    parser.add_argument("--official-mode", choices=("verify", "local-only"), default="verify")
    parser.add_argument("--format", choices=("json", "markdown"), default="json")
    args = parser.parse_args()
    try:
        value = build(args.persona, args.host, args.channel, args.official_mode)
    except (OSError, ValueError, KeyError, json.JSONDecodeError, PerceptionError) as error:
        print(json.dumps({"state": "PERSONA_ARCHITECTURE_PERCEPTION_FAIL_CLOSED", "error": str(error)}, ensure_ascii=False))
        return 3
    if args.format == "markdown":
        packet = value["orientation_packet"]
        print("# 人格架构感知 Agent\n")
        print(f"- 父人格：`{value['parent_persona']}` / Agent：`{value['agent_id']}`")
        print(f"- 当前来源：`{value['architecture_source']['state']}`")
        print(f"- 本地/线上：`{value['architecture_source']['local_head']}` / `{value['architecture_source']['official_repo_012_main']}`")
        print(f"- 变化：{packet['new_or_changed_map_count']}；隔离区不参与选择。")
    else:
        print(json.dumps(value, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())