#!/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())