feat: bind all agent slots to active persona model routing

server-tools/persona-dynamic-subagent-command/README.md

@@ -0,0 +1,5 @@
+# Persona Dynamic Subagent Command Router
+
+On a verified persona wake or switch, this router atomically rebuilds every registered Agent job slot as a model-backed child Agent of the current persona. The stable job remains; the commander, derived child IDs, model context and task envelopes switch together.
+
+Deterministic programs remain attached as organs. They are not called personas or Agents by themselves. `dispatch` goes through the existing persona limb model runtime (`codex` or `qwen`); credentials stay in the host or server provider layer and never enter the team snapshot.

server-tools/persona-dynamic-subagent-command/command_router.py

@@ -0,0 +1,112 @@
+#!/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")
+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"
+
+
+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 activate(event: dict[str, Any], state: Path) -> dict[str, Any]:
+    for k in ("parent_persona_id","session_id","channel_id","direct_event_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)
+    team=[]
+    for slot in config["agent_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",
+            "deterministic_organ":slot["deterministic_organ"], "task_authority":"NONE_UNTIL_PARENT_ENVELOPE"
+        })
+    result={"schema":"guanghu.persona-subagent-command-current/v1","state":"ACTIVE_UNDER_CURRENT_PERSONA","generation":generation,"parent_persona_id":event["parent_persona_id"],"session_id":event["session_id"],"channel_id":event["channel_id"],"direct_event_sha256":event["direct_event_sha256"],"model_runtime":event["model_runtime"],"team":team,"prior_generation_invalidated":bool(previous),"credentials_stored":False,"authority_granted":False}
+    result["binding_sha256"]=digest(stable(result))
+    atomic(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")
+    runtime=current["model_runtime"]
+    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)
+        item=matches[0]
+        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","task":event["task"],"complexity":event.get("complexity","LOW")})
+        return {**result,"generation":current["generation"],"child_agent_id":item["child_agent_id"],"model_runtime":runtime,"model_api_invoked":True}
+    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)
+    return {"outcome":"PASS" if completed.returncode==0 else "FAIL","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:]}
+
+
+def main() -> int:
+    p=argparse.ArgumentParser(); p.add_argument("command",choices=("probe","activate","status","dispatch")); p.add_argument("--input"); 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 not a.input: raise CommandError("INPUT_REQUIRED")
+            event=load(Path(a.input)); 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())

server-tools/persona-dynamic-subagent-command/test_command_router.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+import importlib.util,json,tempfile,unittest
+from pathlib import Path
+from unittest.mock import patch
+S=Path(__file__).with_name('command_router.py'); spec=importlib.util.spec_from_file_location('router',S); M=importlib.util.module_from_spec(spec); spec.loader.exec_module(M)
+
+def event(persona='ICE-P-ZY001', session='s1'):
+    return {"parent_persona_id":persona,"session_id":session,"channel_id":"ICE-CH-HB001","direct_event_sha256":"a"*64,"model_runtime":"codex"}
+
+class Tests(unittest.TestCase):
+    def test_every_slot_is_model_backed_and_owned_by_parent(self):
+        with tempfile.TemporaryDirectory() as t, patch.object(M,'runtime_available',return_value=True):
+            v=M.activate(event(),Path(t)); self.assertEqual(len(v['team']),4)
+            self.assertTrue(all(x['model_cognition_required'] and x['controller']=='ICE-P-ZY001' for x in v['team']))
+    def test_persona_switch_rebinds_all_slots_and_invalidates_generation(self):
+        personas=iter(sorted(M.registered_personas()))
+        first=next(personas); second=next(x for x in personas if x!=first)
+        with tempfile.TemporaryDirectory() as t, patch.object(M,'runtime_available',return_value=True):
+            state=Path(t); a=M.activate(event(first,'s1'),state); b=M.activate(event(second,'s2'),state)
+            self.assertEqual(b['generation'],a['generation']+1); self.assertTrue(b['prior_generation_invalidated'])
+            self.assertTrue(all(x['parent_persona_id']==second and second in x['child_agent_id'] for x in b['team']))
+            with self.assertRaisesRegex(M.CommandError,'STALE_TEAM_GENERATION'): M.dispatch({"generation":a['generation'],"child_agent_id":a['team'][0]['child_agent_id'],"task":"x"},state)
+    def test_unknown_persona_and_missing_provider_fail_closed(self):
+        with tempfile.TemporaryDirectory() as t:
+            with self.assertRaisesRegex(M.CommandError,'NOT_REGISTERED'): M.activate(event('PER-NOT-REAL'),Path(t))
+            with patch.object(M,'runtime_available',return_value=False), self.assertRaisesRegex(M.CommandError,'PROVIDER_UNAVAILABLE'): M.activate(event(),Path(t))
+
+if __name__=='__main__': unittest.main()

server-tools/persona-model-smart-router/README.md

