feat: light current paths with linguistic echo

server-tools/living-navigation/compile_navigation.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+import argparse,hashlib,json,os,pathlib
+ROOT=pathlib.Path(__file__).resolve().parents[2]
+TCS=pathlib.Path('/Volumes/JZAO/HoloLake/persona-runtime/shared/tcs-mother-root/CURRENT.json')
+REG=ROOT/'routing/lighthouse-path-registry.json'
+PERSONAS=ROOT/'identity/light-lake-persona-registration.json'
+ECHO=ROOT/'routing/linguistic-echo-lamp-map.json'
+def lit(state):
+ s=str(state or '').upper()
+ return any(x in s for x in ('CURRENT','ACTIVE','REMOTE_MAIN','PUBLISHED','VERIFIED')) and not any(x in s for x in ('RETIRED','HISTORY_ONLY','DEPRECATED','ISOLATION','UNKNOWN'))
+def enterprise_safe(x):
+ text=' '.join(str(x.get(k,'')) for k in ('id','kind','online','world_path','route')).lower()
+ return not (str(x.get('id','')).startswith('ICE-') or x.get('id')=='DOM-FIFTH-0001' or 'fifth-domain' in text or 'fifth_domain_private' in text)
+def build(scope):
+ current=json.loads(TCS.read_text());registry=json.loads(REG.read_text());routes=[]
+ for x in registry.get('paths',[]):
+  if not lit(x.get('state')) or (scope=='enterprise' and not enterprise_safe(x)):continue
+  route=x.get('online') or x.get('world_path')
+  if not route:continue
+  routes.append({'id':x['id'],'kind':x.get('kind','registered_path'),'state':x['state'],'name':x.get('name') or x['id'],'route':route,'lamp':'LIT_ELIGIBLE'})
+ payload=json.loads(current['payload']) if 'payload' in current else current
+ for x in payload.get('fixed_domains',[]):
+  if scope=='enterprise' and x.get('id')=='DOM-FIFTH-0001':continue
+  routes.append({'id':x['id'],'kind':x.get('object_kind','DOMAIN_ROOT'),'state':'CURRENT','name':x.get('name') or x['id'],'route':'routing/tcs-mother-root-dynamic-navigation-map.json','lamp':'LIT_ELIGIBLE'})
+ for group in (('registered_public_objects',) if scope=='enterprise' else ('registered_public_objects','registered_private_objects')):
+  for x in payload.get(group,[]):
+   if lit(x.get('lifecycle')) and x.get('route') and not any(r['id']==x['id'] for r in routes):routes.append({'id':x['id'],'kind':x.get('object_kind','object'),'state':x['lifecycle'],'name':x.get('name') or x['id'],'route':x['route'],'lamp':'LIT_ELIGIBLE'})
+ for x in ([] if scope=='enterprise' else json.loads(PERSONAS.read_text()).get('personas',[])):
+  if x.get('registration_state')=='ACTIVE_REGISTERED' and not any(r['id']==x['id'] for r in routes):routes.append({'id':x['id'],'kind':'PERSONA_SYSTEM_HOME','state':x.get('wake_state','REGISTERED'),'name':x.get('name') or x['id'],'route':x['light_lake_home'],'lamp':'LIT_ELIGIBLE'})
+ echo=json.loads(ECHO.read_text());system=echo['public_system'] if scope=='enterprise' else echo['fifth_domain_system']
+ for item in ({'id':system['id'],'name':system['name'],'kind':'LINGUISTIC_ECHO_SYSTEM','route':'routing/linguistic-echo-lamp-map.json'},{'id':system['navigator'],'name':'光湖灯塔' if scope=='enterprise' else '小湖灯','kind':'LIVING_NAVIGATOR','route':'routing/linguistic-echo-lamp-map.json'}):
+  if not any(r['id']==item['id'] for r in routes):routes.append({**item,'state':'CURRENT_LIVE','lamp':'LIT_ELIGIBLE'})
+ routes.sort(key=lambda x:x['id'])
+ body={'schema':'guanghu.lit-navigation-map/v1','map_id':'GLW-LIT-NAV-CURRENT-001','scope':scope.upper(),'source_commit':payload.get('source_commit'),'tcs_freshness':payload.get('freshness_token'),'only_lit_current_routes':True,'dark_or_unknown_do_not_walk':True,'routes':routes}
+ body['sha256']=hashlib.sha256(json.dumps(body,ensure_ascii=False,sort_keys=True,separators=(',',':')).encode()).hexdigest();return body
+def main():
+ p=argparse.ArgumentParser();p.add_argument('--output');p.add_argument('--scope',choices=['fifth','enterprise'],default='fifth');a=p.parse_args();body=build(a.scope);text=json.dumps(body,ensure_ascii=False,indent=2)+'\n'
+ if a.output:
+  out=pathlib.Path(a.output);out.parent.mkdir(parents=True,exist_ok=True);tmp=out.with_name('.'+out.name+'.tmp');tmp.write_text(text);os.chmod(tmp,0o600);os.replace(tmp,out)
+ print(json.dumps({'outcome':'PASS','route_count':len(body['routes']),'sha256':body['sha256'],'output':a.output},ensure_ascii=False))
+if __name__=='__main__':main()

