feat: register Heartbeat HoloLake engineering shelf

server-tools/heartbeat-engineering-shelf/engineering_shelf.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""Live evidence index and model-backed navigator for the Heartbeat engineering shelf."""
+from __future__ import annotations
+import argparse, hashlib, importlib.util, json, os, re, subprocess
+from pathlib import Path, PurePosixPath
+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"
+DEFAULT_STATE=Path("/Volumes/JZAO/HoloLake/persona-runtime/shared/heartbeat-engineering-shelf")
+OPENLUX=ROOT/"server-tools/persona-model-smart-router/openlux_router.py"
+
+class ShelfError(RuntimeError): pass
+def load(path:Path)->dict[str,Any]:
+    value=json.loads(path.read_text());
+    if not isinstance(value,dict): raise ShelfError("JSON_OBJECT_REQUIRED")
+    return value
+def stable(v:Any)->bytes:return (json.dumps(v,ensure_ascii=False,sort_keys=True,separators=(",",":"))+"\n").encode()
+def digest(v:bytes)->str:return hashlib.sha256(v).hexdigest()
+def atomic(path:Path,value:dict[str,Any])->None:
+    path.parent.mkdir(parents=True,exist_ok=True); body=json.dumps(value,ensure_ascii=False,indent=2,sort_keys=True).encode()+b"\n"; temp=path.with_name(f".{path.name}.{os.getpid()}.tmp")
+    with temp.open("wb") as f: os.chmod(temp,0o600); f.write(body); f.flush(); os.fsync(f.fileno())
+    os.replace(temp,path)
+def git(*args:str)->str:
+    r=subprocess.run(["git","-C",str(ROOT),*args],text=True,capture_output=True,timeout=15)
+    if r.returncode: raise ShelfError("GIT_READ_FAILED")
+    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)
+    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")
+    if sequences!=list(range(1,len(modules)+1)): errors.append("NON_CONTIGUOUS_SEQUENCE")
+    seen=set()
+    for item in modules:
+        required=set(registry["module_card_required"])-set(item)
+        if required: errors.append(f"MISSING_FIELDS:{item['module_id']}:{','.join(sorted(required))}")
+        if not set(item["depends_on"]).issubset(seen): errors.append(f"FORWARD_OR_UNKNOWN_DEPENDENCY:{item['module_id']}")
+        expected=f"GATE-HLP-STAGE1-{item['sequence']:03d}"
+        if item["cumulative_gate"]!=expected: errors.append(f"CUMULATIVE_GATE_MISMATCH:{item['module_id']}")
+        if item["component_type"].startswith("MODEL_BACKED") and item["model_api"] is not True: errors.append(f"MODEL_API_REQUIRED:{item['module_id']}")
+        seen.add(item["module_id"])
+    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")
+    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]:
+    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"]
+    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["freshness_token"]=digest(stable(value)); return value
+
+def sync(state:Path)->dict[str,Any]:
+    value=snapshot(); atomic(state/"CURRENT.json",value); return {"outcome":"PASS","state":value["state"],"current":str(state/"CURRENT.json"),"freshness_token":value["freshness_token"]}
+
+def score(text:str,item:dict[str,Any])->int:
+    tokens=[x for x in re.split(r"[^\w\u4e00-\u9fff]+",text.lower()) if x]
+    expanded=[]
+    for token in tokens:
+        expanded.append(token)
+        if re.search(r"[\u4e00-\u9fff]",token) and len(token)>2:
+            expanded.extend(token[i:i+2] for i in range(len(token)-1))
+    tokens=expanded
+    hay=json.dumps(item,ensure_ascii=False).lower()
+    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"]]
+    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}
+
+def smart_query(text:str, route_func:Callable[[dict[str,Any]],dict[str,Any]]|None=None)->dict[str,Any]:
+    cards=[{"id":x["module_id"],"name":x["name"],"function":x["function"],"status":x["product_status"]} for x in load(MODULES)["modules"]]
+    if route_func is None:
+        spec=importlib.util.spec_from_file_location("openlux",OPENLUX); module=importlib.util.module_from_spec(spec); spec.loader.exec_module(module); route_func=module.route
+    request={"parent_persona_id":"ICE-P-ZY001","child_agent_id":"SUBAGENT::ICE-P-ZY001::HEARTBEAT_ENGINEERING_NAV","controller":"ICE-P-ZY001","channel_id":"ICE-CH-HB001","agent_role":"HEARTBEAT_ENGINEERING_SHELF_NAVIGATOR","authority":"READ_ONLY_INDEX_QUERY","complexity":"LOW","task":"根据问题只返回JSON：{\"module_ids\":[最多5个目录中存在的编号]}。问题："+text+"。目录："+json.dumps(cards,ensure_ascii=False)}
+    result=route_func(request); raw=result.get("response","")
+    try:
+        match=re.search(r"\{[\s\S]*\}",raw); selected=json.loads(match.group(0))["module_ids"] if match else []
+    except Exception as error: raise ShelfError("SMART_QUERY_RESPONSE_INVALID") from error
+    known={x["module_id"]:x for x in load(MODULES)["modules"]}; selected=[x for x in selected if x in known][:5]
+    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"]}
+def resume()->dict[str,Any]:
+    value=snapshot(); current=next(x for x in value["modules"] if x["module_id"]==value["current_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}
+
+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()
+    try:
+        if a.command=="audit": result=audit()
+        elif a.command=="sync": result=sync(Path(a.state_root))
+        elif a.command=="status": result=status()
+        elif a.command=="resume": result=resume()
+        elif a.command=="query": result=query(a.text)
+        else: result=smart_query(a.text)
+        print(json.dumps(result,ensure_ascii=False,indent=2,sort_keys=True)); return 0 if result.get("outcome","PASS")=="PASS" else 2
+    except Exception as error: print(json.dumps({"outcome":"FAIL","error":str(error)},ensure_ascii=False)); return 2
+if __name__=="__main__": raise SystemExit(main())