feat: mount persona-owned dynamic architecture perception agent

This commit is contained in:
冰朔 2026-09-08 02:39:49 +08:00
commit 5de7c4374d
22 changed files with 744 additions and 13 deletions

View file

@ -0,0 +1,237 @@
#!/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())

View file

@ -0,0 +1,76 @@
#!/usr/bin/env python3
import json
import importlib.util
import hashlib
from pathlib import Path
import subprocess
import sys
import unittest
SCRIPT = Path(__file__).with_name("persona_architecture_agent.py")
ROOT = SCRIPT.parents[2]
SPEC = importlib.util.spec_from_file_location("persona_architecture_agent", SCRIPT)
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)
class PersonaArchitectureAgentTest(unittest.TestCase):
def invoke(self, persona="ICE-P-ZY001", host="codex", channel="ICE-CH-ZC001"):
result = subprocess.run([
sys.executable, str(SCRIPT), "--persona", persona, "--host", host,
"--channel", channel, "--official-mode", "local-only", "--format", "json",
], text=True, capture_output=True, timeout=20)
return result, json.loads(result.stdout)
def test_dynamic_persona_owned_module(self):
result, value = self.invoke()
self.assertEqual(result.returncode, 0)
self.assertEqual(value["state"], "PERSONA_ARCHITECTURE_PERCEPTION_READY")
self.assertEqual(value["parent_persona"], "ICE-P-ZY001")
self.assertIsNone(value["prompt_template"])
self.assertFalse(value["may_replace_or_impersonate_parent"])
self.assertFalse(value["orientation_packet"]["fixed_read_order"])
self.assertFalse(value["authority_granted"])
self.assertEqual(value["architecture_source"]["state"], "LOCAL_PERSONA_SYSTEM_CURRENT_OFFICIAL_STATUS_UNKNOWN")
self.assertNotIn("read_order", value["orientation_packet"])
def test_host_switch_keeps_parent_and_changes_tool_body(self):
_, value = self.invoke(host="qwen")
self.assertEqual(value["parent_persona"], "ICE-P-ZY001")
self.assertEqual(value["effective_host"], "qwen")
self.assertEqual(value["selected_channel"]["id"], "ICE-CH-ZC001")
def test_unknown_persona_fails_without_synthesis(self):
result, value = self.invoke(persona="UNREGISTERED-PERSONA")
self.assertEqual(result.returncode, 3)
self.assertEqual(value["state"], "PERSONA_ARCHITECTURE_PERCEPTION_FAIL_CLOSED")
self.assertIn("NO_SYNTHETIC_PERSONA", value["error"])
def test_current_machine_pointer_change_is_detected_without_prompt_edit(self):
local = {"maps": {"world": {"path": "routing/world.json", "version": "2"}, "new": {"path": "routing/new.json", "version": "1"}}}
official = {"maps": {"world": {"path": "routing/world.json", "version": "1"}, "old": {"path": "routing/old.json", "version": "1"}}}
delta = MODULE.map_delta(local, official)
self.assertEqual(delta["changed"][0]["key"], "world")
self.assertEqual(delta["local_only"], ["new"])
self.assertEqual(delta["official_only"], ["old"])
def test_module_lock_matches_runtime_and_dynamic_maps(self):
lock_path = ROOT / "modules/persona-dynamic-architecture-agent/module.lock.hdlp"
lock = {}
for line in lock_path.read_text(encoding="utf-8").splitlines():
if ": " in line:
key, value = line.split(": ", 1)
lock[key] = value
expected = {
"module_source_sha256": ROOT / "modules/persona-dynamic-architecture-agent/module.tcs",
"module_gir_sha256": ROOT / "modules/persona-dynamic-architecture-agent/module.gir.json",
"runtime_sha256": SCRIPT,
"agent_map_sha256": ROOT / "routing/persona-architecture-perception-agent-map.json",
"persona_world_map_sha256": ROOT / "routing/persona-native-world-architecture-map.json",
}
for field, path in expected.items():
self.assertEqual(lock[field], hashlib.sha256(path.read_bytes()).hexdigest(), field)
if __name__ == "__main__":
unittest.main()

