237 lines
9.9 KiB
Python
237 lines
9.9 KiB
Python
#!/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())
|