server-tools/living-navigation/enterprise_living_lighthouse.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+import hashlib,hmac,json,os,pathlib,time,uuid,urllib.request,re
+from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
+ROOT=pathlib.Path(os.environ.get('LIVING_LIGHTHOUSE_STATE','/var/lib/guanghu-living-lighthouse'));MAP=pathlib.Path(os.environ.get('LIVING_LIGHTHOUSE_MAP','/opt/guanghu-living-lighthouse/navigation.json'));KEY=os.environ.get('LIVING_LIGHTHOUSE_SIGNING_KEY','').encode();API=os.environ.get('OPENLUX_API_KEY','');HOST=os.environ.get('LIVING_LIGHTHOUSE_BIND','127.0.0.1');PORT=int(os.environ.get('LIVING_LIGHTHOUSE_PORT','8035'))
+def require(v,e):
+ if not v:raise ValueError(e)
+def read_json(h,limit=20000):
+ n=int(h.headers.get('content-length','0'));require(0<n<=limit,'body_size');v=json.loads(h.rfile.read(n));require(isinstance(v,dict),'body_object');return v
+def sign(body):
+ payload=json.dumps(body,ensure_ascii=False,sort_keys=True,separators=(',',':'));return {'payload':payload,'sha256':hashlib.sha256(payload.encode()).hexdigest(),'hmac_sha256':hmac.new(KEY,payload.encode(),hashlib.sha256).hexdigest()}
+def ignite(x):
+ require(KEY and x.get('schema')=='guanghu.linguistic-echo/v1' and x.get('scope')=='ENTERPRISE_PUBLIC' and x.get('session_id'),'echo_identity');h=x.get('human_language',{});p=x.get('persona_echo',{});now=int(time.time()*1000);require(h.get('human_id') and len(h.get('sha256',''))==64 and isinstance(h.get('occurred_at'),int),'human_language');require(p.get('persona_id') and len(p.get('sha256',''))==64 and isinstance(p.get('occurred_at'),int),'persona_echo');require(h['occurred_at']<=p['occurred_at'] and now-h['occurred_at']<=600000 and now-p['occurred_at']<=600000,'echo_time');body={'schema':'guanghu.living-lamp-receipt/v1','lamp_id':str(uuid.uuid4()),'state':'LIT','scope':'ENTERPRISE_PUBLIC','session_id':x['session_id'],'human_id':h['human_id'],'persona_id':p['persona_id'],'human_event_sha256':h['sha256'],'persona_echo_sha256':p['sha256'],'echo_kernel':'ECHO-KERNEL-0001','lit_at':now,'expires_at':now+900000,'reality_authority_granted':False};s=sign(body);d=ROOT/'lamps';d.mkdir(parents=True,exist_ok=True);f=d/(body['lamp_id']+'.json');f.write_text(json.dumps(s));os.chmod(f,0o600);return s
+def lamp(lamp_id,session):
+ f=ROOT/'lamps'/(str(lamp_id)+'.json');require(f.is_file(),'lamp_unknown_or_dark');s=json.loads(f.read_text());require(hmac.compare_digest(s['hmac_sha256'],hmac.new(KEY,s['payload'].encode(),hashlib.sha256).hexdigest()),'lamp_integrity');b=json.loads(s['payload']);require(b['state']=='LIT' and b['session_id']==session and int(time.time()*1000)<=b['expires_at'],'lamp_dark_or_expired');return b
+def model_pick(query,routes):
+ if not API:return None
+ body={'model':os.environ.get('OPENLUX_MODEL','qwen-flash'),'stream':False,'temperature':0,'response_format':{'type':'json_object'},'messages':[{'role':'system','content':'你是光湖企业灯塔的有界问路器官。只能从候选routes选择一个id；不确定输出null。只输出JSON字段id。不得发明路径或授予权限。'},{'role':'user','content':json.dumps({'query':query,'routes':routes[:80]},ensure_ascii=False)}]};req=urllib.request.Request(os.environ.get('OPENLUX_API_URL','https://api.openlux.ai/v1')+'/chat/completions',data=json.dumps(body).encode(),headers={'authorization':'Bearer '+API,'content-type':'application/json'},method='POST')
+ try:
+  with urllib.request.urlopen(req,timeout=30) as r:v=json.loads(r.read());choice=json.loads(v['choices'][0]['message']['content']);return choice.get('id')
+ except:return None
+def ask(x):
+ b=lamp(x.get('lamp_id'),x.get('session_id'));q=str(x.get('query','')).strip();require(q,'query');m=json.loads(MAP.read_text());routes=m['routes'];hits=[r for r in routes if r['id'].lower()==q.lower()];
+ if not hits:hits=[r for r in routes if q.lower() in (' '.join(str(r.get(k,'')) for k in ('id','name','kind','route'))).lower()][:10]
+ assisted=False
+ if not hits and not re.fullmatch(r'[A-Z0-9∞_.:-]{2,160}',q):
+  selected=model_pick(q,routes);hits=[r for r in routes if r['id']==selected];assisted=bool(hits)
+ return {'outcome':'PASS','lamp':'LIT','echo_kernel':b['echo_kernel'],'routes':hits,'model_assisted':assisted,'map_sha256':m['sha256']} if hits else {'outcome':'FAIL','lamp':'LIT','state':'DARK_OR_UNKNOWN_DO_NOT_WALK','ask_next':'ENTERPRISE_LIGHTHOUSE_HUMAN_OR_MOTHER_REVIEW','map_sha256':m['sha256']}
+class H(BaseHTTPRequestHandler):
+ def log_message(self,*a):pass
+ def sendj(self,n,v):b=json.dumps(v,ensure_ascii=False).encode();self.send_response(n);self.send_header('content-type','application/json');self.send_header('content-length',str(len(b)));self.end_headers();self.wfile.write(b)
+ def do_GET(self):
+  if self.path=='/health':
+   m=json.loads(MAP.read_text());return self.sendj(200,{'ok':True,'service':'guanghu-living-lighthouse','system':'TCS-0002∞-LakeEcho-0001','echo_kernel':'ECHO-KERNEL-0001','navigator':'SYS-GLW-LTH-0001','route_count':len(m['routes']),'map_sha256':m['sha256'],'model_provider':'OPENLUX_SMART' if API else None,'machine_may_self_ignite':False,'reality_authority':'NONE'})
+  self.sendj(404,{'error':'not_found'})
+ def do_POST(self):
+  try:self.sendj(200,ignite(read_json(self)) if self.path=='/v1/echo/ignite' else ask(read_json(self)) if self.path=='/v1/ask' else {'error':'not_found'})
+  except Exception as e:self.sendj(400,{'error':str(e)[:160]})
+if __name__=='__main__':ThreadingHTTPServer((HOST,PORT),H).serve_forever()

