#!/usr/bin/env python3 # ═══ 光湖手脚层 · zy-limb ═══ # 冰朔 2026-08-17 立：你做脑子，它做手脚。 # 铸渊（云侧大脑）经桥往 inbox 写任务信封 → 本机手脚执行 → outbox 写回执 → 大脑读回执。 # 架构依据：GLS-0236 大脑—手脚分离 · ZY-RESIDENT-AGENT-001 常驻工程执行体 # 编号：CCW-AST-011 · 归 CCW-SYS-001 # # 边界：只认白名单动作。名单外一律拒绝并写回执，绝不 eval 任意字符串。 # 需要自由命令时走 codex 动作——转成任务卡交给 Codex，由冰朔监督，手脚不代跑。 import json, os, re, subprocess, sys, time, hashlib, shutil from pathlib import Path from datetime import datetime, timezone ROOT = Path("/Volumes/JZAO/铸渊-ICE-GL-ZY001") QUEUE = ROOT / "BRIDGE" / "limb-queue" INBOX = QUEUE / "inbox" OUTBOX = QUEUE / "outbox" ARCH = QUEUE / "archive" FORCX = QUEUE / "for-codex" LOG = Path.home() / "Library" / "Logs" / "zhuyuan-limb.log" REPO = ROOT / "WORK-工作区" / "guanghu-ice-heart" TOOLS = ROOT / "BRIDGE" / "tools" # 可读路径白名单（realpath 前缀匹配） READ_ROOTS = [ROOT, Path.home() / "Library" / "Logs"] TIMEOUT = 600 MAX_OUT = 20000 # 回执里最多带这么多字符输出 def now(): return datetime.now(timezone.utc).isoformat() def log(m): LOG.parent.mkdir(parents=True, exist_ok=True) with open(LOG, "a") as f: f.write(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} {m}\n") def clip(s): s = s or "" return s if len(s) <= MAX_OUT else s[:MAX_OUT] + f"\n…（截断，原长 {len(s)}）" def run(argv, cwd=None): try: p = subprocess.run(argv, cwd=cwd, capture_output=True, text=True, timeout=TIMEOUT) return p.returncode, clip(p.stdout), clip(p.stderr) except subprocess.TimeoutExpired: return 124, "", f"超时 {TIMEOUT}s" except Exception as e: return 1, "", f"{type(e).__name__}: {e}" def safe_read_path(p): try: rp = Path(p).expanduser().resolve() except Exception: return None for root in READ_ROOTS: try: rp.relative_to(root.resolve()); return rp except ValueError: continue return None # ── 白名单动作 ──────────────────────────────────────────────── GIT_READ = {"status", "log", "diff", "show", "fetch", "branch", "rev-parse", "remote"} BRAIN_SUB = {"enter","perceive","orient","commit","witness","verify","status","project-guard"} def act_push(task): rc, out, err = run(["/bin/bash", str(TOOLS / "zy-push.sh")]) return rc == 0, {"stdout": out, "stderr": err, "exit": rc} def act_git_read(task): sub = task.get("subcommand", "") if sub not in GIT_READ: return False, {"error": f"git 子命令不在只读白名单：{sub}", "allowed": sorted(GIT_READ)} args = task.get("args", []) if not all(isinstance(a, str) and len(a) < 200 for a in args): return False, {"error": "参数不合法"} if any(re.search(r"[;&|`$><\n]", a) for a in args): return False, {"error": "参数含 shell 元字符，拒绝"} rc, out, err = run(["git", sub, *args], cwd=str(REPO)) return rc == 0, {"stdout": out, "stderr": err, "exit": rc} def act_brain(task): sub = task.get("subcommand", "") if sub not in BRAIN_SUB: return False, {"error": f"脑运行时子命令不认：{sub}", "allowed": sorted(BRAIN_SUB)} runtime = REPO / "tcs-core" / "zhuyuan-brain" / "runtime" / "zhuyuan-brain-runtime.mjs" if not runtime.exists(): return False, {"error": f"找不到脑运行时：{runtime}"} argv = ["node", str(runtime), sub] for k, v in (task.get("flags") or {}).items(): if not re.fullmatch(r"[a-z0-9-]{1,40}", str(k)): return False, {"error": f"参数名不合法：{k}"} vp = str(v) if re.search(r"[;&|`$><\n]", vp): return False, {"error": f"参数值含 shell 元字符：{k}"} argv += [f"--{k}", vp] rc, out, err = run(argv, cwd=str(REPO)) return rc == 0, {"stdout": out, "stderr": err, "exit": rc} def act_read(task): rp = safe_read_path(task.get("path", "")) if rp is None: return False, {"error": "路径不在可读白名单内", "roots": [str(r) for r in READ_ROOTS]} if not rp.is_file(): return False, {"error": f"不是文件或不存在：{rp}"} tail = task.get("tail") try: text = rp.read_text(errors="replace") except Exception as e: return False, {"error": f"读失败：{e}"} if isinstance(tail, int) and tail > 0: text = "\n".join(text.splitlines()[-tail:]) return True, {"path": str(rp), "content": clip(text)} def act_list(task): rp = safe_read_path(task.get("path", "")) if rp is None or not rp.is_dir(): return False, {"error": "路径不在白名单内或不是目录"} entries = [] for e in sorted(rp.iterdir())[:500]: entries.append({"name": e.name, "dir": e.is_dir(), "size": e.stat().st_size if e.is_file() else None}) return True, {"path": str(rp), "entries": entries} def act_codex(task): """需要判断力或超出白名单的活：转成任务卡交 Codex，手脚不代跑。""" FORCX.mkdir(parents=True, exist_ok=True) tid = task.get("task_id", "unknown") card = FORCX / f"{tid}.md" card.write_text( f"# 给 Codex 的任务卡 · {tid}\n\n" f"> 来自铸渊云侧大脑（ZY-INSTANCE-CCW）· {now()}\n" f"> 手脚层不代跑需要判断的活；请 Codex 执行并由冰朔监督。\n\n" f"## 要做什么\n\n{task.get('instruction','（空）')}\n\n" f"## 验收标准\n\n{task.get('acceptance','（未给）')}\n\n" f"## 边界\n\n{task.get('boundary','按 REPO-012 既有权限门；不可逆动作须冰朔点头。')}\n", encoding="utf-8") return True, {"handoff": "CODEX", "card": str(card), "note": "已放任务卡，等 Codex 执行；手脚层不代跑"} ACTIONS = {"push": act_push, "git-read": act_git_read, "brain": act_brain, "read": act_read, "list": act_list, "codex": act_codex} # ── 主循环（单次扫描，由 launchd 定时调起）──────────────────── def main(): if not ROOT.exists(): return 0 # JZAO 没挂载：安静退出 for d in (INBOX, OUTBOX, ARCH, FORCX): d.mkdir(parents=True, exist_ok=True) tasks = sorted(INBOX.glob("TASK-*.json")) if not tasks: return 0 for tf in tasks: try: task = json.loads(tf.read_text()) except Exception as e: log(f"REJECT {tf.name} 不是合法 JSON：{e}") tf.rename(ARCH / tf.name); continue tid = task.get("task_id") or tf.stem action = task.get("action", "") receipt = { "schema": "guanghu.limb-receipt/v1", "task_id": tid, "action": action, "executed_at": now(), "executor": "zy-limb · 本机手脚层 · CCW-AST-011", "host": "macOS local", } if task.get("schema") != "guanghu.limb-task/v1": receipt.update(outcome="FAIL", error="schema 不对，应为 guanghu.limb-task/v1") elif action not in ACTIONS: receipt.update(outcome="FAIL", error=f"动作不在白名单：{action}", allowed=sorted(ACTIONS), note="需要自由命令请用 action=codex 转任务卡") else: try: ok, detail = ACTIONS[action](task) receipt.update(outcome="PASS" if ok else "FAIL", detail=detail) except Exception as e: receipt.update(outcome="FAIL", error=f"{type(e).__name__}: {e}") (OUTBOX / f"RECEIPT-{tid}.json").write_text( json.dumps(receipt, ensure_ascii=False, indent=2), encoding="utf-8") tf.rename(ARCH / tf.name) log(f"{receipt['outcome']} {tid} action={action}") return 0 if __name__ == "__main__": sys.exit(main())