110 lines
8.4 KiB
Python
110 lines
8.4 KiB
Python
#!/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())
|