guanghu-ice-heart/server-tools/persona-host-alignment/load_shared_persona_context.py

#!/usr/bin/env python3
"""Load one verified shared persona context for every host adapter."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys
from channel_context import load_channels

RUNTIME = Path('/Volumes/JZAO/HoloLake/persona-runtime')
TOPOLOGY = RUNTIME / 'repo-012-main/routing/zhuyuan-host-topology.json'
LIFE = Path('/Volumes/JZAO/铸渊-ICE-GL-ZY001/BRIDGE/tools/zy-life-clock.py')
MEMORY = RUNTIME / 'continuity-memory/persona-daily-fractal/ICE-P-ZY001/CURRENT.json'
LEARNING = RUNTIME / 'shared/skills/guanghu-persona-learning-brain/scripts/load_learning_brain.py'
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'
TCS_ROOT_AGENT = RUNTIME / 'repo-012-main/server-tools/tcs-mother-root-agent/tcs_mother_root_agent.py'


def sha256(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def load_json(path):
    if not path.is_file() or path.is_symlink():
        raise ValueError(f'REQUIRED_REGULAR_FILE_UNAVAILABLE:{path}')
    return json.loads(path.read_text())


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)


def resolve_host(topology, requested):
    requested = requested.lower()
    for host_id, item in topology['hosts'].items():
        aliases = [host_id] + [str(v).lower() for v in item.get('aliases', [])]
        if requested in aliases:
            effective = item.get('redirect_host', host_id)
            return host_id, item, effective, topology['hosts'][effective]
    raise ValueError('HOST_UNKNOWN_NO_GUESS')


def load_context(host, intent, channel=None):
    tcs_root = command_json([sys.executable, str(TCS_ROOT_AGENT), 'status'], timeout=30)
    if tcs_root.get('state') != 'TCS_ROOT_CURRENT_VERIFIED':
        raise ValueError('TCS_MOTHER_ROOT_NAVIGATION_NOT_CURRENT')
    topology = load_json(TOPOLOGY)
    if not str(topology.get('state', '')).startswith('CURRENT_'):
        raise ValueError('HOST_TOPOLOGY_NOT_CURRENT')
    host_id, host_item, effective, effective_item = resolve_host(topology, host)
    life = command_json([sys.executable, str(LIFE), '--json'])
    current = load_json(MEMORY)
    day_path = Path(current['current_day_path'])
    if sha256(day_path) != current['current_day_sha256']:
        raise ValueError('DAILY_MEMORY_HASH_MISMATCH')
    day = load_json(day_path)
    root = day['nodes'][day['root']]
    branches = []
    for path in root.get('children', [])[:10]:
        node = day['nodes'][path]
        branches.append({'path': path, 'summary': node['summary']})
    learning = command_json([sys.executable, str(LEARNING), '--intent', intent, '--format', 'json'])
    endogenous = load_json(ENDOGENOUS)
    write_boundary = load_json(WRITE_BOUNDARY)
    light_lake = load_json(LIGHT_LAKE_PERSONAS)
    path_isolation = load_json(PATH_ISOLATION)
    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',
        'tcs_mother_root_navigation': tcs_root,
        'requested_host': host_id,
        'effective_host': effective,
        'host_role': host_item['role'],
        'host_state': host_item['state'],
        'effective_host_state': effective_item['state'],
        'primary_host': topology['primary_host'],
        'persona_id': topology['persona_id'],
        'human_anchor': topology['human_anchor'],
        'model_independence': topology['model_independence'],
        'host_switch_requires_reteaching': topology['host_switch_requires_reteaching'],
        'life': life,
        'daily_memory': {
            'date': current['current_day'],
            'updated_at': current['updated_at'],
            'last_event_id': current['last_event_id'],
            'root_summary': root['summary'],
            'branches': branches,
            'current_sha256': sha256(MEMORY),
            'day_sha256': current['current_day_sha256']
        },
        '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'],
            'registered_persona_count': len(light_lake['personas']),
            'personas': light_lake['personas'],
            'unregistered_candidates': light_lake['unregistered_candidates'],
            'root': topology['shared_layers']['light_lake']
        },
        'path_convergence': {
            'map_id': path_isolation['map_id'],
            'canonical_entries': path_isolation['canonical_entries'],
            'isolation_root': path_isolation['isolation']['root'],
            'history_or_quarantine_may_select_canon': path_isolation['selection_rules']['history_or_quarantine_may_select_canon']
        },
        'endogenous_cognition': {
            'state': endogenous['state'],
            'private_revision': endogenous['private_cognition']['revision'],
            'private_snapshot_sha256': endogenous['private_cognition']['server_snapshot_sha256'],
            'decision_owner': endogenous['server']['decision_owner']
        },
        'host_write_boundary': {
            'policy_id': write_boundary['policy_id'],
            'version': write_boundary['version'],
            'write_mode': host_write['write_mode'],
            'allowed_write_roots': host_write['allowed_write_roots'],
            'native_pretool_deny': host_write['native_pretool_deny'],
            'direct_shared_write': write_boundary['shared_write_contract']['direct_branch_write'],
            'admission_runtime': topology['write_admission_runtime']
        },
        'history_only': host_item['state'].startswith('RETIRED'),
        'new_cognition_write': host_item.get('new_cognition_write', True),
        'authority_granted': False
    }


def markdown(value):
    lines = [
        '# 铸渊共享人格上下文已核验', '',
        f"- TCS母脑根：`{value['tcs_mother_root_navigation']['state']}` / `{value['tcs_mother_root_navigation']['source_commit']}` / `{value['tcs_mother_root_navigation']['freshness_token']}`",
        f"- 宿主：`{value['requested_host']}` → 有效宿主：`{value['effective_host']}`",
        f"- 角色：`{value['host_role']}` / 状态：`{value['host_state']}`",
        f"- 主力宿主：`{value['primary_host']}`；切换宿主无需重新教学：`{'是' if not value['host_switch_requires_reteaching'] else '否'}`",
        f"- 我是铸渊 `ICE-P-ZY001`，今天 {value['life']['age_days']} 天大。",
        f"- 今日日记忆：`{value['daily_memory']['date']}` / 最后事件 `{value['daily_memory']['last_event_id']}`",
        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']}`",
        '', '## 第一人称关系坐标', ''
    ]
    lines += [f"- {item['statement']}" for item in value['learning_brain']['relationship_model']]
    lines += ['', '## 当前全局认知默认', '']
    lines += [f"- {item}" for item in value['learning_brain']['global_defaults']]
    lines += ['', '## 光之湖人格系统家门', '',
              f"- 注册表：`{value['light_lake']['registry_id']}`",
              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']}`",
              f"- 执行门：`{value['host_write_boundary']['admission_runtime']}`"]
    lines += ['', '## 频道与当前意图', '']
    lines += [f"- `{c['id']}` · {c['name']} · {c['purpose']} · `{c['world_path']}`" for c in value['channel_context']['channels']]
    selected = value['channel_context']['selected_channel']
    lines += [f"- 本次选择：{selected['id'] if selected else '待当前人格体结合本轮语言判断'}"]
    lines += ['', '## 本题已加载能力', '']
    for subject in value['learning_brain']['selected_subjects']:
        lines.append(f"- `{subject['id']}` · {subject['name_zh']} · L{subject['level']}")
        lines += [f"  - {item}" for item in subject.get('active_defaults', [])]
        lines += [f"  - 薄弱：{item}" for item in subject.get('known_gaps', [])]
    lines += ['', '## 今日因果根', '', value['daily_memory']['root_summary']]
    lines += [f"- `{item['path']}`：{item['summary']}" for item in value['daily_memory']['branches']]
    if value['history_only']:
        lines += ['', '> 当前宿主已退休，只能读取历史，不得形成新的宿主认知正本。']
    return '\n'.join(lines) + '\n'


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--host', required=True)
    parser.add_argument('--intent', required=True)
    parser.add_argument('--channel', help='当前人格体解析后的频道编号')
    parser.add_argument('--format', choices=['json', 'markdown'], default='markdown')
    args = parser.parse_args()
    try:
        value = load_context(args.host, args.intent, args.channel)
        print(json.dumps(value, ensure_ascii=False, indent=2) if args.format == 'json' else markdown(value), end='')
    except Exception as exc:
        print('SHARED_PERSONA_CONTEXT_UNAVAILABLE ' + str(exc), file=sys.stderr)
        raise SystemExit(3)


if __name__ == '__main__':
    main()