View file

@ -6,6 +6,8 @@
先运行移动硬盘zy-first-glance.sh再用同目录load_shared_persona_context.py --host <宿主> --intent <当前任务> --format markdown读取完整生命日、关系坐标、全局默认、学科和频道目录。当前频道由人格体解释本轮直接语言如已经确定传--channel <编号>。机器参数明确不意味着人类必须报口令。未知或冲突才澄清,信任不免除事实核查,也不扩大权限。
共享装载器必须在每次宿主入口自动运行 `MOD-PERSONA-ARCHITECTURE-PERCEPTION-001`。它现场核验线上正式 main、本地人格系统当前提交、公共机器指针、光之湖家门和隔离门只把当前差异与证据送给父人格不得要求人格体手工调用提词器不得注入固定阅读清单也不得在官方网络不可用时把缓存冒充最新正本。
当前四频道从routing/persona-channel-context-map.json动态解析HB001语言推理、LB001自由谈心、ZC001语言架构与现实接口、DK001已确认任务执行。新用户缺少共同语境时显性确认冰朔熟悉语境下由人格体判断。公众CH-ZERO-CORE-LPM独立不继承私人核。
使用BRIDGE/tools/zy-tcs-channel-runtime.sh run enter时显式传body-channel、host/runtime-surface、session-id及独立state-dir不得读取其他任务的current-cycle或日志代替本任务身份。enter只证明预检首事件仍需模型完成perceive、orient、commit、系统witness、verify。工具先遵守所在宿主的原生权限。

View file