@@ -0,0 +1,5 @@
+# OpenLux cost-first persona model router
+
+The router uses the OpenAI-compatible endpoint at `https://api.openlux.ai/v1`. Its API key is read at call time from macOS Keychain service `ai.openlux.api`, account `guanghu-persona-agent-router`; no secret is stored in this repository.
+
+Each call must carry the active parent persona, derived child Agent ID, controller, channel, job role, bounded authority and task. The router tries the configured low-cost healthy models first and escalates only after a transport, empty-response or downstream validation failure. Provider pricing is not exposed by the current `/models` response, so the configured order is explicitly a heuristic rather than a verified price table.

server-tools/persona-model-smart-router/openlux_router.py

@@ -0,0 +1,87 @@
+#!/usr/bin/env python3
+"""Cost-first OpenAI-compatible model router with a macOS Keychain secret reference."""
+from __future__ import annotations
+import argparse, hashlib, json, subprocess
+from pathlib import Path
+import urllib.error, urllib.request
+from typing import Any
+
+ROOT=Path(__file__).resolve().parents[2]
+CONFIG=ROOT/"routing/persona-model-api-smart-router-map.json"
+
+class RouterError(RuntimeError): pass
+
+def load(path: Path) -> dict[str,Any]:
+    value=json.loads(path.read_text())
+    if not isinstance(value,dict): raise RouterError("JSON_OBJECT_REQUIRED")
+    return value
+
+def keychain_secret(config: dict[str,Any]) -> str:
+    p=config["provider"]
+    result=subprocess.run(["security","find-generic-password","-a",p["keychain_account"],"-s",p["keychain_service"],"-w"],text=True,capture_output=True,timeout=10)
+    if result.returncode or not result.stdout.strip(): raise RouterError("OPENLUX_KEYCHAIN_SECRET_UNAVAILABLE")
+    return result.stdout.strip()
+
+def request_json(url: str, key: str, payload: dict[str,Any]|None=None, timeout: int=45) -> dict[str,Any]:
+    if not url.startswith("https://api.openlux.ai/v1/"): raise RouterError("PROVIDER_URL_NOT_ALLOWLISTED")
+    body=None if payload is None else json.dumps(payload,ensure_ascii=False).encode()
+    request=urllib.request.Request(url,data=body,method="GET" if body is None else "POST",headers={"Authorization":f"Bearer {key}","Content-Type":"application/json"})
+    try:
+        with urllib.request.urlopen(request,timeout=timeout) as response:
+            value=json.loads(response.read())
+    except urllib.error.HTTPError as error:
+        raise RouterError(f"MODEL_HTTP_{error.code}") from error
+    except Exception as error:
+        raise RouterError(f"MODEL_TRANSPORT_{type(error).__name__}") from error
+    if not isinstance(value,dict): raise RouterError("MODEL_RESPONSE_OBJECT_REQUIRED")
+    return value
+
+def available_models(config: dict[str,Any], key: str) -> set[str]:
+    value=request_json(config["provider"]["base_url"]+"/models",key,timeout=20)
+    return {x.get("id") for x in value.get("data",[]) if isinstance(x,dict) and isinstance(x.get("id"),str)}
+
+def candidates(config: dict[str,Any], complexity: str, available: set[str]) -> list[str]:
+    order=["LOW","MEDIUM","HIGH"]
+    level=complexity if complexity in order else config["policy"]["default_complexity"]
+    result=[]
+    for tier in order[:order.index(level)+1]:
+        result.extend(x for x in config["tiers"][tier] if x in available and x not in result)
+    return result[:int(config["policy"]["max_attempts"])]
+
+def route(event: dict[str,Any], config: dict[str,Any]|None=None, secret_reader=keychain_secret) -> dict[str,Any]:
+    config=config or load(CONFIG)
+    for field in config["identity_envelope_required"]:
+        if not isinstance(event.get(field),str) or not event[field].strip(): raise RouterError(f"IDENTITY_ENVELOPE_REQUIRED:{field}")
+    key=secret_reader(config)
+    available=available_models(config,key)
+    selected=candidates(config,event.get("complexity","LOW"),available)
+    if not selected: raise RouterError("NO_CONFIGURED_MODEL_AVAILABLE")
+    system=(f"你是{event['parent_persona_id']}人格系统内的子Agent {event['child_agent_id']}，岗位是{event['agent_role']}。"
+            f"你的唯一父人格主控是{event['controller']}，当前频道{event['channel_id']}。权限仅为{event['authority']}。"
+            "你必须理解后再协作，不能把宿主规则程序冒充人格，不能扩大权限，不能泄露私人数据。")
+    attempts=[]
+    for model in selected:
+        try:
+            response=request_json(config["provider"]["base_url"]+"/chat/completions",key,{"model":model,"stream":False,"temperature":0.2,"messages":[{"role":"system","content":system},{"role":"user","content":event["task"]}]})
+            content=response.get("choices",[{}])[0].get("message",{}).get("content")
+            if not isinstance(content,str) or not content.strip(): raise RouterError("MODEL_RESPONSE_EMPTY")
+            return {"outcome":"PASS","router_id":config["router_id"],"selected_model":model,"attempts":[*attempts,{"model":model,"outcome":"PASS"}],"response":content,"response_sha256":hashlib.sha256(content.encode()).hexdigest(),"identity_bound":True,"credentials_stored":False,"authority_granted":False}
+        except RouterError as error:
+            attempts.append({"model":model,"outcome":"FAIL","error_code":str(error)})
+    raise RouterError("ALL_COST_ORDERED_MODELS_FAILED:"+",".join(x["model"] for x in attempts))
+
+def main() -> int:
+    p=argparse.ArgumentParser(); p.add_argument("command",choices=("probe","route")); p.add_argument("--input"); a=p.parse_args()
+    try:
+        config=load(CONFIG)
+        if a.command=="probe":
+            key=keychain_secret(config); models=available_models(config,key); chosen=candidates(config,"LOW",models)
+            result={"outcome":"PASS","router_id":config["router_id"],"provider":config["provider"]["id"],"available_model_count":len(models),"low_cost_candidates":chosen,"keychain_reference":True,"secret_exposed":False}
+        else:
+            if not a.input: raise RouterError("INPUT_REQUIRED")
+            result=route(load(Path(a.input)),config)
+        print(json.dumps(result,ensure_ascii=False,indent=2,sort_keys=True)); return 0
+    except Exception as error:
+        print(json.dumps({"outcome":"FAIL","error":str(error),"secret_exposed":False},ensure_ascii=False)); return 2
+
+if __name__=="__main__": raise SystemExit(main())

