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