zhuyuan-ice-gl-zy001/BRIDGE/tools/zy-mother-link.py

231 lines
11 KiB
Python

#!/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())