bingshuo/guanghu-ice-heart
Watch
1
0
Fork
You've already forked guanghu-ice-heart
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

feat: run HoloLake stage-one living persona modules

Browse source
This commit is contained in:
冰朔 2026-09-08 20:52:13 +08:00
commit f3eb9e704c
12 changed files with 362 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

14
server-tools/heartbeat-engineering-shelf/engineering_shelf.py
View file

@ -9,6 +9,7 @@ ROOT=Path(__file__).resolve().parents[2]
SHELF=ROOT/"eternal-lake-heart/heartbeat-core/guanghu-world-engineering-system"
SYSTEM=SHELF/"system.json"; MODULES=SHELF/"module-registry.json"; LEGACY=SHELF/"legacy-parts-registry.json"
DEFAULT_STATE=Path("/Volumes/JZAO/HoloLake/persona-runtime/shared/heartbeat-engineering-shelf")
LIVING_STATE=Path("/Volumes/JZAO/HoloLake/persona-runtime/shared/heartbeat-stage1-persona-container")
OPENLUX=ROOT/"server-tools/persona-model-smart-router/openlux_router.py"
class ShelfError(RuntimeError): pass
@ -45,6 +46,11 @@ def audit()->dict[str,Any]:
if container["public_container"]["fifth_domain_access"] is not False: errors.append("PUBLIC_CONTAINER_FIFTH_DOMAIN_ACCESS")
if container["compatibility"]["same_memory"] is not False: errors.append("CONTAINER_MEMORY_COLLAPSE")
if not legacy["parts"]: errors.append("LEGACY_PARTS_EMPTY")
living=registry.get("living_runtime",{})
if living.get("state")!="READY_17_MODULES_API_VERIFIED" or not (LIVING_STATE/"CONTAINER.json").is_file(): errors.append("LIVING_CONTAINER_NOT_READY")
for item in modules:
current=LIVING_STATE/"modules"/item["module_id"]/"CURRENT.json"
if not current.is_file() or load(current).get("state")!="READY_API_VERIFIED": errors.append(f"LIVING_MODULE_NOT_READY:{item['module_id']}")
return {"outcome":"PASS" if not errors else "FAIL","errors":errors,"module_count":len(modules),"legacy_part_count":len(legacy["parts"]),"cumulative_gate_count":len(modules)}
def snapshot()->dict[str,Any]:
@ -54,7 +60,7 @@ def snapshot()->dict[str,Any]:
counts={}
for item in modules: counts[item["product_status"]]=counts.get(item["product_status"],0)+1
current=next((x for x in modules if x["product_status"] not in {"RELEASED","RELEASED_SIGNED_NOTARIZED_PUBLIC"}),None)
value={"schema":"guanghu.heartbeat-engineering-shelf-current/v1","state":"CURRENT_LANGUAGE_PROJECT_INDEX_VERIFIED","system_id":system["system_id"],"shelf_id":system["shelf_id"],"project_id":system["project_id"],"repo012_head":git("rev-parse","HEAD"),"repo014_main":system["official_product_architecture"]["main"],"module_count":len(modules),"status_counts":counts,"current_module_id":current["module_id"] if current else None,"modules":modules,"legacy_parts":legacy["parts"],"reality_engineering_started":False,"zero_core_dispatch":"NOT_ISSUED"}
value={"schema":"guanghu.heartbeat-engineering-shelf-current/v1","state":"CURRENT_LANGUAGE_RUNTIME_17_MODULES_API_VERIFIED","system_id":system["system_id"],"shelf_id":system["shelf_id"],"project_id":system["project_id"],"repo012_head":git("rev-parse","HEAD"),"repo014_main":system["official_product_architecture"]["main"],"module_count":len(modules),"status_counts":counts,"current_language_module_id":None,"current_product_module_id":current["module_id"] if current else None,"modules":modules,"legacy_parts":legacy["parts"],"language_runtime":registry["living_runtime"],"reality_engineering_started":False,"zero_core_dispatch":"NOT_ISSUED"}
value["freshness_token"]=digest(stable(value)); return value
def sync(state:Path)->dict[str,Any]:
@ -90,11 +96,11 @@ def smart_query(text:str, route_func:Callable[[dict[str,Any]],dict[str,Any]]|Non
return {"outcome":"PASS","selected_model":result.get("selected_model"),"matches":[known[x] for x in selected],"model_may_select_only_registered_ids":True,"authority_granted":False}
def status()->dict[str,Any]:
value=snapshot(); return {k:value[k] for k in ["state","system_id","shelf_id","project_id","repo012_head","repo014_main","module_count","status_counts","current_module_id","reality_engineering_started","zero_core_dispatch"]}
value=snapshot(); return {k:value[k] for k in ["state","system_id","shelf_id","project_id","repo012_head","repo014_main","module_count","status_counts","current_language_module_id","current_product_module_id","language_runtime","reality_engineering_started","zero_core_dispatch"]}
def resume()->dict[str,Any]:
value=snapshot(); current=next(x for x in value["modules"] if x["module_id"]==value["current_module_id"])
value=snapshot(); current=next(x for x in value["modules"] if x["module_id"]==value["current_product_module_id"])
prior=[x["module_id"] for x in value["modules"] if x["sequence"]<current["sequence"]]
return {"outcome":"PASS","project_id":value["project_id"],"current":current,"prior_cumulative_scope":prior,"next_gate":current["cumulative_gate"],"reality_engineering_started":False,"requires_dual_signature_before_zero_core":True}
return {"outcome":"PASS","project_id":value["project_id"],"language_runtime_state":"READY_17_MODULES_API_VERIFIED","current_product_module":current,"prior_product_cumulative_scope":prior,"next_product_gate":current["cumulative_gate"],"reality_engineering_started":False,"requires_dual_signature_before_zero_core":True}
def main()->int:
p=argparse.ArgumentParser();p.add_argument("command",choices=("audit","sync","status","query","smart-query","resume"));p.add_argument("--text",default="");p.add_argument("--state-root",default=str(DEFAULT_STATE));a=p.parse_args()

4
server-tools/heartbeat-engineering-shelf/test_engineering_shelf.py
View file

@ -12,8 +12,8 @@ class Tests(unittest.TestCase):
self.assertEqual(M.query('HLP-LANG-MOD-005')['matches'][0]['module_id'],'HLP-LANG-MOD-005')
self.assertTrue(any(x.get('part_id')=='HLP-PART-MODEL-RECEIPT-DONOR-0_5_8' for x in M.query('旧模型回执供体')['matches']))
self.assertTrue(any(x.get('part_id')=='HLP-PART-MODEL-RECEIPT-DONOR-0_5_8' for x in M.query('旧模型回执供体现在能不能用')['matches']))
def test_resume_returns_first_unimplemented_module(self):
r=M.resume();self.assertEqual(r['current']['module_id'],'HLP-LANG-MOD-001');self.assertEqual(r['next_gate'],'GATE-HLP-STAGE1-001')
def test_resume_reports_language_complete_and_first_product_module(self):
r=M.resume();self.assertEqual(r['language_runtime_state'],'READY_17_MODULES_API_VERIFIED');self.assertEqual(r['current_product_module']['module_id'],'HLP-LANG-MOD-001');self.assertEqual(r['next_product_gate'],'GATE-HLP-STAGE1-001')
def test_smart_query_model_can_only_return_registered_ids(self):
def fake(_): return {'response':'{"module_ids":["HLP-LANG-MOD-015","FAKE"]}','selected_model':'mock'}
r=M.smart_query('监管审核',fake);self.assertEqual([x['module_id'] for x in r['matches']],['HLP-LANG-MOD-015'])