fix(hololake): establish language-world-native generation boundary

This commit is contained in:
冰朔 2026-09-09 18:26:26 +08:00
commit 6832933482
30 changed files with 1177 additions and 88 deletions

View file

@ -8,6 +8,7 @@ from typing import Any, Callable
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"
GENERATION=ROOT/"routing/hololake-language-world-native-generation-map.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"
@ -29,7 +30,7 @@ def git(*args:str)->str:
return r.stdout.strip()
def audit()->dict[str,Any]:
system,registry,container,legacy=load(SYSTEM),load(MODULES),load(SHELF/"persona-container-contract.json"),load(LEGACY)
system,registry,container,legacy,generation=load(SYSTEM),load(MODULES),load(SHELF/"persona-container-contract.json"),load(LEGACY),load(GENERATION)
errors=[]; modules=registry["modules"]; ids=[x["module_id"] for x in modules]; sequences=[x["sequence"] for x in modules]
if system["parent_channel"]!="ICE-CH-HB001": errors.append("WRONG_PARENT_CHANNEL")
if len(ids)!=len(set(ids)): errors.append("DUPLICATE_MODULE_ID")
@ -46,8 +47,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")
if generation.get("generation_boundary",{}).get("current_generation",{}).get("id")!="HLP-GEN-LANGUAGE-WORLD-NATIVE-0001": errors.append("CURRENT_GENERATION_MISSING")
if registry.get("generation_disposition",{}).get("current_product_sequence") is not False: errors.append("LEGACY_PRODUCT_SEQUENCE_NOT_FROZEN")
if system.get("generation",{}).get("legacy_resume_forbidden") is not True: errors.append("LEGACY_RESUME_NOT_FORBIDDEN")
living=registry.get("living_runtime",{})
if living.get("state")!="READY_17_CAPABILITY_MODULES_API_AND_ORGAN_VERIFIED" or not (LIVING_STATE/"CONTAINER.json").is_file(): errors.append("LIVING_CONTAINER_NOT_READY")
if living.get("state")!="VERIFIED_17_MODULE_LANGUAGE_RUNTIME_FROZEN_AS_LEGACY_DONOR" or not (LIVING_STATE/"CONTAINER.json").is_file(): errors.append("LEGACY_LIVING_CONTAINER_NOT_PRESERVED")
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_AND_CAPABILITY_VERIFIED": errors.append(f"LIVING_MODULE_NOT_READY:{item['module_id']}")
@ -56,11 +60,10 @@ def audit()->dict[str,Any]:
def snapshot()->dict[str,Any]:
check=audit()
if check["outcome"]!="PASS": raise ShelfError("SHELF_AUDIT_FAILED:"+",".join(check["errors"]))
system,registry,legacy=load(SYSTEM),load(MODULES),load(LEGACY); modules=registry["modules"]
system,registry,legacy,generation=load(SYSTEM),load(MODULES),load(LEGACY),load(GENERATION); modules=registry["modules"]
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_RUNTIME_17_CAPABILITY_MODULES_API_AND_ORGAN_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={"schema":"guanghu.heartbeat-engineering-shelf-current/v1","state":system["state"],"system_id":system["system_id"],"shelf_id":system["shelf_id"],"project_id":system["project_id"],"legacy_project_id":system["legacy_project_id"],"generation":generation["generation_boundary"],"repo012_head":git("rev-parse","HEAD"),"repo014_main":system["official_product_architecture"]["main"],"module_count":0,"legacy_verified_donor_module_count":len(modules),"status_counts":counts,"current_language_module_id":None,"current_product_module_id":None,"current_generation_module_numbering":"NOT_YET_REDERIVED","modules":modules,"legacy_parts":legacy["parts"],"language_runtime":registry["living_runtime"],"reality_engineering_started":False,"zero_core_dispatch":"FROZEN_PENDING_CURRENT_GENERATION_MODULE_REDERIVATION"}
value["freshness_token"]=digest(stable(value)); return value
def sync(state:Path)->dict[str,Any]:
@ -78,7 +81,9 @@ def score(text:str,item:dict[str,Any])->int:
return sum((20 if token==item.get("module_id","").lower() else len(token)) for token in tokens if token in hay)
def query(text:str)->dict[str,Any]:
if not text.strip(): raise ShelfError("QUERY_REQUIRED")
items=[*load(MODULES)["modules"],*load(LEGACY)["parts"]]
generation=load(GENERATION)["generation_boundary"]["current_generation"]
generation_card={"generation_id":generation["id"],"name":generation["name"],"function":"语言唯一入口、人格体自动执行、无限白布实时渲染、模块功能引擎与UI分离","state":generation["lifecycle"]}
items=[generation_card,*load(MODULES)["modules"],*load(LEGACY)["parts"]]
ranked=sorted(((score(text,x),x) for x in items),key=lambda x:(-x[0],json.dumps(x[1],ensure_ascii=False)))
matches=[x for s,x in ranked if s>0][:5]
return {"outcome":"PASS","query":text,"matches":matches,"exact_evidence_only":True,"authority_granted":False}
@ -96,11 +101,10 @@ 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_language_module_id","current_product_module_id","language_runtime","reality_engineering_started","zero_core_dispatch"]}
value=snapshot(); return {k:value[k] for k in ["state","system_id","shelf_id","project_id","legacy_project_id","generation","repo012_head","repo014_main","module_count","legacy_verified_donor_module_count","status_counts","current_language_module_id","current_product_module_id","current_generation_module_numbering","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_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"],"language_runtime_state":"READY_17_CAPABILITY_MODULES_API_AND_ORGAN_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}
value=snapshot()
return {"outcome":"PASS","project_id":value["project_id"],"generation":value["generation"]["current_generation"],"language_runtime_state":"OLD_17_MODULE_RUNTIME_VERIFIED_DONOR_FROZEN","legacy_verified_donor_module_count":value["legacy_verified_donor_module_count"],"current_product_module":None,"current_generation_module_numbering":"NOT_YET_REDERIVED","next_action":"REDERIVE_LANGUAGE_WORLD_NATIVE_MODULE_REGISTRY_FROM_GENERATION_MAP","next_product_gate":None,"reality_engineering_started":False,"zero_core_dispatch":value["zero_core_dispatch"],"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()

View file

@ -7,13 +7,15 @@ class Tests(unittest.TestCase):
r=M.audit();self.assertEqual(r['outcome'],'PASS');self.assertEqual(r['module_count'],17);self.assertEqual(r['cumulative_gate_count'],17)
def test_sync_builds_live_current_without_claiming_engineering(self):
with tempfile.TemporaryDirectory() as t:
M.sync(Path(t));v=M.load(Path(t)/'CURRENT.json');self.assertFalse(v['reality_engineering_started']);self.assertEqual(v['zero_core_dispatch'],'NOT_ISSUED')
M.sync(Path(t));v=M.load(Path(t)/'CURRENT.json');self.assertFalse(v['reality_engineering_started']);self.assertEqual(v['zero_core_dispatch'],'FROZEN_PENDING_CURRENT_GENERATION_MODULE_REDERIVATION');self.assertIsNone(v['current_product_module_id'])
def test_exact_query_finds_current_and_legacy_parts(self):
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_reports_language_complete_and_first_product_module(self):
r=M.resume();self.assertEqual(r['language_runtime_state'],'READY_17_CAPABILITY_MODULES_API_AND_ORGAN_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_resume_reports_generation_rederivation_not_old_module(self):
r=M.resume();self.assertEqual(r['language_runtime_state'],'OLD_17_MODULE_RUNTIME_VERIFIED_DONOR_FROZEN');self.assertIsNone(r['current_product_module']);self.assertIsNone(r['next_product_gate']);self.assertEqual(r['next_action'],'REDERIVE_LANGUAGE_WORLD_NATIVE_MODULE_REGISTRY_FROM_GENERATION_MAP')
def test_language_world_native_generation_is_queryable(self):
r=M.query('无限白布 实时渲染 语言唯一入口');self.assertEqual(r['matches'][0]['generation_id'],'HLP-GEN-LANGUAGE-WORLD-NATIVE-0001')
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'])