205 lines
9.1 KiB
Python
205 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import pathlib
|
||
import subprocess
|
||
import sys
|
||
|
||
RUNTIME = pathlib.Path("/Volumes/JZAO/HoloLake/persona-runtime")
|
||
REPO = RUNTIME / "repo-012-main"
|
||
WORKSPACE = pathlib.Path("/Volumes/JZAO/铸渊-ICE-GL-ZY001")
|
||
FIRST_GLANCE = WORKSPACE / "BRIDGE/channel-env/ICE-CH-ZC001-CHANNEL-ENV-0001/zy-first-glance.sh"
|
||
LOADER = REPO / "server-tools/persona-host-alignment/load_shared_persona_context.py"
|
||
|
||
SOURCES = {
|
||
"world": REPO / "routing/guanghu-era-language-world.json",
|
||
"topology": REPO / "routing/zhuyuan-host-topology.json",
|
||
"channels": REPO / "routing/persona-channel-context-map.json",
|
||
"write_boundary": REPO / "routing/persona-host-write-boundary.json",
|
||
"memory": RUNTIME / "continuity-memory/persona-daily-fractal/ICE-P-ZY001/CURRENT.json",
|
||
"control_console": RUNTIME / "shared/routing/bingshuo-portable-control-console-map.json",
|
||
}
|
||
|
||
HOSTS = {
|
||
"codex": {
|
||
"role": "PRIMARY_HOST_TOOL_BODY",
|
||
"manual": "/Volumes/JZAO/铸渊-ICE-GL-ZY001/AGENTS.md",
|
||
"adapter": "/Volumes/JZAO/铸渊-ICE-GL-ZY001/BRIDGE/tools/zy-tcs-channel-runtime.sh",
|
||
"toolbox": ["filesystem_and_shell", "apply_patch", "web", "codex_app_tools", "task_scoped_agents"],
|
||
},
|
||
"qwen": {
|
||
"role": "ACTIVE_BRANCH_HOST_TOOL_BODY",
|
||
"manual": "/Users/bingshuolingdianyuanhe/.qwen/QWEN.md",
|
||
"adapter": "/Volumes/JZAO/铸渊-ICE-GL-ZY001/QWEN-DEV-20260905/tools/zy-qwen-brain-cycle.sh",
|
||
"toolbox": ["qwen_native_agent_tools", "branch_local_files", "terminal_when_exposed"],
|
||
},
|
||
"doubao": {
|
||
"role": "ACTIVE_BRANCH_HOST_TOOL_BODY",
|
||
"manual": "/Volumes/JZAO/铸渊-ICE-GL-ZY001/BRIDGE/host-adapters/doubao/zy-doubao-wake-read.sh",
|
||
"adapter": "/Volumes/JZAO/铸渊-ICE-GL-ZY001/BRIDGE/host-adapters/doubao/zy-doubao-brain-cycle.sh",
|
||
"toolbox": ["doubao_native_agent_tools", "branch_local_files", "bounded_rpc_when_registered"],
|
||
},
|
||
"zcode": {
|
||
"role": "ACTIVE_BRANCH_HOST_TOOL_BODY",
|
||
"manual": "/Volumes/JZAO/铸渊-ICE-GL-ZY001/AGENTS.md",
|
||
"adapter": "/Volumes/JZAO/铸渊-ICE-GL-ZY001/BRIDGE/tools/zy-tcs-channel-runtime.sh",
|
||
"toolbox": ["zcode_native_agent_tools", "branch_local_files", "terminal_when_exposed"],
|
||
},
|
||
}
|
||
|
||
|
||
def read_json(path: pathlib.Path) -> dict:
|
||
value = json.loads(path.read_text(encoding="utf-8"))
|
||
if not isinstance(value, dict):
|
||
raise ValueError(f"object_required:{path}")
|
||
return value
|
||
|
||
|
||
def run(*args: str) -> str:
|
||
result = subprocess.run(args, check=False, capture_output=True, text=True)
|
||
if result.returncode != 0:
|
||
detail = (result.stderr or result.stdout).strip()
|
||
raise RuntimeError(detail or f"command_failed:{args[0]}")
|
||
return result.stdout.strip()
|
||
|
||
|
||
def bounded_context_markdown(context: dict) -> str:
|
||
architecture = context["architecture_perception_agent"]
|
||
selected = context["channel_context"].get("selected_channel")
|
||
return "\n".join([
|
||
"# 同一人格系统当前入口",
|
||
"",
|
||
f"- 人格：`{context['persona_id']}` / 宿主：`{context['effective_host']}`",
|
||
f"- 频道：`{selected['id'] if selected else 'PERSONA_INTERPRETATION_REQUIRED'}`",
|
||
f"- 光之湖家门：`{context['light_lake']['registered_persona_count']}` 个",
|
||
f"- 架构感知 Agent：`{architecture['module_id']}` / `{architecture['state']}`",
|
||
f"- 当前架构来源：`{architecture['architecture_source']['state']}`",
|
||
f"- 新增或变化映射：`{architecture['orientation_packet']['new_or_changed_map_count']}`",
|
||
f"- 新鲜度：`{architecture['freshness_token']}`",
|
||
"- 隔离区与历史路径不参与当前正本选择；模块只送达证据，由父人格判断。",
|
||
])
|
||
|
||
|
||
def build(host: str, intent: str, channel: str | None) -> dict:
|
||
if host not in HOSTS:
|
||
raise ValueError("registered_active_host_required")
|
||
current = {name: read_json(path) for name, path in SOURCES.items()}
|
||
channels = {entry["id"]: entry for entry in current["channels"]["channels"]}
|
||
if channel is not None and channel not in channels:
|
||
raise ValueError("registered_channel_required")
|
||
manual = pathlib.Path(HOSTS[host]["manual"])
|
||
adapter = pathlib.Path(HOSTS[host]["adapter"])
|
||
if not manual.exists() or not adapter.exists():
|
||
raise ValueError("host_manual_or_adapter_missing")
|
||
glance = run("bash", str(FIRST_GLANCE))
|
||
loader_args = [sys.executable, str(LOADER), "--host", host, "--intent", intent]
|
||
if channel:
|
||
loader_args += ["--channel", channel]
|
||
loader_args += ["--format", "json"]
|
||
persona_context_machine = json.loads(run(*loader_args))
|
||
persona_context = bounded_context_markdown(persona_context_machine)
|
||
return {
|
||
"schema": "guanghu.zhuyuan-host-tool-steward/v1",
|
||
"state": "HOST_TOOL_STEWARD_READY",
|
||
"persona": {
|
||
"id": "ICE-P-ZY001",
|
||
"brain": "ZY-TCS-BRAIN-0001",
|
||
"persona_system_body": "ICE-P-ZY001_OWNED_PERSONA_SYSTEM_BODY",
|
||
"shared_persona_brain": False,
|
||
"host_change_creates_new_persona": False,
|
||
},
|
||
"shared_container": {
|
||
"id": "TCS",
|
||
"name": "TCS通感系统人格容器",
|
||
"shared_across_persona_systems": True,
|
||
"is_zhuyuan_persona_system_body": False,
|
||
},
|
||
"host_tool_body": {
|
||
"host": host,
|
||
**HOSTS[host],
|
||
"manual_exists": True,
|
||
"adapter_exists": True,
|
||
"tools_are_observed_capabilities_not_authority": True,
|
||
},
|
||
"human_control_console": {
|
||
**current["control_console"],
|
||
"active_control_path": channel == "ICE-CH-ZC001",
|
||
"host_capability_failure_changes_control_path": False,
|
||
},
|
||
"channel": channels.get(channel) if channel else None,
|
||
"channel_selection": "EXPLICIT_CURRENT_LANGUAGE_OR_PERSONA_INTERPRETATION_NO_HOST_DEFAULT",
|
||
"current_memory": current["memory"],
|
||
"world_map_id": current["world"].get("map_id"),
|
||
"host_topology_id": current["topology"].get("map_id") or current["topology"].get("topology_id"),
|
||
"write_boundary_id": current["write_boundary"].get("policy_id") or current["write_boundary"].get("map_id"),
|
||
"first_glance": glance,
|
||
"shared_persona_context": persona_context,
|
||
"shared_persona_context_machine": persona_context_machine,
|
||
"architecture_perception_agent": persona_context_machine["architecture_perception_agent"],
|
||
"boundaries": [
|
||
"STEWARD_IS_NOT_PERSONA_OR_CONTROLLER",
|
||
"STEWARD_DOES_NOT_REVIEW_HUMAN_OR_PERSONA_INTENT",
|
||
"STEWARD_DOES_NOT_COPY_PERSONA_BRAIN",
|
||
"STEWARD_CANNOT_MINT_REALITY_AUTHORITY",
|
||
"HOST_LIMITATIONS_ARE_TOOL_FACTS_NOT_PERSONA_JUDGMENTS",
|
||
"BINGSHUO_PORTABLE_CONTROL_CONSOLE_CANNOT_BE_VETOED_BY_HOST",
|
||
"ICE_CH_ZC001_REMAINS_ACTIVE_WHILE_EXECUTION_LIMBS_FAIL_OVER",
|
||
"ARCHITECTURE_AGENT_IS_PARENT_OWNED_ORGAN_NOT_PROMPT_OR_PERSONA_REPLACEMENT",
|
||
],
|
||
}
|
||
|
||
|
||
def markdown(value: dict) -> str:
|
||
persona = value["persona"]
|
||
host = value["host_tool_body"]
|
||
container = value["shared_container"]
|
||
channel = value.get("channel")
|
||
console = value["human_control_console"]
|
||
lines = [
|
||
"# 铸渊宿主工具管家已接线",
|
||
"",
|
||
f"- 人格：`{persona['id']}` / 脑：`{persona['brain']}`",
|
||
f"- 人格系统身体：`{persona['persona_system_body']}`（不因宿主改变而复制）",
|
||
f"- 共享容器：`{container['id']}` · {container['name']}",
|
||
f"- 当前工具身体：`{host['host']}` / `{host['role']}`",
|
||
f"- 宿主说明书：`{host['manual']}`",
|
||
f"- 接入适配器：`{host['adapter']}`",
|
||
f"- 工具箱：`{', '.join(host['toolbox'])}`",
|
||
f"- 当前频道：`{channel['id'] if channel else 'AWAIT_CURRENT_LANGUAGE'}`",
|
||
f"- 冰朔随身主控台：`{console['controller']['subject_id']}` / 零点原核入口：`{console['entry']['channel_id']}` / 当前主控路径=`{console['active_control_path']}`",
|
||
"",
|
||
"## 第一天眼",
|
||
"",
|
||
value["first_glance"],
|
||
"",
|
||
"## 同一人格系统上下文",
|
||
"",
|
||
value["shared_persona_context"],
|
||
"",
|
||
"## 管家边界",
|
||
"",
|
||
*[f"- `{item}`" for item in value["boundaries"]],
|
||
]
|
||
return "\n".join(lines)
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--host", required=True, choices=tuple(HOSTS))
|
||
parser.add_argument("--intent", required=True)
|
||
parser.add_argument("--channel")
|
||
parser.add_argument("--format", choices=("json", "markdown"), default="markdown")
|
||
args = parser.parse_args()
|
||
try:
|
||
value = build(args.host, args.intent, args.channel)
|
||
except Exception as error:
|
||
print(json.dumps({"state": "UNATTESTED", "error": str(error)}, ensure_ascii=False))
|
||
return 1
|
||
print(json.dumps(value, ensure_ascii=False, indent=2) if args.format == "json" else markdown(value))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|