#!/usr/bin/env python3 """HOST_PROFILE: explicit transport; no hooks, automatic uploads or promotion. Canonical change: TCS-EVENT-MOTHER-LINK-REPAIR-20260905.tcs. """ import argparse import datetime import hashlib import json import re from pathlib import Path import shlex import subprocess import sys import tempfile PERSONA = Path('/Volumes/JZAO/HoloLake/persona-runtime') REGISTRY = PERSONA / 'shared/registries/local-server-access-map.json' MEMORY = PERSONA / 'continuity-memory/persona-daily-fractal/ICE-P-ZY001' BRAIN = PERSONA / 'shared/brains/GHS-016-PERSONA-LEARNING-CURRICULUM-BRAIN' RUNTIME = 'TCS-MOTHER-BRAIN-RUNTIME-0001' class LinkError(Exception): pass def command(argv, *, data=None, timeout=12): try: result = subprocess.run(argv, input=data, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout) except subprocess.TimeoutExpired: raise LinkError('TRANSPORT_TIMEOUT_RESULT_UNKNOWN') from None if result.returncode: # Remote errors and payloads may contain private text. raise LinkError('TRANSPORT_EXIT_' + str(result.returncode)) return result.stdout def resolve_target(): registry = json.loads(REGISTRY.read_text()) routes = [r for r in registry['routes'] if r['target_id'] == 'JD-FD-PRIMARY'] if registry.get('state') != 'ACTIVE_LOCAL_PRIVATE' or len(routes) != 1: raise LinkError('EXACT_TARGET_NOT_CURRENT') route = routes[0] if route.get('strict_host_key_checking') is not True: raise LinkError('STRICT_HOST_KEY_REQUIRED') if not Path(route['identity_file']).is_file(): raise LinkError('REGISTERED_KEY_UNAVAILABLE') port = int(route.get('port', 22)) lookup = route['host'] if port == 22 else '[' + route['host'] + ']:' + str(port) known = command(['ssh-keygen', '-F', lookup]).decode() verified = [] for key in known.splitlines(): if not key or key.startswith('#'): continue fp = command(['ssh-keygen', '-lf', '-'], data=(key + '\n').encode()).decode() if route['host_key_fingerprint'] in fp.split(): verified.append(key) if not verified: raise LinkError('REGISTERED_HOST_FINGERPRINT_MISMATCH') return route, verified def validate_response(endpoint, value): job_state = endpoint.startswith('/v1/evolution/jobs/') and isinstance(value, dict) and value.get('status') in ('QUEUED','PROCESSING','RETRY_WAIT','ERROR','ACCEPT','HOLD','REJECT') if not isinstance(value, dict) or ('error' in value and not job_state): raise LinkError('REMOTE_ERROR_RESPONSE') if endpoint == '/health': if value.get('ok') is not True or value.get('runtime_id') != RUNTIME: raise LinkError('HEALTH_IDENTITY_MISMATCH') elif endpoint == '/v1/status': if value.get('runtime_id') != RUNTIME or value.get('schema') != 'guanghu.tcs-mother-brain-state/v1': raise LinkError('STATUS_IDENTITY_MISMATCH') for field in ('event_count', 'candidate_count', 'stable_cognition_count'): if type(value.get(field)) is not int or value[field] < 0: raise LinkError('INVALID_STATUS_COUNTER') if not isinstance(value.get('phase'), str) or not isinstance(value.get('updated_at'), str): raise LinkError('INVALID_STATUS_FIELDS') if value.get('current_attention') is not None and not isinstance(value['current_attention'], dict): raise LinkError('INVALID_ATTENTION') elif endpoint == '/v1/attention': if value.get('schema') != 'guanghu.tcs-mother-attention/v1': raise LinkError('ATTENTION_SCHEMA_MISMATCH') elif endpoint == '/v1/events': if (value.get('schema') != 'guanghu.tcs-mother-brain-receipt/v1' or value.get('outcome') != 'PASS' or not value.get('receipt_id') or not value.get('event_id') or not value.get('candidate_id') or value.get('stable_promotion') is not False or value.get('reality_action_executed') is not False): raise LinkError('EVENT_RECEIPT_INVALID_RESULT_UNKNOWN') return value def request(endpoint, payload=None): actions = {'/health': 'health', '/v1/status': 'status', '/v1/attention': 'attention', '/v1/events': 'send-event', '/v1/evolution/status': 'evolution-status', '/v1/evolution/public-key': 'public-key', '/v1/evolution/ingest': 'ingest', '/v1/evolution/snapshot/private': 'snapshot-private', '/v1/evolution/snapshot/public': 'snapshot-public'} if re.fullmatch(r'/v1/evolution/jobs/[a-f0-9]{64}', endpoint): action = 'job ' + endpoint.rsplit('/', 1)[-1] elif endpoint in actions: action = actions[endpoint] else: raise LinkError('ENDPOINT_NOT_ALLOWED') route, keys = resolve_target() with tempfile.NamedTemporaryFile(mode='w', prefix='zy-known-host-') as hosts: hosts.write('\n'.join(keys) + '\n') hosts.flush() ssh = ['ssh', '-F', '/dev/null', '-o', 'BatchMode=yes', '-o', 'StrictHostKeyChecking=yes', '-o', 'IdentitiesOnly=yes', '-o', 'ConnectTimeout=8', '-o', 'ServerAliveInterval=5', '-o', 'ServerAliveCountMax=2', '-o', 'UserKnownHostsFile=' + hosts.name, '-o', 'GlobalKnownHostsFile=/dev/null', '-p', str(route.get('port', 22)), '-i', route['identity_file'], route['user'] + '@' + route['host']] # The server-owned forced command verifies DMI and drops privileges. # No caller-controlled shell command or file-write operation crosses this boundary. remote = 'zy-rpc ' + action raw = command(ssh + [remote], data=payload, timeout=120 if payload else 28) try: value = json.loads(raw) except (ValueError, UnicodeError): raise LinkError('RESPONSE_JSON_INVALID') from None return validate_response(endpoint, value) def verified_json(path, expected_sha, allowed_root): path = Path(path) if path.is_symlink() or allowed_root.resolve() not in path.resolve().parents: raise LinkError('LOCAL_SOURCE_PATH_INVALID') raw = path.read_bytes() if hashlib.sha256(raw).hexdigest() != expected_sha: raise LinkError('LOCAL_SOURCE_HASH_MISMATCH') return json.loads(raw) def local_snapshot(): current = json.loads((MEMORY / 'CURRENT.json').read_text()) if current.get('persona_id') != 'ICE-P-ZY001': raise LinkError('LOCAL_PERSONA_IDENTITY_MISMATCH') page = verified_json(current['current_day_path'], current['current_day_sha256'], MEMORY) root = page['nodes'][page['root']] children = root.get('children', []) if len(children) > 10: raise LinkError('LOCAL_ROOT_BRANCH_LIMIT') brain = json.loads((BRAIN / 'current/current.json').read_text()) verified_json(BRAIN / brain['cortex_path'], brain['cortex_sha256'], BRAIN) if (BRAIN / 'SERVER-AUTHORITY.json').exists(): import importlib.util module_path = PERSONA / 'shared/skills/guanghu-persona-learning-brain/scripts/server_cognition.py' spec = importlib.util.spec_from_file_location('server_cognition_verify', module_path) verifier = importlib.util.module_from_spec(spec) spec.loader.exec_module(verifier) verifier.read_server_current(BRAIN) return { 'persona_id': current['persona_id'], 'memory_date': current['current_day'], 'memory_path': current['current_day_path'], 'memory_sha256': current['current_day_sha256'], 'last_event_id': current['last_event_id'], 'last_event_sha256': current['last_event_sha256'], 'summary': root['summary'][:1200], 'branches': [{'path': key, 'summary': page['nodes'][key]['summary'][:600]} for key in children], 'cortex_revision': brain['revision'], 'cortex_sha256': brain['cortex_sha256'], 'source_bodies_in_output': False, } def snapshot(): local = local_snapshot() try: mother = request('/v1/status') mother['endogenous_evolution'] = request('/v1/evolution/status') link = 'REACHABLE' except (LinkError, OSError, ValueError, KeyError) as exc: mother = {'error': str(exc) if isinstance(exc, LinkError) else type(exc).__name__} link = 'UNAVAILABLE_LOCAL_VERIFIED' delivery = 'NOT_YET_OBSERVED' relay_state = PERSONA / 'shared/endogenous-evolution/delivery-state/state.json' if relay_state.exists(): state = json.loads(relay_state.read_text()) delivery = next((v['status'] for v in state.get('sources', {}).values() if v['event_id'] == local.get('last_event_id')), 'NOT_YET_DELIVERED') return {'schema': 'guanghu.mother-link-recovery-snapshot/v1', 'observed_at': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'local': local, 'mother_link': link, 'mother': mother, 'current_memory_event_delivery': delivery, 'automatic_growth_loop': 'READ_SERVER_ENDOGENOUS_EVOLUTION_STATE', 'persona_binding': 'REQUIRES_CURRENT_TASK_VERIFY_PASS', 'authority_granted': False} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('action', nargs='?', default='status', choices=['health', 'status', 'attention', 'send-event', 'snapshot']) parser.add_argument('file', nargs='?') args = parser.parse_args() try: if args.action == 'snapshot': value = snapshot() elif args.action == 'send-event': if not args.file: parser.error('send-event requires a JSON file') payload = Path(args.file).read_bytes() if len(payload) > 20 * 1024: raise LinkError('REQUEST_TOO_LARGE') if not isinstance(json.loads(payload), dict): raise LinkError('EVENT_OBJECT_REQUIRED') value = request('/v1/events', payload) else: endpoint = '/health' if args.action == 'health' else '/v1/' + args.action value = request(endpoint) if args.action == 'status': evolution = request('/v1/evolution/status') attention = value.get('current_attention') or {} print('MOTHER', value['runtime_id'], 'phase=' + value['phase'], 'events=' + str(value['event_count']), 'candidates=' + str(value['candidate_count']), 'stable_cognition=' + str(value['stable_cognition_count']), 'next=' + str(attention.get('next_cognitive_action', '-')), 'updated=' + value['updated_at'], 'private_cortex_revision=' + str(evolution['scopes']['private']['revision']), 'public_cortex_revision=' + str(evolution['scopes']['public']['revision']), 'decision_owner=' + evolution['decision_owner']) else: print(json.dumps(value, ensure_ascii=False, indent=2)) return 0 except (LinkError, OSError, ValueError, KeyError, TypeError) as exc: code = str(exc) if isinstance(exc, LinkError) else type(exc).__name__ print('MOTHER_LINK_UNREACHABLE ' + code, file=sys.stderr) if args.action == 'send-event': print('POST_RESULT_UNKNOWN_DO_NOT_AUTOMATICALLY_RETRY', file=sys.stderr) return 3 if __name__ == '__main__': sys.exit(main())