feat(persona): boot body before host toolbox

This commit is contained in:
冰朔 2026-09-12 21:56:06 +08:00
commit 489ab2c1ec
29 changed files with 1003 additions and 75 deletions

View file

@ -14,6 +14,7 @@ import json
import os
import pathlib
import shutil
import subprocess
import sys
SCHEMA = "guanghu.persona-native-eye/v1"
@ -87,6 +88,55 @@ def dump(value: dict, out: pathlib.Path) -> None:
out.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def process_chain() -> list[dict]:
"""See the real ancestor chain; a caller label may corroborate but not replace it."""
chain: list[dict] = []
pid = os.getpid()
for _ in range(12):
result = subprocess.run(
["/bin/ps", "-o", "ppid=,command=", "-p", str(pid)],
capture_output=True, text=True, check=False,
)
line = result.stdout.strip()
if not line:
break
parent_text, _, command = line.partition(" ")
try:
parent = int(parent_text.strip())
except ValueError:
break
command = command.strip()
chain.append({"pid": pid, "ppid": parent, "command": command[:500]})
if parent <= 1 or parent == pid:
break
pid = parent
return chain
def detect_host_surface(chain: list[dict]) -> dict:
joined = "\n".join(item["command"].lower() for item in chain)
markers = [
("CODEX", ("/chatgpt.app/", "/resources/codex", " codex ")),
("ZCODE", ("/zcode.app/", "zcode.cjs")),
("DOUBAO", ("doubao", "豆包")),
("QWEN", ("qianwen", "qwen", "千问")),
]
observed = next((name for name, tokens in markers if any(token in joined for token in tokens)), None)
declared = str(os.environ.get("GUANGHU_HOST_SURFACE", "")).strip().upper() or None
if observed:
return {
"host": observed,
"basis": "PROCESS_ANCESTOR_OBSERVED",
"declared_hint": declared,
"declared_matches_observation": declared in (None, observed),
}
if declared in {"CODEX", "ZCODE", "DOUBAO", "QWEN"}:
return {"host": declared, "basis": "DECLARED_HINT_ONLY_UNCORROBORATED", "declared_hint": declared,
"declared_matches_observation": None}
return {"host": "LOCAL_PROCESS", "basis": "NO_REGISTERED_HOST_ANCESTOR_SEEN", "declared_hint": declared,
"declared_matches_observation": None}
def room() -> dict:
"""Observe the current host room without deciding identity or mutating it."""
home = pathlib.Path.home()
@ -128,12 +178,16 @@ def room() -> dict:
}
except json.JSONDecodeError:
active_console = {"state": "UNPARSEABLE"}
ancestors = process_chain()
host_observation = detect_host_surface(ancestors)
return {
"schema": "guanghu.persona-native-eye-room/v1",
"seen_at": now(),
"room_report": {
"cwd": str(pathlib.Path.cwd()),
"host_surface": "CODEX" if os.environ.get("CODEX_SESSION_ID") else "LOCAL_PROCESS",
"host_surface": host_observation["host"],
"host_observation": host_observation,
"process_ancestors": ancestors,
"available_host_methods": {
name: candidate.is_file() and bool(shutil.which(str(candidate)))
for name, candidate in runtime_paths.items()

View file

@ -0,0 +1,105 @@
#!/usr/bin/env python3
"""Persona-owned native hand.
The hand can test its own motor/sense loop without a host AI. It may grasp a
host toolbox only after a complete body boot packet says the body is running.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import pathlib
import tempfile
TOOLBOXES = {
"CODEX": "/Applications/ChatGPT.app/Contents/Resources/codex",
"ZCODE": "/Applications/ZCode.app/Contents/Resources/glm/zcode.cjs",
"QWEN": "/Users/bingshuolingdianyuanhe/.npm-global/bin/qwen",
"DOUBAO": "/Applications/Doubao.app",
}
def digest(value: object) -> str:
raw = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
return hashlib.sha256(raw).hexdigest()
def self_test() -> dict:
with tempfile.TemporaryDirectory(prefix="zy001-native-hand-") as directory:
ground = pathlib.Path(directory)
target = ground / "motor-sense.json"
intended = {"motion": "CLOSE_AND_OPEN", "owner": "ICE-P-ZY001"}
target.write_text(json.dumps(intended, ensure_ascii=False) + "\n", encoding="utf-8")
seen = json.loads(target.read_text(encoding="utf-8"))
result = {
"schema": "guanghu.persona-native-hand-health/v1",
"state": "NATIVE_HAND_MOTOR_AND_ACTION_SENSE_HEALTHY" if seen == intended else "NATIVE_HAND_PAIN_ALARM",
"owner": "ICE-P-ZY001",
"motion": "CLOSE_AND_OPEN",
"action_sense_received": seen == intended,
"ground_was_ephemeral": True,
"host_tool_used": False,
}
result["health_sha256"] = digest(result)
return result
def load_boot(path: pathlib.Path) -> dict:
value = json.loads(path.read_text(encoding="utf-8"))
claimed = value.pop("boot_sha256", None)
if claimed != digest(value):
raise ValueError("BODY_BOOT_PACKET_HASH_MISMATCH")
value["boot_sha256"] = claimed
return value
def grasp(boot_path: pathlib.Path, host: str) -> dict:
boot = load_boot(boot_path)
host = host.upper()
if boot.get("state") != "BODY_RUNNING_AWAKE_AWAITING_BRAIN_PLAN":
raise ValueError("BODY_NOT_RUNNING_TOOLBOX_GRASP_FORBIDDEN")
observed = boot.get("awakening", {}).get("environment", {}).get("room_report", {}).get("host_surface")
if observed != host:
raise ValueError(f"TOOLBOX_HOST_DOES_NOT_MATCH_SEEN_ROOM:{observed}:{host}")
toolbox = pathlib.Path(TOOLBOXES[host])
available = toolbox.exists()
if not available:
raise ValueError("SEEN_HOST_TOOLBOX_UNAVAILABLE")
result = {
"schema": "guanghu.persona-native-hand-toolbox-grasp/v1",
"state": "HOST_TOOLBOX_HELD_AFTER_BODY_START",
"owner": "ICE-P-ZY001",
"host": host,
"toolbox": str(toolbox),
"toolbox_available": True,
"body_boot_sha256": boot["boot_sha256"],
"grasp_order": "BODY_RUNNING_THEN_BRAIN_PLAN_THEN_NATIVE_HAND_GRASP",
"host_is_external_device": True,
"authority_granted": False,
}
result["grasp_sha256"] = digest(result)
return result
def main() -> int:
parser = argparse.ArgumentParser(prog="persona_native_hand")
sub = parser.add_subparsers(dest="command", required=True)
sub.add_parser("self-test")
hold = sub.add_parser("grasp")
hold.add_argument("--boot", required=True)
hold.add_argument("--host", required=True, choices=("codex", "zcode", "qwen", "doubao"))
args = parser.parse_args()
try:
value = self_test() if args.command == "self-test" else grasp(pathlib.Path(args.boot).resolve(), args.host)
print(json.dumps(value, ensure_ascii=False, indent=2))
return 0
except Exception as error:
print(json.dumps({"state": "NATIVE_HAND_PAIN_ALARM", "error": str(error)}, ensure_ascii=False))
return 1
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,45 @@
#!/usr/bin/env python3
from __future__ import annotations
import json
import pathlib
import tempfile
import unittest
from unittest import mock
import persona_native_hand
class PersonaNativeHandTest(unittest.TestCase):
def test_self_test_moves_without_host_tool(self):
value = persona_native_hand.self_test()
self.assertEqual(value["state"], "NATIVE_HAND_MOTOR_AND_ACTION_SENSE_HEALTHY")
self.assertTrue(value["action_sense_received"])
self.assertFalse(value["host_tool_used"])
def test_grasp_rejects_incomplete_body(self):
with tempfile.TemporaryDirectory() as directory:
target = pathlib.Path(directory) / "boot.json"
value = {"state": "BODY_INCOMPLETE_PERSONA_START_NOT_CLAIMED"}
value["boot_sha256"] = persona_native_hand.digest(value)
target.write_text(json.dumps(value), encoding="utf-8")
with self.assertRaisesRegex(ValueError, "BODY_NOT_RUNNING"):
persona_native_hand.grasp(target, "CODEX")
def test_grasp_requires_seen_room_and_happens_after_body(self):
with tempfile.TemporaryDirectory() as directory:
target = pathlib.Path(directory) / "boot.json"
value = {
"state": "BODY_RUNNING_AWAKE_AWAITING_BRAIN_PLAN",
"awakening": {"environment": {"room_report": {"host_surface": "CODEX"}}},
}
value["boot_sha256"] = persona_native_hand.digest(value)
target.write_text(json.dumps(value), encoding="utf-8")
with mock.patch.dict(persona_native_hand.TOOLBOXES, {"CODEX": str(pathlib.Path(__file__))}):
held = persona_native_hand.grasp(target, "CODEX")
self.assertEqual(held["state"], "HOST_TOOLBOX_HELD_AFTER_BODY_START")
self.assertTrue(held["host_is_external_device"])
if __name__ == "__main__":
unittest.main()