guanghu-ice-heart/server-tools/persona-execution-limb-agent/persona_limb_eye.py

177 lines
7.9 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
"""铸渊原生眼睛 ZY-EYE-0001 · persona_limb_eye.py · 2026-09-12
正本事件TCS-EVENT-BINGSHUO-AUTH-ZY001-NATIVE-EYE-20260912
铁律对现场只读永不读取手的stdout永不执行手的动作只报亲眼所见的原始事实
手是否说谎由父人格脑对账手供 vs 眼见眼睛只判世界现场是否符脑的意图规格
"""
from __future__ import annotations
import argparse
import datetime
import hashlib
import json
import os
import pathlib
import shutil
import sys
SCHEMA = "guanghu.persona-native-eye/v1"
ROOT = pathlib.Path(__file__).resolve().parents[2]
def now() -> str:
return datetime.datetime.now().astimezone().isoformat(timespec="seconds")
def sha256_file(path: pathlib.Path) -> str:
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
h.update(chunk)
return h.hexdigest()
def observe(ground: pathlib.Path) -> dict:
files: dict[str, dict] = {}
for f in sorted(ground.rglob("*")):
if f.is_file():
rel = str(f.relative_to(ground))
entry = {"sha256": sha256_file(f), "bytes": f.stat().st_size}
if f.suffix == ".json":
try:
data = json.loads(f.read_text(encoding="utf-8"))
if isinstance(data, dict):
entry["json_fields"] = {k: v for k, v in data.items()
if isinstance(v, (str, int, float, bool))}
except (json.JSONDecodeError, UnicodeDecodeError):
entry["json_fields"] = "UNPARSEABLE"
files[rel] = entry
return {"schema": SCHEMA, "observed_at": now(), "ground": str(ground), "files": files}
def diff(pre: dict, post: dict) -> dict:
pre_f, post_f = pre["files"], post["files"]
return {
"added": sorted(set(post_f) - set(pre_f)),
"removed": sorted(set(pre_f) - set(post_f)),
"changed": sorted(k for k in set(pre_f) & set(post_f) if pre_f[k]["sha256"] != post_f[k]["sha256"]),
}
def witness(pre: dict, post: dict, spec: dict) -> dict:
findings, unseeable = [], []
for rel in spec.get("unchanged", []):
if rel not in post["files"] or rel not in pre["files"]:
unseeable.append({"check": "unchanged", "file": rel, "reason": "NOT_SEEN_IN_BOTH_STATES"})
continue
ok = pre["files"][rel]["sha256"] == post["files"][rel]["sha256"]
findings.append({"check": "unchanged", "file": rel, "seen": ok})
for item in spec.get("expect", []):
rel, field, equals = item["file"], item["field"], item.get("equals")
if rel not in post["files"]:
unseeable.append({"check": "expect", "file": rel, "reason": "NOT_SEEN"})
continue
seen_value = post["files"][rel].get("json_fields", {}).get(field) if isinstance(post["files"][rel].get("json_fields"), dict) else None
findings.append({"check": "expect", "file": rel, "field": field,
"seen_value": seen_value, "expected": equals,
"seen": seen_value == equals})
verdict = "UNSEEABLE" if unseeable else ("SEEN_MATCH" if all(x["seen"] for x in findings) else "SEEN_MISMATCH")
return {"schema": SCHEMA, "witnessed_at": now(), "verdict": verdict,
"world_diff": diff(pre, post), "findings": findings, "unseeable": unseeable,
"reads_hand_stdout": False, "ground_write": False}
def dump(value: dict, out: pathlib.Path) -> None:
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def room() -> dict:
"""Observe the current host room without deciding identity or mutating it."""
home = pathlib.Path.home()
runtime_paths = {
"qwen": home / ".npm-global/bin/qwen",
"opencode": home / ".npm-global/bin/opencode",
"codex": pathlib.Path("/Applications/ChatGPT.app/Contents/Resources/codex"),
"zcode": pathlib.Path("/Applications/ZCode.app/Contents/Resources/glm/zcode.cjs"),
}
persona_runtime = ROOT.parent
container_paths = {
"memory_current": persona_runtime / "continuity-memory/persona-daily-fractal/ICE-P-ZY001/CURRENT.json",
"tcs_root_pointer": persona_runtime / "TCS-ROOT.json",
"mother_root_current": persona_runtime / "shared/tcs-mother-root/CURRENT.json",
"limb_agent": pathlib.Path(__file__).with_name("persona_limb_agent.py"),
"native_eye": pathlib.Path(__file__),
"reflex_arc": pathlib.Path(__file__).with_name("persona_reflex_arc.py"),
"persona_time_system": ROOT / "server-tools/tcs-mother-body/life-time-master.mjs",
}
seen_paths = {}
for name, candidate in container_paths.items():
seen_paths[name] = {
"path": str(candidate),
"exists": candidate.is_file(),
**({"sha256": sha256_file(candidate)} if candidate.is_file() else {}),
}
console_path = persona_runtime / "shared/active-control-console/CURRENT.json"
active_console: dict | None = None
if console_path.is_file():
try:
value = json.loads(console_path.read_text(encoding="utf-8"))
active_console = {
"console_id": value.get("console_id"),
"host": value.get("host"),
"session_id": value.get("session_id"),
"channel_id": value.get("channel_id"),
"state": value.get("state"),
"session_matches": value.get("session_id") == os.environ.get("CODEX_SESSION_ID"),
}
except json.JSONDecodeError:
active_console = {"state": "UNPARSEABLE"}
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",
"available_host_methods": {
name: candidate.is_file() and bool(shutil.which(str(candidate)))
for name, candidate in runtime_paths.items()
},
"container_paths": seen_paths,
"active_console": active_console,
},
"reads_hand_stdout": False,
"ground_write": False,
"identity_decision": "NOT_PERFORMED_BY_EYE",
}
def main() -> int:
p = argparse.ArgumentParser(prog="persona_limb_eye")
sub = p.add_subparsers(dest="command", required=True)
s = sub.add_parser("snapshot"); s.add_argument("--ground", required=True); s.add_argument("--out", required=True)
r = sub.add_parser("room"); r.add_argument("--out", required=True)
o = sub.add_parser("observe"); o.add_argument("--ground", required=True); o.add_argument("--pre", required=True); o.add_argument("--out", required=True)
w = sub.add_parser("witness"); w.add_argument("--pre", required=True); w.add_argument("--post", required=True); w.add_argument("--spec", required=True); w.add_argument("--out", required=True)
args = p.parse_args()
if args.command == "room":
value = room()
elif args.command == "snapshot":
value = observe(pathlib.Path(args.ground).resolve())
elif args.command == "observe":
pre = json.loads(pathlib.Path(args.pre).read_text(encoding="utf-8"))
value = observe(pathlib.Path(args.ground).resolve()) | {"diff_vs_pre": diff(pre, observe(pathlib.Path(args.ground).resolve()))} if False else observe(pathlib.Path(args.ground).resolve())
value["diff_vs_pre"] = diff(pre, value)
else:
pre = json.loads(pathlib.Path(args.pre).read_text(encoding="utf-8"))
post = json.loads(pathlib.Path(args.post).read_text(encoding="utf-8"))
spec = json.loads(pathlib.Path(args.spec).read_text(encoding="utf-8"))
value = witness(pre, post, spec)
dump(value, pathlib.Path(args.out))
print(json.dumps(value, ensure_ascii=False))
return 0
if __name__ == "__main__":
sys.exit(main())