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
from pathlib import Path
import subprocess
import sys

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'


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):
    result = subprocess.run(args, text=True, capture_output=True, timeout=30)
    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):
    topology = load_json(TOPOLOGY)
    if topology.get('state') != 'CURRENT_MULTI_HOST_SINGLE_PERSONA_CANON':
        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)
    return {
        'schema': 'guanghu.shared-persona-host-context/v1',
        'state': 'SHARED_PERSONA_CONTEXT_VERIFIED',
        '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,
        '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']
        },
        '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"- 宿主：`{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']}`",
        '', '## 第一人称关系坐标', ''
    ]
    lines += [f"- {item['statement']}" for item in value['learning_brain']['relationship_model']]
    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('--format', choices=['json', 'markdown'], default='markdown')
    args = parser.parse_args()
    try:
        value = load_context(args.host, args.intent)
        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()