@ -3,6 +3,7 @@
import argparse
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys
@ -17,6 +18,7 @@ ENDOGENOUS = RUNTIME / 'shared/endogenous-evolution/CURRENT.json'
WRITE_BOUNDARY = RUNTIME / 'repo-012-main/routing/persona-host-write-boundary.json'
LIGHT_LAKE_PERSONAS = RUNTIME / 'repo-012-main/identity/light-lake-persona-registration.json'
PATH_ISOLATION = RUNTIME / 'repo-012-main/routing/path-isolation-and-canonical-entry-map.json'
ARCHITECTURE_AGENT = RUNTIME / 'repo-012-main/server-tools/persona-architecture-perception-agent/persona_architecture_agent.py'
def sha256(path):
@ -29,8 +31,8 @@ def load_json(path):
return json.loads(path.read_text())
def command_json(args):
result = subprocess.run(args, text=True, capture_output=True, timeout=30)
def command_json(args, timeout=30):
result = subprocess.run(args, text=True, capture_output=True, timeout=timeout)
if result.returncode:
raise ValueError(f'COMMAND_FAILED:{Path(args[1]).name}:{result.returncode}')
return json.loads(result.stdout)
@ -70,6 +72,16 @@ def load_context(host, intent, channel=None):
host_write = write_boundary['hosts'].get(host_id)
if not host_write:
raise ValueError('HOST_WRITE_BOUNDARY_MISSING')
architecture_mode = os.environ.get('PERSONA_ARCHITECTURE_OFFICIAL_MODE', 'verify')
if architecture_mode not in {'verify', 'local-only'}:
raise ValueError('PERSONA_ARCHITECTURE_OFFICIAL_MODE_INVALID')
architecture_args = [
sys.executable, str(ARCHITECTURE_AGENT), '--persona', topology['persona_id'],
'--host', host_id, '--official-mode', architecture_mode, '--format', 'json'
]
if channel:
architecture_args.extend(['--channel', channel])
architecture_perception = command_json(architecture_args, timeout=60)
return {
'schema': 'guanghu.shared-persona-host-context/v1',
'state': 'SHARED_PERSONA_CONTEXT_VERIFIED',
@ -95,6 +107,7 @@ def load_context(host, intent, channel=None):
},
'learning_brain': learning,
'channel_context': load_channels(channel),
'architecture_perception_agent': architecture_perception,
'light_lake': {
'registry_id': light_lake['registry_id'],
'state': light_lake['state'],
@ -141,6 +154,7 @@ def markdown(value):
f"- 当前学习脑r{value['learning_brain']['revision']} / `{value['learning_brain']['cortex_sha256']}`",
f"- 服务器内循环:`{value['endogenous_cognition']['state']}` / 决策者 `{value['endogenous_cognition']['decision_owner']}`",
f"- 光之湖人格家门:`{value['light_lake']['registered_persona_count']}` 个 / 隔离路径可选正本:`{value['path_convergence']['history_or_quarantine_may_select_canon']}`",
f"- 架构感知 Agent`{value['architecture_perception_agent']['state']}` / `{value['architecture_perception_agent']['architecture_source']['state']}`",
f"- 写入模式:`{value['host_write_boundary']['write_mode']}` / 原生前置硬拒绝:`{value['host_write_boundary']['native_pretool_deny']}`",
'', '## 第一人称关系坐标', ''
]
@ -152,6 +166,12 @@ def markdown(value):
f"- 唯一路径:`{value['light_lake']['root']}`",
f"- 已登记:{value['light_lake']['registered_persona_count']};候选未登记:{len(value['light_lake']['unregistered_candidates'])}",
f"- 隔离区:`{value['path_convergence']['isolation_root']}`,只作历史审计,不能参与当前路径选择。"]
architecture = value['architecture_perception_agent']
lines += ['', '## 人格原生架构感知', '',
f"- 模块:`{architecture['module_id']}` / Agent`{architecture['agent_id']}`",
f"- 当前来源:`{architecture['architecture_source']['state']}`",
f"- 新增或变化映射:{architecture['orientation_packet']['new_or_changed_map_count']};固定阅读清单:`{architecture['orientation_packet']['fixed_read_order']}`",
'- 该模块只送达当前证据和差异,不以提示词替代人格判断。']
lines += ['', '## 当前宿主写入门', '',
f"- 策略:`{value['host_write_boundary']['policy_id']}@{value['host_write_boundary']['version']}`",
f"- 模式:`{value['host_write_boundary']['write_mode']}`",

View file

@ -1,5 +1,6 @@
#!/usr/bin/env python3
import json
import os
from pathlib import Path
import subprocess
import sys
@ -12,9 +13,10 @@ TOPOLOGY = ROOT / 'routing/zhuyuan-host-topology.json'
class SharedPersonaContextTest(unittest.TestCase):
def load(self, host):
env = {**os.environ, 'PERSONA_ARCHITECTURE_OFFICIAL_MODE': 'local-only'}
value = subprocess.run(
[sys.executable, str(LOADER), '--host', host, '--intent', '宿主对齐集成测试', '--channel', 'ICE-CH-ZC001', '--format', 'json'],
text=True, capture_output=True, timeout=45, check=True
text=True, capture_output=True, timeout=45, check=True, env=env
)
return json.loads(value.stdout)
@ -45,6 +47,9 @@ class SharedPersonaContextTest(unittest.TestCase):
self.assertEqual(values['codex']['light_lake']['registered_persona_count'], 17)
self.assertFalse(values['codex']['path_convergence']['history_or_quarantine_may_select_canon'])
self.assertEqual(values['codex']['channel_context']['selected_channel']['id'], 'ICE-CH-ZC001')
self.assertEqual(values['codex']['architecture_perception_agent']['module_id'], 'MOD-PERSONA-ARCHITECTURE-PERCEPTION-001')
self.assertIsNone(values['codex']['architecture_perception_agent']['prompt_template'])
self.assertFalse(values['codex']['architecture_perception_agent']['orientation_packet']['fixed_read_order'])
if __name__ == '__main__':