#!/usr/bin/env python3 """Atomically bind every registered model-backed Agent slot to the current persona.""" from __future__ import annotations import argparse, hashlib, importlib.util, json, os from pathlib import Path import shutil import subprocess from typing import Any ROOT = Path(__file__).resolve().parents[2] MAP = ROOT / "routing/persona-dynamic-subagent-command-map.json" PERSONAS = ROOT / "routing/persona-system-canonical-map.json" DEFAULT_STATE = Path("/Volumes/JZAO/HoloLake/persona-runtime/shared/persona-subagent-command/codex") RUNTIMES = {"codex":"/Applications/ChatGPT.app/Contents/Resources/codex", "qwen":"/Users/bingshuolingdianyuanhe/.npm-global/bin/qwen"} OPENLUX_ROUTER = ROOT / "server-tools/persona-model-smart-router/openlux_router.py" SHARED_COGNITION = Path('/Volumes/JZAO/HoloLake/persona-runtime/shared/tcs-shared-cognition/CURRENT.json') class CommandError(RuntimeError): pass 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 load(p: Path) -> dict[str, Any]: v=json.loads(p.read_text()) if not isinstance(v,dict): raise CommandError("JSON_OBJECT_REQUIRED") return v def runtime_available(name: str) -> bool: if name == "openlux-smart": result=subprocess.run(["security","find-generic-password","-a","guanghu-persona-agent-router","-s","ai.openlux.api"],capture_output=True,timeout=10) return result.returncode == 0 if name == "hololake-model-gateway": return bool(os.environ.get("HOLOLAKE_MODEL_GATEWAY_URL")) path=RUNTIMES.get(name) return bool(path and Path(path).is_file() and shutil.which(path)) def registered_personas() -> set[str]: return {x["id"] for x in load(PERSONAS)["persona_systems"]} 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 persist(path: Path, value: dict[str, Any]) -> None: value.pop("binding_sha256", None) value["binding_sha256"] = digest(stable(value)) atomic(path, value) def activate(event: dict[str, Any], state: Path) -> dict[str, Any]: for k in ("parent_persona_id","session_id","channel_id","current_intent_sha256","model_runtime"): if not isinstance(event.get(k),str) or not event[k].strip(): raise CommandError(f"REQUIRED_FIELD:{k}") if event["parent_persona_id"] not in registered_personas(): raise CommandError("PARENT_PERSONA_NOT_REGISTERED") if not runtime_available(event["model_runtime"]): raise CommandError("MODEL_PROVIDER_UNAVAILABLE_NO_AGENT_ACTIVATION") current_path=state/"CURRENT.json" previous=load(current_path) if current_path.is_file() else None generation=(previous or {}).get("generation",0)+1 config=load(MAP) shared=load(SHARED_COGNITION) if digest(shared['artifact']['payload'].encode()) != shared['artifact']['sha256']: raise CommandError('SHARED_COGNITION_HASH_MISMATCH') team=[] for slot in [*config["agent_slots"], *config.get("treatment_slots", [])]: team.append({ "child_agent_id":f"SUBAGENT::{event['parent_persona_id']}::{slot['slot_id']}", "slot_id":slot["slot_id"], "agent_role":slot["agent_role"], "parent_persona_id":event["parent_persona_id"], "controller":event["parent_persona_id"], "model_cognition_required":True, "model_runtime":event["model_runtime"], "cognition_adapter":"server-tools/persona-execution-limb-agent/persona_limb_agent.py", "shared_cognition_sha256":shared['artifact']['sha256'], "deterministic_organ":slot.get("deterministic_organ"), "task_authority":"NONE_UNTIL_PARENT_ENVELOPE", "runtime_state":"DORMANT_READY", "model_context_loaded":False }) result={"schema":"guanghu.persona-subagent-command-current/v1","state":"READY_UNDER_CURRENT_PERSONA_NO_AGENT_RUNNING","generation":generation,"parent_persona_id":event["parent_persona_id"],"session_id":event["session_id"],"channel_id":event["channel_id"],"current_intent_sha256":event["current_intent_sha256"],"model_runtime":event["model_runtime"],"shared_cognition":{"revision":shared['brain']['revision'],"sha256":shared['artifact']['sha256'],"principles":shared['brain']['principles'],"truth_order":shared['brain']['truth_order']},"team":team,"active_agent_count":0,"prior_generation_invalidated":bool(previous),"credentials_stored":False,"authority_granted":False} persist(current_path,result) return result def dispatch(event: dict[str, Any], state: Path) -> dict[str, Any]: current=load(state/"CURRENT.json") if event.get("generation") != current["generation"]: raise CommandError("STALE_TEAM_GENERATION") matches=[x for x in current["team"] if x["child_agent_id"]==event.get("child_agent_id")] if len(matches)!=1: raise CommandError("CHILD_AGENT_NOT_IN_ACTIVE_TEAM") if not isinstance(event.get("task"),str) or not event["task"].strip(): raise CommandError("TASK_REQUIRED") if any(x.get("runtime_state") == "ACTIVE" for x in current["team"]): raise CommandError("ANOTHER_AGENT_ALREADY_ACTIVE") item=matches[0] item["runtime_state"]="ACTIVE"; item["model_context_loaded"]=True current["state"]="ONE_AGENT_ACTIVE"; current["active_agent_count"]=1 persist(state/"CURRENT.json",current) runtime=current["model_runtime"] try: if runtime == "openlux-smart": spec=importlib.util.spec_from_file_location("openlux_router",OPENLUX_ROUTER); module=importlib.util.module_from_spec(spec); spec.loader.exec_module(module) result=module.route({"parent_persona_id":current["parent_persona_id"],"child_agent_id":item["child_agent_id"],"controller":current["parent_persona_id"],"channel_id":current["channel_id"],"agent_role":item["agent_role"],"authority":"CURRENT_PARENT_TASK_ENVELOPE_ONLY","shared_cognition":current["shared_cognition"],"task":event["task"],"complexity":event.get("complexity","LOW")}) outcome=result["outcome"] returned={**result,"generation":current["generation"],"child_agent_id":item["child_agent_id"],"model_runtime":runtime,"model_api_invoked":True} else: if runtime not in {"codex","qwen"}: raise CommandError("GATEWAY_DISPATCH_NOT_CONFIGURED") cmd=[os.sys.executable,str(ROOT/"server-tools/persona-execution-limb-agent/persona_limb_agent.py"),"run","--persona",current["parent_persona_id"],"--task-id",event.get("task_id","SUBAGENT-TASK"),"--task",event["task"],"--runtime",runtime,"--cwd",event.get("cwd",str(ROOT))] completed=subprocess.run(cmd,text=True,capture_output=True,timeout=240,check=False) outcome="PASS" if completed.returncode==0 else "FAIL" returned={"outcome":outcome,"generation":current["generation"],"child_agent_id":event["child_agent_id"],"model_runtime":runtime,"model_api_invoked":True,"receipt":completed.stdout[-20000:],"error":completed.stderr[-2000:]} item["model_context_loaded"]=False if outcome == "PASS": item["runtime_state"]="DORMANT_READY" current["state"]="READY_UNDER_CURRENT_PERSONA_NO_AGENT_RUNNING"; current["active_agent_count"]=0 item["last_receipt_sha256"]=digest(stable({k:v for k,v in returned.items() if k not in {"receipt","response","error"}})) returned["final_slot_state"]="DORMANT_READY" else: item["runtime_state"]="SYMPTOM_REPORTED" current["state"]="SYMPTOM_REPORTED_TREATMENT_AVAILABLE"; current["active_agent_count"]=0 current["treatment_signal"]={"failed_slot_id":item["slot_id"],"failure_class":"MODEL_OR_EXECUTION_FAILURE","treatment_router":"TCS-AGENT-TREATMENT-ROUTER-001"} returned["final_slot_state"]="SYMPTOM_REPORTED" persist(state/"CURRENT.json",current) return returned except Exception as error: item["runtime_state"]="SYMPTOM_REPORTED"; item["model_context_loaded"]=False current["state"]="SYMPTOM_REPORTED_TREATMENT_AVAILABLE"; current["active_agent_count"]=0 current["treatment_signal"]={"failed_slot_id":item["slot_id"],"failure_class":type(error).__name__,"treatment_router":"TCS-AGENT-TREATMENT-ROUTER-001"} persist(state/"CURRENT.json",current) raise def main() -> int: p=argparse.ArgumentParser(); p.add_argument("command",choices=("probe","activate","status","dispatch")); p.add_argument("--input"); p.add_argument("--stdin",action="store_true"); p.add_argument("--state-root",default=str(DEFAULT_STATE)); a=p.parse_args(); state=Path(a.state_root) try: if a.command=="probe": result={"outcome":"PASS","model_runtimes":{k:runtime_available(k) for k in [*RUNTIMES,"openlux-smart","hololake-model-gateway"]},"repository_credentials":False} elif a.command=="status": result=load(state/"CURRENT.json") if (state/"CURRENT.json").is_file() else {"outcome":"FAIL","state":"NO_ACTIVE_COMMANDER"} else: if a.stdin: event=json.load(os.sys.stdin) elif a.input: event=load(Path(a.input)) else: raise CommandError("INPUT_REQUIRED") result=activate(event,state) if a.command=="activate" else dispatch(event,state) print(json.dumps(result,ensure_ascii=False,indent=2,sort_keys=True)); return 0 except Exception as e: print(json.dumps({"outcome":"FAIL","error":str(e)},ensure_ascii=False)); return 2 if __name__=="__main__": raise SystemExit(main())