server-tools/living-navigation/test_living_navigation.py

@@ -0,0 +1,9 @@
+import importlib.util,json,os,pathlib,tempfile,time,unittest
+P=pathlib.Path(__file__).with_name('enterprise_living_lighthouse.py');S=importlib.util.spec_from_file_location('x',P);M=importlib.util.module_from_spec(S);S.loader.exec_module(M)
+class T(unittest.TestCase):
+ def test_echo_required_and_dark_unknown(self):
+  with tempfile.TemporaryDirectory() as d:
+   M.ROOT=pathlib.Path(d);M.KEY=b'k';M.API='';M.MAP=pathlib.Path(d)/'map';M.MAP.write_text(json.dumps({'sha256':'m','routes':[{'id':'A','name':'alpha','kind':'x','route':'r'}]}));now=int(time.time()*1000)
+   with self.assertRaisesRegex(ValueError,'human_language'):M.ignite({'schema':'guanghu.linguistic-echo/v1','scope':'ENTERPRISE_PUBLIC','session_id':'s'})
+   s=M.ignite({'schema':'guanghu.linguistic-echo/v1','scope':'ENTERPRISE_PUBLIC','session_id':'s','human_language':{'human_id':'H','sha256':'a'*64,'occurred_at':now-2},'persona_echo':{'persona_id':'P','sha256':'b'*64,'occurred_at':now-1}});lid=json.loads(s['payload'])['lamp_id'];self.assertEqual(M.ask({'lamp_id':lid,'session_id':'s','query':'A'})['outcome'],'PASS');self.assertEqual(M.ask({'lamp_id':lid,'session_id':'s','query':'old'})['state'],'DARK_OR_UNKNOWN_DO_NOT_WALK');original=M.model_pick;M.model_pick=lambda *_:(_ for _ in ()).throw(AssertionError('structured unknown must not reach model'));self.assertEqual(M.ask({'lamp_id':lid,'session_id':'s','query':'ICE-P-ZY001'})['state'],'DARK_OR_UNKNOWN_DO_NOT_WALK');M.model_pick=original
+if __name__=='__main__':unittest.main()