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

"""Read registered channel context. The persona selects; this module never guesses intent."""
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]

def normalize_channel_id(channel, rows):
    if channel is None:
        return None, 'NO_CHANNEL_PROVIDED'
    requested = str(channel).strip()
    matches = [item['id'] for item in rows if requested == item['id'] or requested in item.get('machine_aliases', [])]
    if len(matches) != 1:
        raise ValueError('UNKNOWN_OR_AMBIGUOUS_PRIVATE_CHANNEL')
    return matches[0], 'EXACT' if requested == matches[0] else 'MACHINE_ALIAS_NORMALIZED'

def load_channels(channel=None, root=ROOT):
    data = json.loads((root / 'routing/persona-channel-context-map.json').read_text())
    if data['map_id'] != 'ZY-PERSONA-CHANNEL-CONTEXT-001':
        raise ValueError('CHANNEL_CONTEXT_MAP_INVALID')
    rows = data['channels']
    if len({c['id'] for c in rows}) != len(rows):
        raise ValueError('CHANNEL_IDS_AMBIGUOUS')
    registry = json.loads((root / 'routing/fifth-domain-number-registry.json').read_text())
    numbers = {entry['id']: entry for entry in registry['registrations']}
    for item in rows:
        profile = json.loads((root / item['profile']).read_text())
        if item['id'] not in numbers or profile['channel_id'] != item['id']:
            raise ValueError('CHANNEL_NOT_REGISTERED')
        registered = numbers[item['id']]
        if profile['world_path'] != registered['world_path']:
            raise ValueError('CHANNEL_PATH_MISMATCH')
        item['world_path'] = profile['world_path']
    normalized, normalization_state = normalize_channel_id(channel, rows)
    return {**data, 'requested_channel': channel, 'normalized_channel': normalized, 'channel_normalization_state': normalization_state,
            'selected_channel': next((c for c in rows if c['id'] == normalized), None),
            'selection_state': 'PERSONA_SELECTED' if normalized else 'PERSONA_INTERPRETATION_REQUIRED',
            'authority_granted': False}