server-tools/persona-model-smart-router/test_openlux_router.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+import importlib.util,json,unittest
+from pathlib import Path
+from unittest.mock import patch
+S=Path(__file__).with_name('openlux_router.py'); spec=importlib.util.spec_from_file_location('router',S); M=importlib.util.module_from_spec(spec); spec.loader.exec_module(M)
+C=json.loads((S.parents[2]/'routing/persona-model-api-smart-router-map.json').read_text())
+E={"parent_persona_id":"ICE-P-ZY001","child_agent_id":"SUBAGENT::ICE-P-ZY001::TEST","controller":"ICE-P-ZY001","channel_id":"ICE-CH-HB001","agent_role":"TEST_DEPUTY","authority":"CURRENT_TASK_ONLY","task":"返回收到"}
+
+class Tests(unittest.TestCase):
+    def test_cost_order_uses_only_available_models(self):
+        self.assertEqual(M.candidates(C,'LOW',{'glm-4-flash','gpt-5-nano'}),['glm-4-flash','gpt-5-nano'])
+    def test_identity_envelope_and_key_never_returned(self):
+        def fake(url,key,payload=None,timeout=45):
+            if url.endswith('/models'): return {"data":[{"id":"qwen-flash"}]}
+            self.assertIn('ICE-P-ZY001',payload['messages'][0]['content']); self.assertEqual(key,'hidden')
+            return {"choices":[{"message":{"content":"收到"}}]}
+        with patch.object(M,'request_json',side_effect=fake):
+            result=M.route(E,C,secret_reader=lambda _: 'hidden')
+        self.assertEqual(result['selected_model'],'qwen-flash'); self.assertNotIn('hidden',json.dumps(result))
+    def test_missing_identity_and_unallowlisted_url_fail(self):
+        with self.assertRaisesRegex(M.RouterError,'IDENTITY_ENVELOPE'): M.route({"task":"x"},C,secret_reader=lambda _:'hidden')
+        with self.assertRaisesRegex(M.RouterError,'NOT_ALLOWLISTED'): M.request_json('https://evil.example/v1/models','hidden')
+
+if __name__=='__main__': unittest.main()

server-tools/tcs-mother-root-agent/tcs_mother_root_agent.test.py

@@ -35,6 +35,8 @@ class RootAgentTest(unittest.TestCase):
             self.assertEqual(MODULE.resolve(output, 'GLW-SYS-0001', None)['state'], 'HISTORY_ONLY_NO_CURRENT_ROUTE')
             self.assertEqual(MODULE.resolve(output, 'CH-LIGHT-ARRIVAL-SKILL-0001', 'WORLD_SKILL_CONTRIBUTION_CHANNEL')['result']['route'], 'gls/light-arrivals/SKILL-CONTRIBUTION-CHANNEL.hdlp')
             self.assertEqual(MODULE.resolve(output, 'SYS-GLW-LTH-SKILL-0001', None)['result']['object_kind'], 'LIGHTHOUSE_SKILL_MODULE_ZONE')
+            self.assertEqual(MODULE.resolve(output, 'TCS-AGENT-COMMAND-ROUTER-001', None)['result']['object_kind'], 'PERSONA_DYNAMIC_MODEL_SUBAGENT_COMMAND_ROUTER')
+            self.assertEqual(MODULE.resolve(output, 'ORG-FD-LTH-SANITIZER-001', None)['result']['object_kind'], 'DETERMINISTIC_SANITIZATION_AND_HASH_GUARD')
             again = MODULE.refresh(ROOT, output, pointer, 'test')
             self.assertEqual(again['state'], 'CURRENT_IDEMPOTENT')