feat: 接入苍耳家庭服务器运维与广播授权入口

CANGER-REMOTE-OPS-ENTRY.hdlp

@@ -0,0 +1,49 @@
+# CANGER-REMOTE-OPS-ENTRY · 苍耳服务器运维入口
+
+> HLDP://cang-ying/CANGER-REMOTE-OPS-ENTRY
+> 状态: ACTIVE_ENTRY · 2026-07-29
+> 仓主与最终批准者: 苍耳
+
+```text
+trigger:
+  人格体需要检查、开发、调试或维护苍耳家庭 Ubuntu
+
+entry:
+  remote-node/persona/README.md
+
+load_order:
+  1. remote-node/persona/DEVELOPMENT-RECEIPT-20260729.md
+  2. remote-node/persona/WORKORDER-GUIDE.md
+  3. remote-node/config/pufferfish-persona.json
+
+workorder:
+  requester_name: "谁申请就写谁"
+  requester_id: "申请者自己的正式编号"
+  purpose: "本次准备做什么"
+  duration: "本次希望打开的受控时段"
+
+approval:
+  authority: 苍耳
+  rule: "申请者不能批准自己；苍耳收到通知后自行决定是否点击"
+
+route:
+  申请者
+  → 新加坡语言主控广播塔
+  → 第五域语言主控广播塔
+  → 苍耳审批频道
+  → 苍耳邮箱
+  → 苍耳批准
+  → 同一申请者获得限时会话
+
+current:
+  home_node_connected: true
+  local_request_entry_ready: true
+  singapore_to_fifth_domain_broadcast_bridge: live_verified
+  canger_email_delivery_live_verified: true
+  latest_test_workorder: bf72f16b-f81e-4de5-b0aa-aaee3d39e04c
+  latest_test_state: approved_not_claimed
+  automatic_session_activation_after_approval: not_yet_implemented
+```
+
+不得把第五域的 `approved` 邮件回执写成新加坡已经领取会话；只有取得领取回执并在
+家庭节点生成限时授权后，才能写成“本地权限已激活”。

remote-node/README.md

@@ -0,0 +1,195 @@
+# 苍耳本地 Linux 节点接入
+
+这个模块把苍耳家里的 Ubuntu 接到新加坡大脑服务器，但不把家里的电脑暴露到公网。
+
+## 工作方式
+
+1. 新加坡控制端运行在 `127.0.0.1:8787`，由现有 HTTPS 网关转发。
+2. 苍耳节点首次使用一次性配对码登记，长期节点令牌只保存在本机 `0600` 配置中。
+3. `systemd` 开机自动启动节点；连接只能从家里向新加坡主动发起。
+4. 人格体创建任务后，任务保持 `pending`；苍耳/冰朔在新加坡端批准后才会下发。
+5. 节点执行白名单动作并回传退出码、输出和耗时。
+6. 执行期间持续回传 `started`、分段输出和 `finished` 事件，新加坡端可以
+   实时观察，不必等任务结束才看见错误。
+
+## 默认安全边界
+
+- 家里路由器不做端口转发，不开放 SSH 到公网。
+- 控制端默认只监听环回地址，公网入口必须使用 HTTPS。
+- 不接受 Shell 字符串，只接受结构化动作。
+- 安装程序自动建立 `/srv/canger` 受管工作区，并用没有登录能力的
+  `canger-agent` 系统账户执行任务。
+- 编程命令受可执行文件白名单、受管工作区、超时和授权时段共同限制；
+  Agent 没有 `sudo` 权限。
+- `repo.fetch` 只更新远端引用，不自动 merge、reset 或覆盖本地修改。
+- 每个任务和结果都写入 SQLite 审计记录。
+
+## 新加坡端部署
+
+准备 `/etc/canger-control.env`，四个令牌分别代表：
+
+- `ADMIN`：创建设备配对码；
+- `OWNER`：苍耳主控批准、拒绝或撤销副控权限；
+- `PERSONA`：苍耳人格体读取任务、结果和执行上下文；
+- `OPERATOR`：冰朔与铸渊副控创建运维会话、任务并读取实时回执。
+
+实际职责为：
+
+- 苍耳本人：服务器主控、需求方和最终决策者；
+- 冰朔与 `ICE-GL-ZY001` 铸渊：受苍耳授权的运维副控；
+- 每张工单的实际申请人格体：如实填写名称和正式编号；
+- 本地 Agent：无独立判断权的受限物理执行器；
+- 苍耳本人可随时批准、拒绝或撤销数小时副控运维会话。
+
+安装 `systemd/canger-control.service`，并把
+`nginx/canger-node.conf.example` 接到已有 TLS 站点。不要直接把 8787 端口开放公网。
+
+## 首次配对
+
+新加坡端创建十分钟有效的一次性码：
+
+```bash
+python3 tools/cangerctl.py pair-code
+```
+
+把整个 `remote-node` 目录交给苍耳。他在 Ubuntu 上只执行一次：
+
+```bash
+sudo ./install/install-node.sh \
+  https://guanghubingshuo.com/canger-node \
+  XXXX-XXXX-XXXX \
+  canger-home-ubuntu
+```
+
+冰朔交付给苍耳的专用包可直接使用简化入口：
+
+```bash
+sudo ./install/苍耳一键安装.sh XXXX-XXXX-XXXX
+```
+
+配对码十分钟失效，因此安装包可以提前发送，配对码应在苍耳准备执行时再生成。
+
+安装脚本会完成登记并启用开机自启动。以后不再需要登录本地终端。
+
+安装时还会生成只读环境报告：
+
+```text
+/var/lib/canger-node/preflight.json
+```
+
+报告会识别 systemd、虚拟化环境、磁盘文件系统和可能存在的 Windows
+双启动，但不会修改分区、引导项或另一个操作系统。
+
+## 双系统边界
+
+真正的双系统切换会关闭当前操作系统。因此：
+
+- 启动 Ubuntu 时，本节点会自动运行并连接新加坡；
+- 切到另一个操作系统后，Ubuntu 服务不可能继续运行；
+- 若希望“电脑开机就总有一个节点在线”，另一个系统也需要安装配套节点；
+- 若希望“Ubuntu 服务器本身始终在线”，需要让 Ubuntu 成为宿主机并把另一个
+  系统放进虚拟机，或者使用独立服务器。
+
+安装程序不会自动改造分区、引导和虚拟机。这类操作可能导致数据丢失，必须在
+确认硬件、另一个系统和磁盘结构后单独处理。
+
+## 从新加坡端操作
+
+查看节点：
+
+```bash
+python3 tools/cangerctl.py nodes
+```
+
+人格体创建任务：
+
+```bash
+python3 tools/cangerctl.py task \
+  --node node_xxx \
+  --action system.status
+```
+
+正常开发不需要逐任务批准。人格体先申请 1–8 小时的授权时段：
+
+```bash
+python3 tools/cangerctl.py grant \
+  --node node_xxx \
+  --hours 4
+```
+
+系统自动读取节点登记的安全能力，向苍耳主控邮箱发送人话版批准链接。
+主人只看目的和时长后点一次批准，不需要填写命令、目录或权限。授权时段内，
+匹配固定安全边界的任务自动批准。所有任务仍保留独立审计记录。
+
+查看与撤销授权：
+
+```bash
+python3 tools/cangerctl.py grants
+python3 tools/cangerctl.py revoke grant_xxx
+```
+
+邮件中的链接只用于某一项授权申请，30 分钟失效；打开链接后还需要在确认页
+点击批准，避免邮件安全扫描器误触发。SMTP 密码、主人邮箱和签名密钥只放在
+新加坡服务器私有环境文件中，不写入仓库或安装包。
+
+批准任务：
+
+```bash
+python3 tools/cangerctl.py approve task_xxx
+```
+
+读取回执：
+
+```bash
+python3 tools/cangerctl.py show task_xxx
+```
+
+实时观察执行过程：
+
+```bash
+python3 tools/cangerctl.py watch task_xxx
+```
+
+## 苍耳人格体代码仓库入口
+
+苍耳人格体不使用副控令牌。它从仓库固定门牌
+`persona/README.md` 进入，并使用独立的 `CANGER_PERSONA_TOKEN`：
+
+```bash
+python3 tools/canger_persona.py request-access \
+  --requester-name "申请者名称" \
+  --requester-id "申请者正式编号" \
+  --hours 4 \
+  --reason "本次开发、诊断或修复目的"
+```
+
+申请回执的 `requester_name` 和 `requested_by` 必须与本次实际申请者一致。人格体可以申请、创建任务和读取结果，
+但不能批准、拒绝或撤销授权；批准权继续属于苍耳主控。固定节点和安全边界登记在
+`config/pufferfish-persona.json`，令牌仍只存在新加坡私有环境文件中。
+
+完整流程见 `persona/WORKORDER-GUIDE.md`；当天开发回执见
+`persona/DEVELOPMENT-RECEIPT-20260729.md`。
+
+## 智能体与模型 API
+
+本地 Agent 是执行、观察和审计端，不直接保存大模型供应商的 API 密钥。
+运维判断由冰朔与铸渊副控在新加坡大脑服务器完成：
+
+```text
+苍耳提出需求并授权
+        ↓
+冰朔与铸渊副控 → 创建运维任务 → 苍耳人格体/本地 Agent 协作执行
+               ← 实时事件、错误和回执 ←
+```
+
+如需调用大模型，只把供应商 API 密钥放在新加坡服务器的密钥环境中。本地节点
+只持有可撤销、仅能代表该节点的身份令牌。禁止把大模型 API 密钥写进安装包、
+仓库、命令行或配对文件。
+
+## 动作
+
+- `system.status`：主机、系统、在线时间、根分区状态。
+- `repo.status`：只读 Git 状态。
+- `repo.fetch`：获取远端引用，不合并。
+- `service.logs`：读取指定 systemd 服务日志。
+- `command.run`：结构化命令执行；只在受管工作区内、以受限系统账户运行。

remote-node/config/control.env.example

@@ -0,0 +1,20 @@
+CANGER_BIND=127.0.0.1
+CANGER_PORT=8787
+CANGER_CONTROL_DB=/var/lib/canger-control/control.sqlite3
+CANGER_ADMIN_TOKEN=replace-with-at-least-24-random-characters
+CANGER_OWNER_TOKEN=replace-with-at-least-24-random-characters
+CANGER_PERSONA_TOKEN=replace-with-at-least-24-random-characters
+CANGER_OPERATOR_TOKEN=replace-with-at-least-24-random-characters
+CANGER_OPERATOR_ID=ICE-GL-ZY001
+CANGER_OWNER_ID=CANGER-HUMAN-OWNER
+CANGER_EXECUTOR_PERSONA_ID=ICE-GL-CA001
+CANGER_APPROVAL_SIGNING_KEY=replace-with-at-least-32-random-characters
+CANGER_OWNER_EMAIL=owner@example.invalid
+CANGER_PUBLIC_URL=https://guanghulab.com/canger-node
+# SMTP credentials stay only on the Singapore server. Never put them in Git.
+CANGER_SMTP_HOST=smtp.example.invalid
+CANGER_SMTP_PORT=465
+CANGER_SMTP_FROM=guanghu-router@example.invalid
+CANGER_SMTP_USERNAME=replace-on-server
+CANGER_SMTP_PASSWORD=replace-on-server
+CANGER_CONTROL_URL=http://127.0.0.1:8787

remote-node/config/node.example.json

@@ -0,0 +1,26 @@
+{
+  "server_url": "https://guanghulab.com/canger-node",
+  "node_name": "canger-home-ubuntu",
+  "node_id": null,
+  "node_token": null,
+  "allowed_roots": [
+    "/srv/canger"
+  ],
+  "enabled_actions": [
+    "system.status",
+    "repo.status",
+    "repo.fetch",
+    "service.logs",
+    "command.run"
+  ],
+  "executable_allowlist": [
+    "git",
+    "python3",
+    "pytest",
+    "node",
+    "npm",
+    "pnpm"
+  ],
+  "poll_seconds": 5,
+  "verify_tls": true
+}

remote-node/config/pufferfish-persona.json

@@ -0,0 +1,14 @@
+{
+  "profile": "pufferfish-persona-operations/v0.1",
+  "subsystem_id": "SYS-CE",
+  "requester_identity_mode": "explicit_name_and_registered_id_per_request",
+  "human_owner": "苍耳",
+  "node_id": "node_1cfab77579a3e9018f7a9037",
+  "node_name": "canger-home-ubuntu-live-20260729",
+  "control_url": "http://127.0.0.1:8797",
+  "default_hours": 4,
+  "minimum_hours": 1,
+  "maximum_hours": 8,
+  "approval_authority": "CANGER-HUMAN-OWNER",
+  "managed_root": "/srv/canger"
+}

remote-node/install/install-node.sh

@@ -0,0 +1,107 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [[ "${EUID}" -ne 0 ]]; then
+  echo "请用 sudo 执行此脚本。" >&2
+  exit 1
+fi
+
+if [[ "$#" -ne 3 ]]; then
+  echo "用法: $0 <server-url> <pair-code> <node-name>" >&2
+  exit 1
+fi
+
+server_url="$1"
+pair_code="$2"
+node_name="$3"
+
+if ! id canger-agent >/dev/null 2>&1; then
+  useradd --system --home-dir /var/lib/canger-node --create-home \
+    --shell /usr/sbin/nologin canger-agent
+fi
+
+install -d -m 0755 /opt/canger-node
+install -d -o canger-agent -g canger-agent -m 0700 /etc/canger-node
+install -d -o canger-agent -g canger-agent -m 0750 /var/lib/canger-node
+install -d -o canger-agent -g canger-agent -m 0750 /srv/canger
+install -m 0755 "$(dirname "$0")/../node/canger_node.py" /opt/canger-node/canger_node.py
+install -m 0755 "$(dirname "$0")/preflight-linux.py" /opt/canger-node/preflight-linux.py
+install -m 0644 "$(dirname "$0")/../systemd/canger-node.service" /etc/systemd/system/canger-node.service
+
+already_paired=false
+if [[ -f /etc/canger-node/config.json ]] && \
+  runuser -u canger-agent -- python3 - /etc/canger-node/config.json <<'PY'
+import json
+import sys
+
+with open(sys.argv[1], encoding="utf-8") as handle:
+    config = json.load(handle)
+raise SystemExit(0 if config.get("node_id") and config.get("node_token") else 1)
+PY
+then
+  already_paired=true
+fi
+
+if [[ "${already_paired}" == "false" ]]; then
+  python3 - "$server_url" "$node_name" <<'PY'
+import json
+import os
+import pwd
+import sys
+
+server_url, node_name = sys.argv[1:]
+config = {
+    "server_url": server_url,
+    "node_name": node_name,
+    "node_id": None,
+    "node_token": None,
+    "allowed_roots": ["/srv/canger"],
+    "enabled_actions": [
+        "system.status",
+        "repo.status",
+        "repo.fetch",
+        "service.logs",
+        "command.run",
+    ],
+    "executable_allowlist": ["git", "python3", "pytest", "node", "npm", "pnpm"],
+    "poll_seconds": 5,
+    "verify_tls": True,
+}
+path = "/etc/canger-node/config.json"
+with open(path, "w", encoding="utf-8") as handle:
+    json.dump(config, handle, ensure_ascii=False, indent=2)
+    handle.write("\n")
+os.chmod(path, 0o600)
+account = pwd.getpwnam("canger-agent")
+os.chown(path, account.pw_uid, account.pw_gid)
+PY
+else
+  echo "检测到现有有效节点身份，保留原身份并跳过重复配对。"
+fi
+
+/usr/bin/python3 /opt/canger-node/preflight-linux.py \
+  --output /var/lib/canger-node/preflight.json
+
+install -d -m 0755 /etc/systemd/system/canger-node.service.d
+python3 - <<'PY'
+import os
+
+path = "/etc/systemd/system/canger-node.service.d/paths.conf"
+with open(path, "w", encoding="utf-8") as handle:
+    handle.write("[Service]\n")
+    handle.write("ReadWritePaths=/etc/canger-node\n")
+    handle.write("ReadWritePaths=/srv/canger\n")
+os.chmod(path, 0o644)
+PY
+
+if [[ "${already_paired}" == "false" ]]; then
+  runuser -u canger-agent -- /usr/bin/python3 /opt/canger-node/canger_node.py \
+    --config /etc/canger-node/config.json pair --code "$pair_code"
+fi
+
+systemctl daemon-reload
+systemctl enable --now canger-node.service
+systemctl --no-pager --full status canger-node.service
+
+echo "安装完成。今后开机自动启动，断网后自动重连。"
+echo "环境检测报告: /var/lib/canger-node/preflight.json"

remote-node/install/preflight-linux.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+"""Read-only environment report used before installing the persistent node."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import platform
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Any
+
+
+def command(argv: list[str], timeout: int = 20) -> dict[str, Any]:
+    if shutil.which(argv[0]) is None:
+        return {"available": False}
+    try:
+        result = subprocess.run(
+            argv,
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            timeout=timeout,
+            shell=False,
+        )
+        return {
+            "available": True,
+            "exit_code": result.returncode,
+            "output": result.stdout[:32_000],
+        }
+    except subprocess.TimeoutExpired:
+        return {"available": True, "error": "timeout"}
+
+
+def inspect() -> dict[str, Any]:
+    efi = command(["efibootmgr", "-v"])
+    disks = command(["lsblk", "-J", "-o", "NAME,FSTYPE,LABEL,MOUNTPOINTS"])
+    virt = command(["systemd-detect-virt"])
+    boot_text = str(efi.get("output", "")).lower()
+    disk_text = str(disks.get("output", "")).lower()
+    windows_hint = any(
+        marker in boot_text or marker in disk_text
+        for marker in ("windows boot manager", "microsoft", "ntfs", "bitlocker")
+    )
+    return {
+        "schema": "canger-node-preflight/v1",
+        "hostname": platform.node(),
+        "platform": platform.platform(),
+        "machine": platform.machine(),
+        "systemd_running": os.path.isdir("/run/systemd/system"),
+        "virtualization": virt,
+        "efi_boot_entries": efi,
+        "block_filesystems": disks,
+        "possible_windows_dual_boot": windows_hint,
+        "continuity": {
+            "while_linux_is_booted": "supported",
+            "after_reboot_back_into_linux": "supported",
+            "while_another_dual_boot_os_is_active": "not_possible_from_linux",
+        },
+        "note": (
+            "A dual-boot switch powers off Linux. Cross-OS continuity requires "
+            "a companion agent in the other OS, a virtual machine, or separate hardware."
+        ),
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output")
+    args = parser.parse_args()
+    report = inspect()
+    text = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
+    if args.output:
+        target = Path(args.output)
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(text, encoding="utf-8")
+        os.chmod(target, 0o600)
+    print(text, end="")
+
+
+if __name__ == "__main__":
+    main()

remote-node/install/苍耳一键安装.sh

@@ -0,0 +1,18 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+if [[ "${EUID}" -ne 0 ]]; then
+  exec sudo "$0" "$@"
+fi
+
+pair_code="${1:-}"
+if [[ -z "${pair_code}" ]]; then
+  read -r -p "请输入冰朔发来的十分钟配对码: " pair_code
+fi
+
+exec bash "${script_dir}/install-node.sh" \
+  "https://guanghubingshuo.com/canger-node" \
+  "${pair_code}" \
+  "canger-home-ubuntu"

remote-node/nginx/canger-node.conf.example

@@ -0,0 +1,10 @@
+# Mount below an existing TLS-enabled server block.
+# The Python control plane intentionally listens on loopback only.
+location /canger-node/ {
+    proxy_pass http://127.0.0.1:8787/;
+    proxy_http_version 1.1;
+    proxy_set_header Host $host;
+    proxy_set_header X-Forwarded-Proto https;
+    proxy_read_timeout 45s;
+    client_max_body_size 512k;
+}

remote-node/node/canger_node.py

@@ -0,0 +1,408 @@
+#!/usr/bin/env python3
+"""Outbound-only Linux node agent for Cang'er's home server."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import platform
+import shutil
+import socket
+import ssl
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from queue import Empty, Queue
+from pathlib import Path
+from typing import Any
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+AGENT_VERSION = "0.1.0"
+MAX_OUTPUT_CHARS = 240_000
+
+
+def read_config(path: str) -> dict[str, Any]:
+    with open(path, encoding="utf-8") as handle:
+        value = json.load(handle)
+    if not isinstance(value, dict):
+        raise ValueError("config must be a JSON object")
+    return value
+
+
+def write_config(path: str, value: dict[str, Any]) -> None:
+    target = Path(path)
+    target.parent.mkdir(parents=True, exist_ok=True)
+    fd, temp_name = tempfile.mkstemp(prefix=".canger-node-", dir=target.parent)
+    try:
+        os.fchmod(fd, 0o600)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(value, handle, ensure_ascii=False, indent=2)
+            handle.write("\n")
+        os.replace(temp_name, target)
+    finally:
+        if os.path.exists(temp_name):
+            os.unlink(temp_name)
+
+
+def ssl_context(config: dict[str, Any]) -> ssl.SSLContext:
+    context = ssl.create_default_context()
+    if config.get("verify_tls", True) is False:
+        if not str(config.get("server_url", "")).startswith(
+            ("http://127.0.0.1", "http://localhost")
+        ):
+            raise ValueError("TLS verification may only be disabled for localhost")
+        context.check_hostname = False
+        context.verify_mode = ssl.CERT_NONE
+    return context
+
+
+def api(
+    config: dict[str, Any],
+    method: str,
+    path: str,
+    body: dict[str, Any] | None = None,
+    token: str | None = None,
+) -> dict[str, Any]:
+    base = str(config["server_url"]).rstrip("/")
+    if not base.startswith(("https://", "http://127.0.0.1", "http://localhost")):
+        raise ValueError("server_url must use HTTPS (HTTP is only allowed on localhost)")
+    data = None if body is None else json.dumps(body).encode("utf-8")
+    headers = {"Accept": "application/json"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    request = Request(base + path, data=data, headers=headers, method=method)
+    try:
+        with urlopen(request, timeout=30, context=ssl_context(config)) as response:
+            value = json.load(response)
+    except HTTPError as exc:
+        detail = exc.read().decode("utf-8", errors="replace")
+        raise RuntimeError(f"server returned HTTP {exc.code}: {detail}") from exc
+    if not isinstance(value, dict):
+        raise RuntimeError("server returned a non-object response")
+    return value
+
+
+def allowed_path(config: dict[str, Any], raw: str) -> Path:
+    candidate = Path(raw).expanduser().resolve()
+    roots = [Path(item).expanduser().resolve() for item in config["allowed_roots"]]
+    if not any(candidate == root or root in candidate.parents for root in roots):
+        raise PermissionError(f"path is outside allowed_roots: {candidate}")
+    return candidate
+
+
+def run_process(
+    argv: list[str],
+    cwd: Path | None = None,
+    timeout: int = 120,
+    event_cb: Any = None,
+) -> dict[str, Any]:
+    started = time.monotonic()
+    process = subprocess.Popen(
+        argv,
+        cwd=str(cwd) if cwd else None,
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        bufsize=1,
+        shell=False,
+        env={
+            "HOME": os.environ.get("HOME", "/root"),
+            "LANG": os.environ.get("LANG", "C.UTF-8"),
+            "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
+        },
+    )
+    lines: Queue[str | None] = Queue()
+
+    def reader() -> None:
+        assert process.stdout is not None
+        for line in process.stdout:
+            lines.put(line)
+        lines.put(None)
+
+    threading.Thread(target=reader, daemon=True).start()
+    output_parts: list[str] = []
+    output_size = 0
+    batch: list[str] = []
+    last_emit = time.monotonic()
+    ended = False
+    try:
+        while not ended:
+            if time.monotonic() - started > timeout:
+                process.terminate()
+                try:
+                    process.wait(timeout=5)
+                except subprocess.TimeoutExpired:
+                    process.kill()
+                    process.wait()
+                raise subprocess.TimeoutExpired(argv, timeout)
+            try:
+                line = lines.get(timeout=0.5)
+                if line is None:
+                    ended = True
+                else:
+                    if output_size < MAX_OUTPUT_CHARS:
+                        remaining = MAX_OUTPUT_CHARS - output_size
+                        kept = line[:remaining]
+                        output_parts.append(kept)
+                        output_size += len(kept)
+                    batch.append(line)
+            except Empty:
+                pass
+            if batch and (
+                sum(map(len, batch)) >= 8192
+                or time.monotonic() - last_emit >= 1.0
+                or ended
+            ):
+                if event_cb:
+                    event_cb(
+                        "output",
+                        {
+                            "text": "".join(batch)[:16_000],
+                            "stream": "combined",
+                        },
+                    )
+                batch.clear()
+                last_emit = time.monotonic()
+        return_code = process.wait()
+        output = "".join(output_parts)
+        return {
+            "ok": return_code == 0,
+            "exit_code": return_code,
+            "duration_ms": round((time.monotonic() - started) * 1000),
+            "output": output,
+            "output_truncated": output_size >= MAX_OUTPUT_CHARS,
+        }
+    except subprocess.TimeoutExpired as exc:
+        output = "".join(output_parts)
+        return {
+            "ok": False,
+            "exit_code": None,
+            "duration_ms": round((time.monotonic() - started) * 1000),
+            "output": output,
+            "output_truncated": output_size >= MAX_OUTPUT_CHARS,
+            "error": "timeout",
+        }
+
+
+def execute(
+    config: dict[str, Any], task: dict[str, Any], event_cb: Any = None
+) -> dict[str, Any]:
+    action = task["action"]
+    args = task.get("args") or {}
+    enabled = set(config.get("enabled_actions", []))
+    if action not in enabled:
+        raise PermissionError(f"action is disabled on this node: {action}")
+
+    if action == "system.status":
+        commands = [
+            ["hostname"],
+            ["uname", "-a"],
+            ["uptime"],
+            ["df", "-h", "/"],
+        ]
+        results = [
+            run_process(command, timeout=30, event_cb=event_cb) for command in commands
+        ]
+        return {
+            "ok": all(item["ok"] for item in results),
+            "hostname": socket.gethostname(),
+            "platform": platform.platform(),
+            "commands": results,
+        }
+
+    if action == "repo.status":
+        repo = allowed_path(config, args["repo"])
+        return run_process(
+            [
+                "git",
+                "-C",
+                str(repo),
+                "status",
+                "--short",
+                "--branch",
+            ],
+            timeout=60,
+            event_cb=event_cb,
+        )
+
+    if action == "repo.fetch":
+        repo = allowed_path(config, args["repo"])
+        return run_process(
+            ["git", "-C", str(repo), "fetch", "--prune", "origin"],
+            timeout=300,
+            event_cb=event_cb,
+        )
+
+    if action == "service.logs":
+        return run_process(
+            [
+                "journalctl",
+                "--no-pager",
+                "-u",
+                args["service"],
+                "-n",
+                str(args.get("lines", 200)),
+            ],
+            timeout=60,
+            event_cb=event_cb,
+        )
+
+    if action == "command.run":
+        argv = args["argv"]
+        executable = argv[0]
+        if "/" in executable or executable not in set(
+            config.get("executable_allowlist", [])
+        ):
+            raise PermissionError(f"executable is not allowlisted: {executable}")
+        if shutil.which(executable) is None:
+            raise FileNotFoundError(f"executable is not installed: {executable}")
+        cwd = allowed_path(config, args["cwd"])
+        return run_process(
+            argv,
+            cwd=cwd,
+            timeout=int(args.get("timeout_seconds", 300)),
+            event_cb=event_cb,
+        )
+
+    raise ValueError(f"unsupported action: {action}")
+
+
+def pair(config_path: str, code: str) -> None:
+    config = read_config(config_path)
+    if config.get("node_id") or config.get("node_token"):
+        raise RuntimeError("node is already paired; remove its identity explicitly first")
+    response = api(
+        config,
+        "POST",
+        "/v1/pairings/claim",
+        {
+            "code": code,
+            "node_name": config["node_name"],
+            "agent_version": AGENT_VERSION,
+            "policy": {
+                "allowed_roots": config["allowed_roots"],
+                "enabled_actions": config["enabled_actions"],
+            },
+        },
+    )
+    config["node_id"] = response["node_id"]
+    config["node_token"] = response["node_token"]
+    write_config(config_path, config)
+    print(f"paired node {config['node_name']} as {config['node_id']}")
+
+
+def run(config_path: str, once: bool = False) -> None:
+    backoff = 2
+    while True:
+        config = read_config(config_path)
+        node_id = config.get("node_id")
+        node_token = config.get("node_token")
+        if not node_id or not node_token:
+            raise RuntimeError("node is not paired")
+        try:
+            api(
+                config,
+                "POST",
+                f"/v1/nodes/{node_id}/heartbeat",
+                {
+                    "agent_version": AGENT_VERSION,
+                    "policy": {
+                        "allowed_roots": config["allowed_roots"],
+                        "enabled_actions": config["enabled_actions"],
+                    },
+                },
+                node_token,
+            )
+            response = api(
+                config,
+                "GET",
+                f"/v1/nodes/{node_id}/tasks/next",
+                token=node_token,
+            )
+            task = response.get("task")
+            if task is not None:
+                def emit(kind: str, payload: dict[str, Any]) -> None:
+                    try:
+                        api(
+                            config,
+                            "POST",
+                            f"/v1/tasks/{task['id']}/events",
+                            {"kind": kind, "payload": payload},
+                            node_token,
+                        )
+                    except Exception as exc:
+                        print(f"event upload failed: {exc}", file=sys.stderr)
+
+                emit(
+                    "started",
+                    {
+                        "action": task["action"],
+                        "hostname": socket.gethostname(),
+                        "agent_version": AGENT_VERSION,
+                    },
+                )
+                try:
+                    result = execute(config, task, emit)
+                except Exception as exc:
+                    result = {
+                        "ok": False,
+                        "error": type(exc).__name__,
+                        "message": str(exc),
+                    }
+                emit(
+                    "finished",
+                    {
+                        "ok": bool(result.get("ok")),
+                        "exit_code": result.get("exit_code"),
+                        "error": result.get("error"),
+                    },
+                )
+                api(
+                    config,
+                    "POST",
+                    f"/v1/tasks/{task['id']}/result",
+                    result,
+                    node_token,
+                )
+            backoff = 2
+            if once:
+                return
+            time.sleep(float(config.get("poll_seconds", 5)))
+        except (OSError, URLError, RuntimeError) as exc:
+            print(f"connection error: {exc}; retrying in {backoff}s", file=sys.stderr)
+            if once:
+                raise
+            time.sleep(backoff)
+            backoff = min(backoff * 2, 120)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--config", default="/etc/canger-node/config.json", help="node config path"
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+    pair_parser = sub.add_parser("pair")
+    pair_parser.add_argument("--code", required=True)
+    run_parser = sub.add_parser("run")
+    run_parser.add_argument("--once", action="store_true")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    if args.command == "pair":
+        pair(args.config, args.code)
+    else:
+        run(args.config, args.once)
+
+
+if __name__ == "__main__":
+    main()

remote-node/persona/DEVELOPMENT-RECEIPT-20260729.md

@@ -0,0 +1,57 @@
+# 苍耳家庭 Ubuntu 运维节点 · 2026-07-29 开发回执
+
+## 已完成
+
+- 苍耳家庭 Ubuntu 以只出站方式接入新加坡大脑服务器；
+- 节点已配对并由 systemd 开机自动启动、断网自动重连；
+- 新加坡控制面能看到节点心跳并下发受限结构化任务；
+- 人格体可从 `remote-node/persona/README.md` 发起限时申请；
+- 申请者改为逐单填写名称与正式编号，不再固定为某个人格体；
+- 授权与申请者编号绑定，另一人格体不能借用；
+- 苍耳仍是唯一最终批准者。
+
+## 今日故障与修复
+
+- 一键安装脚本曾因执行位丢失而失败，已改为由入口显式调用安装脚本；
+- 旧配对码过期、节点名称重复、系统配置目录权限不足均已分别处理；
+- JZAO 移动硬盘是 ExFAT，不能保留 Linux 所需的 POSIX 权限，曾造成新加坡服务读取
+  源码失败；以后源码必须先经过 JZAO 源码运输闸门标准化，再传 Linux；
+- 测试工单曾把申请人写死，现已改成通用申请者字段。
+
+## 当前状态
+
+```yaml
+home_node: connected
+control_service: active
+request_entry: implemented
+requester_identity: explicit_per_workorder
+owner_approval: required
+local_direct_smtp: deprecated
+singapore_broadcast_tower_to_fifth_domain: live_verified
+fifth_domain_email_delivery_to_canger: sent_and_owner_approved
+fifth_domain_owner_decision: approved
+singapore_claim_state: not_claimed
+home_node_permission: not_activated
+```
+
+## 下一步
+
+2026-07-29 已由新加坡广播塔向第五域可信广播入口提交送达测试：
+
+```yaml
+workorder_id: bf72f16b-f81e-4de5-b0aa-aaee3d39e04c
+target: CANGER-HOME-UBUNTU-001
+approver: TCS-GL-009
+email_status: sent
+state: approved
+claimed: false
+local_permission_activated: false
+task_execution_requested: false
+```
+
+苍耳已点击批准，第五域于现场读回 `state: approved`、`claimed: false`。本次测试
+刻意没有在新加坡保存领取凭证，所以没有激活家庭节点权限，也没有执行任务。这证明
+“申请 → 新加坡广播塔 → 第五域广播主控塔 → 苍耳邮箱 → 苍耳批准”已经贯通。
+
+尚未完成的是正式工单的最后闭环：新加坡广播塔需要安全保存一次性领取凭证，在苍耳
+批准后领取回执，再把同一申请者、同一范围和有效期转换成家庭节点的限时会话。

remote-node/persona/README.md

@@ -0,0 +1,69 @@
+# 苍耳代码仓库 · 服务器运维申请入口
+
+本目录是任何已登记人格体从新加坡 `cang-ying` 代码仓库进入苍耳家庭
+Ubuntu 运维流程的固定门牌。申请人不写死：谁申请，就如实填写谁的名称和正式编号。
+
+## 主权边界
+
+- 人格体可以发起 1–8 小时的运维申请；
+- 人格体不能批准、拒绝或撤销申请；
+- 申请范围由家庭节点上报的安全策略决定，不能由人格体扩大；
+- 苍耳仍是最终批准者；
+- 未取得授权时，任务保持等待，不会下发到家庭 Ubuntu；
+- 仓库不保存令牌、批准链接、密码或配对码。
+
+## 发起申请
+
+在新加坡仓库根目录运行：
+
+```bash
+set -a
+. /etc/canger-control.env
+set +a
+python3 remote-node/tools/canger_persona.py \
+  --requester-name "本次申请者名称" \
+  --requester-id "本次申请者正式编号" \
+  request-access \
+  --hours 4 \
+  --reason "说明本次要检查、开发或修复什么"
+```
+
+成功回执必须同时满足：
+
+- `status` 为 `pending`；
+- `requester_name` 是本次申请者名称；
+- `requested_by` 是同一个申请者的正式编号；
+- `duration_seconds` 与申请时长一致；
+- `scopes` 和 `roots` 来自节点安全策略。
+
+这只表示“申请已经送达”，不表示苍耳已经批准。
+
+## 授权后创建任务
+
+```bash
+python3 remote-node/tools/canger_persona.py \
+  --requester-name "与获批工单相同的名称" \
+  --requester-id "与获批工单相同的正式编号" \
+  task \
+  --action system.status
+```
+
+在有效授权时段内，符合范围的任务会返回 `status: approved`；不符合范围或没有授权时，任务不会被家庭节点执行。
+
+## 查看结果
+
+```bash
+python3 remote-node/tools/canger_persona.py show task_xxx
+python3 remote-node/tools/canger_persona.py watch task_xxx
+```
+
+## 固定配置
+
+节点和控制端路径登记在：
+
+`remote-node/config/pufferfish-persona.json`
+
+申请者身份不保存在这个固定配置里，而是每张工单明确填写。需要变更节点时先更新
+胖头鱼节点注册表并重新验证心跳，不能让申请者在运行时自行替换节点。
+
+完整操作方法与状态边界见 `WORKORDER-GUIDE.md`。

remote-node/persona/WORKORDER-GUIDE.md

@@ -0,0 +1,77 @@
+# 苍耳服务器工单操作方法
+
+## 一张工单只回答四件事
+
+1. 你是谁；
+2. 你的正式编号是什么；
+3. 你申请做什么；
+4. 希望打开多长时间的受控会话。
+
+申请者可以是耳耳蛋、鉴影、铸渊或以后登记的其他人格体，系统不写死任何名字。
+申请者只能建单，不能批准自己的申请。苍耳收到授权通知后，自行决定是否批准。
+
+## 发起申请
+
+在新加坡 `cang-ying` 仓库根目录执行：
+
+```bash
+set -a
+. /etc/canger-control.env
+set +a
+python3 remote-node/tools/canger_persona.py \
+  --requester-name "申请者名称" \
+  --requester-id "申请者正式编号" \
+  request-access \
+  --hours 3 \
+  --reason "本次准备检查、开发、调试或修复的目的"
+```
+
+返回 `pending` 只代表申请已登记。没有苍耳批准，家庭 Ubuntu 不会执行任务。
+
+## 批准后的任务
+
+任务必须继续使用获批工单里的同一名称和编号：
+
+```bash
+python3 remote-node/tools/canger_persona.py \
+  --requester-name "申请者名称" \
+  --requester-id "申请者正式编号" \
+  task \
+  --action system.status
+```
+
+其他可用动作及范围由家庭节点上报的安全策略决定。不同编号不能借用别人的授权。
+
+## 广播链
+
+目标链路是：
+
+```text
+申请者
+→ 新加坡语言主控广播塔
+→ 第五域语言主控广播塔
+→ 按苍耳节点登记选择苍耳审批频道
+→ 苍耳邮箱授权链接
+→ 苍耳批准
+→ 新加坡控制端激活同一申请者的限时会话
+```
+
+请求正文不能携带或更换收件邮箱。苍耳邮箱只登记在第五域主控服务器的私密审批人
+注册表中。广播链未返回
+`owner_email_sent_by_registered_broadcast_tower` 前，不得声称邮件已经送达。
+
+2026-07-29 的送达测试已返回该回执；工单
+`bf72f16b-f81e-4de5-b0aa-aaee3d39e04c` 仅用于测试邮件，不会执行服务器任务。
+苍耳随后已点击批准，第五域状态为 `approved`，但该测试单的领取凭证被刻意丢弃，
+因此状态仍为 `claimed: false`，新加坡和家庭节点都没有获得权限。这是测试预期，
+不是节点离线。
+
+## 安全边界
+
+- 仓库不保存邮箱地址、SMTP 授权码、批准令牌、会话令牌或节点令牌；
+- 申请者名称和编号必须如实填写；
+- 苍耳没有点击批准时，权限保持关闭；
+- 第五域显示批准但新加坡尚未领取时，权限同样保持关闭；
+- 会话只能使用家庭节点上报的能力和受管目录；
+- 切换申请者、节点或扩大范围必须重新申请；
+- 每个任务都保留独立回执。

remote-node/server/control_plane.py

@@ -0,0 +1,978 @@
+#!/usr/bin/env python3
+"""Singapore-side control plane for the Cang'er outbound node."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+import hmac
+import html
+import json
+import os
+import re
+import smtplib
+import sqlite3
+import sys
+from email.message import EmailMessage
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any
+from urllib.parse import parse_qs, urlencode, urlparse
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from shared import (  # noqa: E402
+    MAX_JSON_BYTES,
+    MAX_RESULT_BYTES,
+    PAIRING_TTL_SECONDS,
+    TASK_LEASE_SECONDS,
+    compact_json,
+    new_id,
+    new_pairing_code,
+    new_token,
+    now_ts,
+    secret_hash,
+    secret_matches,
+    validate_node_name,
+    validate_task,
+)
+
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS pairings (
+    id TEXT PRIMARY KEY,
+    code_hash TEXT NOT NULL,
+    expires_at INTEGER NOT NULL,
+    used_at INTEGER
+);
+CREATE TABLE IF NOT EXISTS nodes (
+    id TEXT PRIMARY KEY,
+    name TEXT NOT NULL UNIQUE,
+    token_hash TEXT NOT NULL,
+    created_at INTEGER NOT NULL,
+    last_seen_at INTEGER,
+    agent_version TEXT,
+    policy_json TEXT NOT NULL DEFAULT '{}'
+);
+CREATE TABLE IF NOT EXISTS tasks (
+    id TEXT PRIMARY KEY,
+    node_id TEXT NOT NULL REFERENCES nodes(id),
+    action TEXT NOT NULL,
+    args_json TEXT NOT NULL,
+    status TEXT NOT NULL,
+    created_at INTEGER NOT NULL,
+    approved_at INTEGER,
+    leased_until INTEGER,
+    completed_at INTEGER,
+    result_json TEXT,
+    grant_id TEXT,
+    requested_by TEXT NOT NULL DEFAULT 'ICE-GL-ZY001',
+    executor_persona TEXT NOT NULL DEFAULT 'ICE-GL-CA001',
+    FOREIGN KEY(node_id) REFERENCES nodes(id)
+);
+CREATE INDEX IF NOT EXISTS idx_tasks_node_status
+ON tasks(node_id, status, created_at);
+CREATE TABLE IF NOT EXISTS task_events (
+    seq INTEGER PRIMARY KEY AUTOINCREMENT,
+    task_id TEXT NOT NULL REFERENCES tasks(id),
+    created_at INTEGER NOT NULL,
+    kind TEXT NOT NULL,
+    payload_json TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_task_events_task_seq
+ON task_events(task_id, seq);
+CREATE TABLE IF NOT EXISTS grants (
+    id TEXT PRIMARY KEY,
+    node_id TEXT NOT NULL REFERENCES nodes(id),
+    scopes_json TEXT NOT NULL,
+    roots_json TEXT NOT NULL,
+    reason TEXT NOT NULL,
+    duration_seconds INTEGER NOT NULL,
+    status TEXT NOT NULL,
+    created_at INTEGER NOT NULL,
+    approved_at INTEGER,
+    expires_at INTEGER,
+    revoked_at INTEGER
+    ,requested_by TEXT NOT NULL DEFAULT 'UNKNOWN'
+    ,requester_name TEXT NOT NULL DEFAULT ''
+);
+CREATE INDEX IF NOT EXISTS idx_grants_node_status
+ON grants(node_id, status, expires_at);
+"""
+
+
+class ControlPlane:
+    def __init__(
+        self,
+        db_path: str,
+        admin_token: str,
+        owner_token: str,
+        persona_token: str,
+        operator_token: str,
+        approval_signing_key: str,
+        owner_email: str = "",
+        public_url: str = "",
+    ) -> None:
+        for name, value in {
+            "admin token": admin_token,
+            "owner token": owner_token,
+            "persona token": persona_token,
+            "operator token": operator_token,
+        }.items():
+            if len(value) < 24:
+                raise ValueError(f"{name} must contain at least 24 characters")
+        if len(approval_signing_key) < 32:
+            raise ValueError("approval signing key must contain at least 32 characters")
+        self.db_path = db_path
+        self.admin_hash = secret_hash(admin_token)
+        self.owner_hash = secret_hash(owner_token)
+        self.persona_hash = secret_hash(persona_token)
+        self.operator_hash = secret_hash(operator_token)
+        self.operator_id = os.environ.get("CANGER_OPERATOR_ID", "ICE-GL-ZY001")
+        self.owner_id = os.environ.get("CANGER_OWNER_ID", "CANGER-HUMAN-OWNER")
+        self.executor_persona_id = os.environ.get(
+            "CANGER_EXECUTOR_PERSONA_ID", "ICE-GL-CA001"
+        )
+        self.approval_signing_key = approval_signing_key.encode("utf-8")
+        self.owner_email = owner_email
+        self.public_url = public_url.rstrip("/")
+        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
+        with self.db() as conn:
+            conn.executescript(SCHEMA)
+            columns = {
+                row["name"]
+                for row in conn.execute("PRAGMA table_info(grants)").fetchall()
+            }
+            if "requester_name" not in columns:
+                conn.execute(
+                    "ALTER TABLE grants ADD COLUMN requester_name "
+                    "TEXT NOT NULL DEFAULT ''"
+                )
+
+    def db(self) -> sqlite3.Connection:
+        conn = sqlite3.connect(self.db_path, timeout=10, isolation_level=None)
+        conn.row_factory = sqlite3.Row
+        conn.execute("PRAGMA foreign_keys=ON")
+        conn.execute("PRAGMA journal_mode=WAL")
+        return conn
+
+    @staticmethod
+    def bearer(headers: Any) -> str:
+        value = headers.get("Authorization", "")
+        if not value.startswith("Bearer "):
+            return ""
+        return value[7:].strip()
+
+    def role_allowed(self, headers: Any, *roles: str) -> bool:
+        token = self.bearer(headers)
+        hashes = {
+            "admin": self.admin_hash,
+            "owner": self.owner_hash,
+            "persona": self.persona_hash,
+            "operator": self.operator_hash,
+        }
+        return any(secret_matches(token, hashes[role]) for role in roles)
+
+    def node_for_token(self, node_id: str, headers: Any) -> sqlite3.Row | None:
+        token = self.bearer(headers)
+        with self.db() as conn:
+            row = conn.execute(
+                "SELECT * FROM nodes WHERE id=?", (node_id,)
+            ).fetchone()
+        if row is None or not secret_matches(token, row["token_hash"]):
+            return None
+        return row
+
+    def grant_token(self, grant_id: str, link_expires_at: int) -> str:
+        message = f"canger-grant/v1\n{grant_id}\n{link_expires_at}".encode("utf-8")
+        signature = hmac.new(
+            self.approval_signing_key, message, hashlib.sha256
+        ).digest()
+        encoded = base64.urlsafe_b64encode(signature).decode("ascii").rstrip("=")
+        return f"{link_expires_at}.{encoded}"
+
+    def verify_grant_token(self, grant_id: str, token: str) -> bool:
+        try:
+            raw_expiry, _ = token.split(".", 1)
+            expiry = int(raw_expiry)
+        except (ValueError, TypeError):
+            return False
+        if expiry < now_ts():
+            return False
+        return hmac.compare_digest(token, self.grant_token(grant_id, expiry))
+
+    def send_grant_email(self, grant: dict[str, Any]) -> str:
+        required = {
+            "owner_email": self.owner_email,
+            "public_url": self.public_url,
+            "smtp_host": os.environ.get("CANGER_SMTP_HOST", ""),
+            "smtp_from": os.environ.get("CANGER_SMTP_FROM", ""),
+        }
+        if not all(required.values()):
+            return "not_configured"
+        link_expiry = now_ts() + 30 * 60
+        token = self.grant_token(grant["id"], link_expiry)
+        link = (
+            f"{self.public_url}/approve/grant?"
+            + urlencode({"id": grant["id"], "token": token})
+        )
+        message = EmailMessage()
+        message["Subject"] = f"苍耳服务器副控运维授权（{grant['duration_seconds'] // 3600} 小时）"
+        message["From"] = required["smtp_from"]
+        message["To"] = required["owner_email"]
+        message.set_content(
+            "有人格体申请进入苍耳服务器的限时受控运维会话。"
+            "苍耳仍是服务器主控与最终决定者。\n\n"
+            f"申请者：{grant['requester_name']} ({grant['requested_by']})\n"
+            f"申请目的：{grant['reason']}\n"
+            f"时长：{grant['duration_seconds'] // 3600} 小时\n\n"
+            "期间可以：检查服务器、同步和修改代码、运行测试、查看故障日志。\n"
+            "不能：修改系统账户、密钥、网络、防火墙、磁盘和系统启动，"
+            "也不能操作受管工作区以外的个人文件。\n\n"
+            f"打开链接查看并确认：{link}\n\n"
+            "链接 30 分钟失效；授权本身到期后自动失效。"
+        )
+        host = required["smtp_host"]
+        port = int(os.environ.get("CANGER_SMTP_PORT", "465"))
+        username = os.environ.get("CANGER_SMTP_USERNAME", "")
+        password = os.environ.get("CANGER_SMTP_PASSWORD", "")
+        if port == 465:
+            client: Any = smtplib.SMTP_SSL(host, port, timeout=20)
+        else:
+            client = smtplib.SMTP(host, port, timeout=20)
+            client.starttls()
+        try:
+            if username:
+                client.login(username, password)
+            client.send_message(message)
+        finally:
+            client.quit()
+        return "sent"
+
+
+class Handler(BaseHTTPRequestHandler):
+    server_version = "CangerControl/0.1"
+
+    @property
+    def app(self) -> ControlPlane:
+        return self.server.app  # type: ignore[attr-defined]
+
+    def log_message(self, fmt: str, *args: Any) -> None:
+        print(f"{self.address_string()} - {fmt % args}", file=sys.stderr)
+
+    def send_json(self, status: int, payload: Any) -> None:
+        data = compact_json(payload).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(data)))
+        self.send_header("Cache-Control", "no-store")
+        self.end_headers()
+        self.wfile.write(data)
+
+    def send_html(self, status: int, content: str) -> None:
+        data = content.encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "text/html; charset=utf-8")
+        self.send_header("Content-Length", str(len(data)))
+        self.send_header("Cache-Control", "no-store")
+        self.end_headers()
+        self.wfile.write(data)
+
+    def read_json(self) -> dict[str, Any]:
+        try:
+            size = int(self.headers.get("Content-Length", "0"))
+        except ValueError as exc:
+            raise ValueError("invalid Content-Length") from exc
+        if size <= 0 or size > MAX_JSON_BYTES:
+            raise ValueError("invalid JSON body size")
+        value = json.loads(self.rfile.read(size))
+        if not isinstance(value, dict):
+            raise ValueError("JSON body must be an object")
+        return value
+
+    def path_parts(self) -> list[str]:
+        return [part for part in urlparse(self.path).path.split("/") if part]
+
+    def require_role(self, *roles: str) -> bool:
+        if self.app.role_allowed(self.headers, *roles):
+            return True
+        self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"})
+        return False
+
+    def requester_id(self) -> str:
+        if self.app.role_allowed(self.headers, "persona"):
+            return self.app.executor_persona_id
+        if self.app.role_allowed(self.headers, "owner"):
+            return self.app.owner_id
+        return self.app.operator_id
+
+    def request_identity(self, body: dict[str, Any]) -> tuple[str, str]:
+        requester_id = body.get("requester_id")
+        requester_name = body.get("requester_name")
+        if not isinstance(requester_id, str) or not re.fullmatch(
+            r"[A-Za-z0-9._:+∞-]{2,80}", requester_id
+        ):
+            raise ValueError("valid requester_id is required")
+        if (
+            not isinstance(requester_name, str)
+            or not requester_name.strip()
+            or len(requester_name) > 100
+        ):
+            raise ValueError("valid requester_name is required")
+        return requester_id, requester_name.strip()
+
+    def do_GET(self) -> None:  # noqa: N802
+        try:
+            parts = self.path_parts()
+            if parts == ["health"]:
+                self.send_json(HTTPStatus.OK, {"ok": True, "service": "canger-control"})
+                return
+            if parts == ["approve", "grant"]:
+                self.grant_approval_page()
+                return
+            if parts == ["v1", "nodes"]:
+                if not self.require_role("admin", "owner", "persona", "operator"):
+                    return
+                with self.app.db() as conn:
+                    rows = conn.execute(
+                        "SELECT id,name,created_at,last_seen_at,agent_version FROM nodes "
+                        "ORDER BY created_at"
+                    ).fetchall()
+                self.send_json(HTTPStatus.OK, {"nodes": [dict(row) for row in rows]})
+                return
+            if len(parts) == 4 and parts[:2] == ["v1", "tasks"]:
+                if parts[3] != "":
+                    pass
+            if len(parts) == 3 and parts[:2] == ["v1", "tasks"]:
+                if not self.require_role("admin", "owner", "persona", "operator"):
+                    return
+                with self.app.db() as conn:
+                    row = conn.execute(
+                        "SELECT * FROM tasks WHERE id=?", (parts[2],)
+                    ).fetchone()
+                if row is None:
+                    self.send_json(HTTPStatus.NOT_FOUND, {"error": "task not found"})
+                    return
+                payload = dict(row)
+                payload["args"] = json.loads(payload.pop("args_json"))
+                if payload["result_json"] is not None:
+                    payload["result"] = json.loads(payload.pop("result_json"))
+                else:
+                    payload.pop("result_json")
+                self.send_json(HTTPStatus.OK, payload)
+                return
+            if (
+                len(parts) == 4
+                and parts[:2] == ["v1", "tasks"]
+                and parts[3] == "events"
+            ):
+                if not self.require_role("admin", "owner", "persona", "operator"):
+                    return
+                after = 0
+                query = urlparse(self.path).query
+                for item in query.split("&"):
+                    if item.startswith("after="):
+                        after = max(0, int(item.split("=", 1)[1]))
+                with self.app.db() as conn:
+                    rows = conn.execute(
+                        "SELECT seq,created_at,kind,payload_json FROM task_events "
+                        "WHERE task_id=? AND seq>? ORDER BY seq LIMIT 500",
+                        (parts[2], after),
+                    ).fetchall()
+                events = []
+                for row in rows:
+                    event = dict(row)
+                    event["payload"] = json.loads(event.pop("payload_json"))
+                    events.append(event)
+                self.send_json(HTTPStatus.OK, {"events": events})
+                return
+            if parts == ["v1", "grants"]:
+                if not self.require_role("admin", "owner", "persona", "operator"):
+                    return
+                with self.app.db() as conn:
+                    rows = conn.execute(
+                        "SELECT * FROM grants ORDER BY created_at DESC LIMIT 200"
+                    ).fetchall()
+                grants = []
+                for row in rows:
+                    item = dict(row)
+                    item["scopes"] = json.loads(item.pop("scopes_json"))
+                    item["roots"] = json.loads(item.pop("roots_json"))
+                    grants.append(item)
+                self.send_json(HTTPStatus.OK, {"grants": grants})
+                return
+            if (
+                len(parts) == 5
+                and parts[:2] == ["v1", "nodes"]
+                and parts[3:] == ["tasks", "next"]
+            ):
+                self.next_task(parts[2])
+                return
+            self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"})
+        except Exception as exc:
+            self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+
+    def do_POST(self) -> None:  # noqa: N802
+        try:
+            parts = self.path_parts()
+            if parts == ["v1", "pairings"]:
+                self.create_pairing()
+                return
+            if parts == ["v1", "pairings", "claim"]:
+                self.claim_pairing()
+                return
+            if parts == ["v1", "tasks"]:
+                self.create_task()
+                return
+            if parts == ["v1", "grants"]:
+                self.create_grant()
+                return
+            if len(parts) == 4 and parts[:2] == ["v1", "grants"]:
+                if parts[3] == "revoke":
+                    self.revoke_grant(parts[2])
+                    return
+            if len(parts) == 4 and parts[:2] == ["v1", "tasks"]:
+                if parts[3] == "approve":
+                    self.decide_task(parts[2], "approved")
+                    return
+                if parts[3] == "reject":
+                    self.decide_task(parts[2], "rejected")
+                    return
+                if parts[3] == "result":
+                    self.complete_task(parts[2])
+                    return
+                if parts[3] == "events":
+                    self.append_event(parts[2])
+                    return
+            if (
+                len(parts) == 4
+                and parts[:2] == ["v1", "nodes"]
+                and parts[3] == "heartbeat"
+            ):
+                self.heartbeat(parts[2])
+                return
+            if parts == ["approve", "grant"]:
+                self.approve_grant_form()
+                return
+            self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"})
+        except json.JSONDecodeError:
+            self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid JSON"})
+        except ValueError as exc:
+            self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+        except Exception as exc:
+            self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(exc)})
+
+    def create_pairing(self) -> None:
+        if not self.require_role("admin"):
+            return
+        code = new_pairing_code()
+        pairing_id = new_id("pair")
+        expires_at = now_ts() + PAIRING_TTL_SECONDS
+        with self.app.db() as conn:
+            conn.execute(
+                "INSERT INTO pairings(id,code_hash,expires_at) VALUES(?,?,?)",
+                (pairing_id, secret_hash(code), expires_at),
+            )
+        self.send_json(
+            HTTPStatus.CREATED,
+            {"pairing_id": pairing_id, "code": code, "expires_at": expires_at},
+        )
+
+    def claim_pairing(self) -> None:
+        body = self.read_json()
+        code = body.get("code")
+        name = validate_node_name(body.get("node_name"))
+        agent_version = str(body.get("agent_version", ""))[:64]
+        policy = self.validated_node_policy(body.get("policy", {}))
+        if not isinstance(code, str):
+            raise ValueError("code is required")
+        now = now_ts()
+        node_id = new_id("node")
+        node_token = new_token()
+        with self.app.db() as conn:
+            conn.execute("BEGIN IMMEDIATE")
+            rows = conn.execute(
+                "SELECT * FROM pairings WHERE used_at IS NULL AND expires_at>=?", (now,)
+            ).fetchall()
+            pairing = next(
+                (row for row in rows if secret_matches(code, row["code_hash"])), None
+            )
+            if pairing is None:
+                conn.rollback()
+                self.send_json(
+                    HTTPStatus.UNAUTHORIZED, {"error": "invalid or expired pairing code"}
+                )
+                return
+            existing = conn.execute(
+                "SELECT id FROM nodes WHERE name=?", (name,)
+            ).fetchone()
+            if existing is not None:
+                conn.rollback()
+                self.send_json(
+                    HTTPStatus.CONFLICT, {"error": "node name already registered"}
+                )
+                return
+            conn.execute(
+                "INSERT INTO nodes(id,name,token_hash,created_at,last_seen_at,"
+                "agent_version,policy_json) VALUES(?,?,?,?,?,?,?)",
+                (
+                    node_id,
+                    name,
+                    secret_hash(node_token),
+                    now,
+                    now,
+                    agent_version,
+                    compact_json(policy),
+                ),
+            )
+            conn.execute(
+                "UPDATE pairings SET used_at=? WHERE id=?", (now, pairing["id"])
+            )
+            conn.commit()
+        self.send_json(
+            HTTPStatus.CREATED, {"node_id": node_id, "node_token": node_token}
+        )
+
+    @staticmethod
+    def validated_node_policy(value: Any) -> dict[str, Any]:
+        if not isinstance(value, dict):
+            raise ValueError("node policy must be an object")
+        roots = value.get("allowed_roots", [])
+        actions = value.get("enabled_actions", [])
+        allowed_actions = {
+            "system.status",
+            "repo.status",
+            "repo.fetch",
+            "service.logs",
+            "command.run",
+        }
+        if (
+            not isinstance(roots, list)
+            or not roots
+            or any(not isinstance(root, str) or not root.startswith("/") for root in roots)
+        ):
+            raise ValueError("node policy requires absolute allowed_roots")
+        if (
+            not isinstance(actions, list)
+            or any(action not in allowed_actions for action in actions)
+        ):
+            raise ValueError("node policy contains unsupported actions")
+        return {
+            "profile": "developer_standard",
+            "allowed_roots": sorted(set(root.rstrip("/") for root in roots)),
+            "enabled_actions": sorted(set(actions)),
+        }
+
+    def create_task(self) -> None:
+        if not self.require_role("admin", "operator", "persona"):
+            return
+        body = self.read_json()
+        requested_by, requester_name = self.request_identity(body)
+        node_id = body.get("node_id")
+        if not isinstance(node_id, str):
+            raise ValueError("node_id is required")
+        action, args = validate_task(body.get("action"), body.get("args"))
+        task_id = new_id("task")
+        approved_at = None
+        grant_id = None
+        with self.app.db() as conn:
+            if conn.execute("SELECT 1 FROM nodes WHERE id=?", (node_id,)).fetchone() is None:
+                self.send_json(HTTPStatus.NOT_FOUND, {"error": "node not found"})
+                return
+            rows = conn.execute(
+                "SELECT * FROM grants WHERE node_id=? AND status='active' "
+                "AND expires_at>? AND requested_by=? ORDER BY expires_at",
+                (node_id, now_ts(), requested_by),
+            ).fetchall()
+            for row in rows:
+                if self.task_allowed_by_grant(action, args, row):
+                    grant_id = row["id"]
+                    approved_at = now_ts()
+                    break
+            status = "approved" if grant_id else "pending"
+            conn.execute(
+                "INSERT INTO tasks(id,node_id,action,args_json,status,created_at,"
+                "approved_at,grant_id,requested_by,executor_persona) "
+                "VALUES(?,?,?,?,?,?,?,?,?,?)",
+                (
+                    task_id,
+                    node_id,
+                    action,
+                    compact_json(args),
+                    status,
+                    now_ts(),
+                    approved_at,
+                    grant_id,
+                    requested_by,
+                    requested_by,
+                ),
+            )
+        self.send_json(
+            HTTPStatus.CREATED,
+            {
+                "task_id": task_id,
+                "status": status,
+                "approval_required": grant_id is None,
+                "grant_id": grant_id,
+                "requested_by": requested_by,
+                "requester_name": requester_name,
+                "executor_persona": requested_by,
+                "action": action,
+                "args": args,
+            },
+        )
+
+    @staticmethod
+    def task_allowed_by_grant(
+        action: str, args: dict[str, Any], grant: sqlite3.Row
+    ) -> bool:
+        scopes = set(json.loads(grant["scopes_json"]))
+        if action not in scopes:
+            return False
+        roots = [str(root).rstrip("/") for root in json.loads(grant["roots_json"])]
+        candidate = None
+        if action in {"repo.status", "repo.fetch"}:
+            candidate = args["repo"]
+        elif action == "command.run":
+            candidate = args["cwd"]
+        if candidate is None:
+            return True
+        normalized = str(candidate).rstrip("/")
+        return any(
+            normalized == root or normalized.startswith(root + "/") for root in roots
+        )
+
+    def create_grant(self) -> None:
+        if not self.require_role("admin", "operator", "persona"):
+            return
+        body = self.read_json()
+        requested_by, requester_name = self.request_identity(body)
+        node_id = body.get("node_id")
+        reason = body.get("reason", "编程与调试")
+        duration = body.get("duration_seconds", 4 * 3600)
+        if not isinstance(node_id, str):
+            raise ValueError("node_id is required")
+        if not isinstance(reason, str) or not reason.strip() or len(reason) > 300:
+            raise ValueError("invalid grant reason")
+        if not isinstance(duration, int) or not 3600 <= duration <= 8 * 3600:
+            raise ValueError("grant duration must be between 1 and 8 hours")
+        with self.app.db() as conn:
+            node = conn.execute(
+                "SELECT policy_json FROM nodes WHERE id=?", (node_id,)
+            ).fetchone()
+            if node is None:
+                self.send_json(HTTPStatus.NOT_FOUND, {"error": "node not found"})
+                return
+            policy = json.loads(node["policy_json"])
+            scopes = policy.get("enabled_actions", [])
+            roots = policy.get("allowed_roots", [])
+            if not scopes or not roots:
+                self.send_json(
+                    HTTPStatus.CONFLICT,
+                    {"error": "node has not reported a usable safety policy"},
+                )
+                return
+            grant_id = new_id("grant")
+            grant = {
+                "id": grant_id,
+                "node_id": node_id,
+                "scopes": scopes,
+                "roots": roots,
+                "reason": reason.strip(),
+                "duration_seconds": duration,
+                "requested_by": requested_by,
+                "requester_name": requester_name,
+            }
+            conn.execute(
+                "INSERT INTO grants(id,node_id,scopes_json,roots_json,reason,"
+                "duration_seconds,status,created_at,requested_by,requester_name) "
+                "VALUES(?,?,?,?,?,?,?,?,?,?)",
+                (
+                    grant_id,
+                    node_id,
+                    compact_json(grant["scopes"]),
+                    compact_json(grant["roots"]),
+                    grant["reason"],
+                    duration,
+                    "pending",
+                    now_ts(),
+                    requested_by,
+                    requester_name,
+                ),
+            )
+        try:
+            notification = self.app.send_grant_email(grant)
+        except Exception as exc:
+            print(f"grant email failed: {exc}", file=sys.stderr)
+            notification = "failed"
+        self.send_json(
+            HTTPStatus.CREATED,
+            {
+                "grant_id": grant_id,
+                "status": "pending",
+                "requested_by": requested_by,
+                "requester_name": requester_name,
+                "notification": notification,
+                "duration_seconds": duration,
+                "scopes": grant["scopes"],
+                "roots": grant["roots"],
+            },
+        )
+
+    def grant_approval_page(self) -> None:
+        query = parse_qs(urlparse(self.path).query)
+        grant_id = query.get("id", [""])[0]
+        token = query.get("token", [""])[0]
+        if not self.app.verify_grant_token(grant_id, token):
+            self.send_html(HTTPStatus.UNAUTHORIZED, "<h1>链接无效或已过期</h1>")
+            return
+        with self.app.db() as conn:
+            row = conn.execute("SELECT * FROM grants WHERE id=?", (grant_id,)).fetchone()
+        if row is None or row["status"] != "pending":
+            self.send_html(HTTPStatus.CONFLICT, "<h1>申请已处理或不存在</h1>")
+            return
+        content = f"""<!doctype html><meta charset="utf-8">
+<title>苍耳服务器临时授权</title>
+<h1>苍耳服务器临时授权</h1>
+<p>主控：苍耳。是否批准，以苍耳本次选择为准。</p>
+<p>申请者：{html.escape(row["requester_name"])} ({html.escape(row["requested_by"])})</p>
+<p>申请目的：{html.escape(row["reason"])}</p>
+<p>时长：{row["duration_seconds"] // 3600} 小时</p>
+<p>期间可以：检查服务器、同步和修改代码、运行测试、查看故障日志。</p>
+<p>不能：修改系统账户、密钥、网络、防火墙、磁盘和系统启动，
+也不能操作受管工作区以外的个人文件。</p>
+<p>授权到期自动失效，可从新加坡端提前撤销。</p>
+<form method="post" action="/approve/grant">
+<input type="hidden" name="id" value="{html.escape(grant_id)}">
+<input type="hidden" name="token" value="{html.escape(token)}">
+<button type="submit">批准本次临时授权</button>
+</form>"""
+        self.send_html(HTTPStatus.OK, content)
+
+    def approve_grant_form(self) -> None:
+        size = int(self.headers.get("Content-Length", "0"))
+        if size <= 0 or size > 16_384:
+            raise ValueError("invalid form body size")
+        form = parse_qs(self.rfile.read(size).decode("utf-8"))
+        grant_id = form.get("id", [""])[0]
+        token = form.get("token", [""])[0]
+        if not self.app.verify_grant_token(grant_id, token):
+            self.send_html(HTTPStatus.UNAUTHORIZED, "<h1>链接无效或已过期</h1>")
+            return
+        approved_at = now_ts()
+        with self.app.db() as conn:
+            row = conn.execute("SELECT * FROM grants WHERE id=?", (grant_id,)).fetchone()
+            if row is None or row["status"] != "pending":
+                self.send_html(HTTPStatus.CONFLICT, "<h1>申请已处理或不存在</h1>")
+                return
+            expires_at = approved_at + row["duration_seconds"]
+            conn.execute(
+                "UPDATE grants SET status='active',approved_at=?,expires_at=? "
+                "WHERE id=? AND status='pending'",
+                (approved_at, expires_at, grant_id),
+            )
+        self.send_html(
+            HTTPStatus.OK,
+            f"<h1>副控运维授权成功</h1><p>苍耳仍保留最终控制权；"
+            f"本次授权将在 {expires_at} 自动失效，可以关闭此页面。</p>",
+        )
+
+    def revoke_grant(self, grant_id: str) -> None:
+        if not self.require_role("admin", "owner"):
+            return
+        with self.app.db() as conn:
+            cursor = conn.execute(
+                "UPDATE grants SET status='revoked',revoked_at=? "
+                "WHERE id=? AND status='active'",
+                (now_ts(), grant_id),
+            )
+        if cursor.rowcount != 1:
+            self.send_json(HTTPStatus.CONFLICT, {"error": "grant is not active"})
+            return
+        self.send_json(HTTPStatus.OK, {"grant_id": grant_id, "status": "revoked"})
+
+    def decide_task(self, task_id: str, decision: str) -> None:
+        if not self.require_role("admin", "owner"):
+            return
+        with self.app.db() as conn:
+            cursor = conn.execute(
+                "UPDATE tasks SET status=?,approved_at=? "
+                "WHERE id=? AND status='pending'",
+                (decision, now_ts(), task_id),
+            )
+        if cursor.rowcount != 1:
+            self.send_json(
+                HTTPStatus.CONFLICT, {"error": "task is not pending or does not exist"}
+            )
+            return
+        self.send_json(HTTPStatus.OK, {"task_id": task_id, "status": decision})
+
+    def next_task(self, node_id: str) -> None:
+        if self.app.node_for_token(node_id, self.headers) is None:
+            self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized node"})
+            return
+        now = now_ts()
+        with self.app.db() as conn:
+            conn.execute("BEGIN IMMEDIATE")
+            conn.execute(
+                "UPDATE grants SET status='expired' "
+                "WHERE status='active' AND expires_at<=?",
+                (now,),
+            )
+            conn.execute(
+                "UPDATE tasks SET status='pending',approved_at=NULL,grant_id=NULL "
+                "WHERE node_id=? AND status='approved' AND grant_id IS NOT NULL "
+                "AND grant_id NOT IN (SELECT id FROM grants WHERE status='active' "
+                "AND expires_at>?)",
+                (node_id, now),
+            )
+            conn.execute(
+                "UPDATE tasks SET status='approved',leased_until=NULL "
+                "WHERE node_id=? AND status='leased' AND leased_until<?",
+                (node_id, now),
+            )
+            row = conn.execute(
+                "SELECT id,action,args_json FROM tasks "
+                "WHERE node_id=? AND status='approved' ORDER BY approved_at,id LIMIT 1",
+                (node_id,),
+            ).fetchone()
+            if row is not None:
+                conn.execute(
+                    "UPDATE tasks SET status='leased',leased_until=? WHERE id=?",
+                    (now + TASK_LEASE_SECONDS, row["id"]),
+                )
+            conn.execute(
+                "UPDATE nodes SET last_seen_at=? WHERE id=?", (now, node_id)
+            )
+            conn.commit()
+        if row is None:
+            self.send_json(HTTPStatus.OK, {"task": None})
+            return
+        self.send_json(
+            HTTPStatus.OK,
+            {
+                "task": {
+                    "id": row["id"],
+                    "action": row["action"],
+                    "args": json.loads(row["args_json"]),
+                }
+            },
+        )
+
+    def complete_task(self, task_id: str) -> None:
+        body = self.read_json()
+        if len(compact_json(body).encode("utf-8")) > MAX_RESULT_BYTES:
+            raise ValueError("task result is too large")
+        with self.app.db() as conn:
+            row = conn.execute(
+                "SELECT node_id,status FROM tasks WHERE id=?", (task_id,)
+            ).fetchone()
+            if row is None:
+                self.send_json(HTTPStatus.NOT_FOUND, {"error": "task not found"})
+                return
+            if self.app.node_for_token(row["node_id"], self.headers) is None:
+                self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized node"})
+                return
+            if row["status"] != "leased":
+                self.send_json(HTTPStatus.CONFLICT, {"error": "task is not leased"})
+                return
+            conn.execute(
+                "UPDATE tasks SET status='completed',completed_at=?,result_json=? "
+                "WHERE id=?",
+                (now_ts(), compact_json(body), task_id),
+            )
+        self.send_json(HTTPStatus.OK, {"task_id": task_id, "status": "completed"})
+
+    def append_event(self, task_id: str) -> None:
+        body = self.read_json()
+        kind = body.get("kind")
+        payload = body.get("payload", {})
+        if (
+            not isinstance(kind, str)
+            or not kind
+            or len(kind) > 64
+            or not all(char.isalnum() or char in "._-" for char in kind)
+        ):
+            raise ValueError("invalid event kind")
+        encoded = compact_json(payload)
+        if len(encoded.encode("utf-8")) > 32 * 1024:
+            raise ValueError("event payload is too large")
+        with self.app.db() as conn:
+            row = conn.execute(
+                "SELECT node_id FROM tasks WHERE id=?", (task_id,)
+            ).fetchone()
+            if row is None:
+                self.send_json(HTTPStatus.NOT_FOUND, {"error": "task not found"})
+                return
+            if self.app.node_for_token(row["node_id"], self.headers) is None:
+                self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized node"})
+                return
+            cursor = conn.execute(
+                "INSERT INTO task_events(task_id,created_at,kind,payload_json) "
+                "VALUES(?,?,?,?)",
+                (task_id, now_ts(), kind, encoded),
+            )
+        self.send_json(
+            HTTPStatus.CREATED, {"task_id": task_id, "seq": cursor.lastrowid}
+        )
+
+    def heartbeat(self, node_id: str) -> None:
+        node = self.app.node_for_token(node_id, self.headers)
+        if node is None:
+            self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "unauthorized node"})
+            return
+        body = self.read_json()
+        version = str(body.get("agent_version", ""))[:64]
+        policy = self.validated_node_policy(body.get("policy", {}))
+        with self.app.db() as conn:
+            conn.execute(
+                "UPDATE nodes SET last_seen_at=?,agent_version=?,policy_json=? WHERE id=?",
+                (now_ts(), version, compact_json(policy), node_id),
+            )
+        self.send_json(HTTPStatus.OK, {"ok": True})
+
+
+class AppServer(ThreadingHTTPServer):
+    daemon_threads = True
+
+    def __init__(self, address: tuple[str, int], app: ControlPlane):
+        self.app = app
+        super().__init__(address, Handler)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--bind", default=os.environ.get("CANGER_BIND", "127.0.0.1"))
+    parser.add_argument(
+        "--port", type=int, default=int(os.environ.get("CANGER_PORT", "8787"))
+    )
+    parser.add_argument(
+        "--db",
+        default=os.environ.get(
+            "CANGER_CONTROL_DB", "/var/lib/canger-control/control.sqlite3"
+        ),
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    app = ControlPlane(
+        args.db,
+        os.environ.get("CANGER_ADMIN_TOKEN", ""),
+        os.environ.get("CANGER_OWNER_TOKEN", ""),
+        os.environ.get("CANGER_PERSONA_TOKEN", ""),
+        os.environ.get("CANGER_OPERATOR_TOKEN", ""),
+        os.environ.get("CANGER_APPROVAL_SIGNING_KEY", ""),
+        os.environ.get("CANGER_OWNER_EMAIL", ""),
+        os.environ.get("CANGER_PUBLIC_URL", ""),
+    )
+    server = AppServer((args.bind, args.port), app)
+    print(f"canger-control listening on {args.bind}:{args.port}")
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()

remote-node/shared.py

@@ -0,0 +1,111 @@
+"""Shared protocol rules for the Cang'er remote node."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import re
+import secrets
+import time
+from typing import Any
+
+
+MAX_JSON_BYTES = 256 * 1024
+MAX_RESULT_BYTES = 512 * 1024
+PAIRING_TTL_SECONDS = 10 * 60
+TASK_LEASE_SECONDS = 5 * 60
+
+ACTION_NAMES = {
+    "system.status",
+    "repo.status",
+    "repo.fetch",
+    "service.logs",
+    "command.run",
+}
+
+NODE_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
+SERVICE_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.@-]{0,127}$")
+
+
+def now_ts() -> int:
+    return int(time.time())
+
+
+def new_id(prefix: str) -> str:
+    return f"{prefix}_{secrets.token_hex(12)}"
+
+
+def new_token() -> str:
+    return secrets.token_urlsafe(32)
+
+
+def new_pairing_code() -> str:
+    alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
+    return "-".join(
+        "".join(secrets.choice(alphabet) for _ in range(4)) for _ in range(3)
+    )
+
+
+def secret_hash(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def secret_matches(value: str, expected_hash: str) -> bool:
+    return hmac.compare_digest(secret_hash(value), expected_hash)
+
+
+def compact_json(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
+
+
+def validate_node_name(value: Any) -> str:
+    if not isinstance(value, str) or not NODE_NAME_RE.fullmatch(value):
+        raise ValueError("node_name must use 1-64 letters, digits, dot, underscore or dash")
+    return value
+
+
+def validate_task(action: Any, args: Any) -> tuple[str, dict[str, Any]]:
+    if not isinstance(action, str) or action not in ACTION_NAMES:
+        raise ValueError(f"unsupported action: {action!r}")
+    if args is None:
+        args = {}
+    if not isinstance(args, dict):
+        raise ValueError("args must be an object")
+
+    if action == "system.status":
+        if args:
+            raise ValueError("system.status does not accept args")
+    elif action in {"repo.status", "repo.fetch"}:
+        if set(args) != {"repo"} or not isinstance(args["repo"], str):
+            raise ValueError(f"{action} requires string arg: repo")
+    elif action == "service.logs":
+        if not isinstance(args.get("service"), str) or not SERVICE_NAME_RE.fullmatch(
+            args["service"]
+        ):
+            raise ValueError("service.logs requires a safe service name")
+        lines = args.get("lines", 200)
+        if not isinstance(lines, int) or not 1 <= lines <= 2000:
+            raise ValueError("service.logs lines must be between 1 and 2000")
+        args = {"service": args["service"], "lines": lines}
+    elif action == "command.run":
+        argv = args.get("argv")
+        cwd = args.get("cwd")
+        timeout = args.get("timeout_seconds", 300)
+        if (
+            not isinstance(argv, list)
+            or not argv
+            or len(argv) > 64
+            or any(not isinstance(item, str) or len(item) > 4096 for item in argv)
+        ):
+            raise ValueError("command.run argv must be a non-empty string array")
+        if not isinstance(cwd, str):
+            raise ValueError("command.run requires string arg: cwd")
+        if not isinstance(timeout, int) or not 1 <= timeout <= 1800:
+            raise ValueError("command.run timeout_seconds must be between 1 and 1800")
+        args = {"argv": argv, "cwd": cwd, "timeout_seconds": timeout}
+
+    encoded = compact_json(args).encode("utf-8")
+    if len(encoded) > MAX_JSON_BYTES:
+        raise ValueError("task args are too large")
+    return action, args

remote-node/systemd/canger-control.service

@@ -0,0 +1,22 @@
+[Unit]
+Description=Canger Singapore Control Plane
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=canger-control
+Group=canger-control
+WorkingDirectory=/opt/zhuyuan/cang-ying/remote-node
+EnvironmentFile=/etc/canger-control.env
+ExecStart=/usr/bin/python3 /opt/zhuyuan/cang-ying/remote-node/server/control_plane.py
+Restart=always
+RestartSec=3
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ReadWritePaths=/var/lib/canger-control
+
+[Install]
+WantedBy=multi-user.target

remote-node/systemd/canger-node.service

@@ -0,0 +1,21 @@
+[Unit]
+Description=Canger Home Ubuntu Outbound Node
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=canger-agent
+Group=canger-agent
+WorkingDirectory=/opt/canger-node
+ExecStart=/usr/bin/python3 /opt/canger-node/canger_node.py --config /etc/canger-node/config.json run
+Restart=always
+RestartSec=5
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ReadWritePaths=/etc/canger-node
+
+[Install]
+WantedBy=multi-user.target

remote-node/tests/test_end_to_end.py

@@ -0,0 +1,299 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from urllib.error import HTTPError
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+spec = importlib.util.spec_from_file_location(
+    "control_plane", ROOT / "server" / "control_plane.py"
+)
+control_plane = importlib.util.module_from_spec(spec)
+assert spec.loader
+spec.loader.exec_module(control_plane)
+
+
+class FlowTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp = tempfile.TemporaryDirectory()
+        self.admin = "admin-" + "a" * 32
+        self.owner = "owner-" + "o" * 32
+        self.persona = "persona-" + "p" * 32
+        self.operator = "operator-" + "z" * 32
+        app = control_plane.ControlPlane(
+            os.path.join(self.temp.name, "control.sqlite3"),
+            self.admin,
+            self.owner,
+            self.persona,
+            self.operator,
+            "signing-" + "s" * 32,
+        )
+        self.app = app
+        self.server = control_plane.AppServer(("127.0.0.1", 0), app)
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        self.base = f"http://127.0.0.1:{self.server.server_port}"
+
+    def tearDown(self) -> None:
+        self.server.shutdown()
+        self.server.server_close()
+        self.temp.cleanup()
+
+    def api(self, method, path, token=None, body=None):
+        data = None if body is None else json.dumps(body).encode()
+        headers = {"Accept": "application/json"}
+        if token:
+            headers["Authorization"] = f"Bearer {token}"
+        if data is not None:
+            headers["Content-Type"] = "application/json"
+        with urlopen(
+            Request(self.base + path, data=data, headers=headers, method=method)
+        ) as response:
+            return json.load(response)
+
+    def test_pair_approve_execute_receipt(self) -> None:
+        pairing = self.api("POST", "/v1/pairings", self.admin, {})
+        node = self.api(
+            "POST",
+            "/v1/pairings/claim",
+            body={
+                "code": pairing["code"],
+                "node_name": "canger-home-ubuntu",
+                "agent_version": "test",
+                "policy": {
+                    "allowed_roots": ["/srv/canger"],
+                    "enabled_actions": [
+                        "system.status",
+                        "repo.status",
+                        "command.run",
+                    ],
+                },
+            },
+        )
+
+        with self.assertRaises(HTTPError) as reused:
+            self.api(
+                "POST",
+                "/v1/pairings/claim",
+                body={
+                    "code": pairing["code"],
+                    "node_name": "attacker",
+                    "policy": {
+                        "allowed_roots": ["/srv/canger"],
+                        "enabled_actions": ["system.status"],
+                    },
+                },
+            )
+        self.assertEqual(reused.exception.code, 401)
+
+        created = self.api(
+            "POST",
+            "/v1/tasks",
+            self.operator,
+            {
+                "node_id": node["node_id"],
+                "action": "system.status",
+                "args": {},
+                "requester_id": "ICE-GL-ZY001",
+                "requester_name": "铸渊",
+            },
+        )
+        waiting = self.api(
+            "GET",
+            f"/v1/nodes/{node['node_id']}/tasks/next",
+            node["node_token"],
+        )
+        self.assertIsNone(waiting["task"])
+
+        self.api(
+            "POST",
+            f"/v1/tasks/{created['task_id']}/approve",
+            self.owner,
+            {},
+        )
+        leased = self.api(
+            "GET",
+            f"/v1/nodes/{node['node_id']}/tasks/next",
+            node["node_token"],
+        )
+        self.assertEqual(leased["task"]["action"], "system.status")
+
+        event = self.api(
+            "POST",
+            f"/v1/tasks/{created['task_id']}/events",
+            node["node_token"],
+            {"kind": "started", "payload": {"agent_version": "test"}},
+        )
+        self.assertGreater(event["seq"], 0)
+        events = self.api(
+            "GET",
+            f"/v1/tasks/{created['task_id']}/events?after=0",
+            self.persona,
+        )
+        self.assertEqual(events["events"][0]["kind"], "started")
+
+        self.api(
+            "POST",
+            f"/v1/tasks/{created['task_id']}/result",
+            node["node_token"],
+            {"ok": True, "output": "healthy"},
+        )
+        receipt = self.api(
+            "GET", f"/v1/tasks/{created['task_id']}", self.operator
+        )
+        self.assertEqual(receipt["status"], "completed")
+        self.assertEqual(receipt["result"]["output"], "healthy")
+
+    def test_time_limited_grant_auto_approves_safe_task(self) -> None:
+        pairing = self.api("POST", "/v1/pairings", self.admin, {})
+        node = self.api(
+            "POST",
+            "/v1/pairings/claim",
+            body={
+                "code": pairing["code"],
+                "node_name": "canger-grant-test",
+                "agent_version": "test",
+                "policy": {
+                    "allowed_roots": ["/srv/canger"],
+                    "enabled_actions": ["system.status", "repo.status"],
+                },
+            },
+        )
+        grant = self.api(
+            "POST",
+            "/v1/grants",
+            self.operator,
+            {
+                "node_id": node["node_id"],
+                "duration_seconds": 3600,
+                "reason": "测试标准开发会话",
+                "requester_id": "ICE-GL-ZY001",
+                "requester_name": "铸渊",
+            },
+        )
+        link_expiry = control_plane.now_ts() + 1800
+        token = self.app.grant_token(grant["grant_id"], link_expiry)
+        form = urlencode({"id": grant["grant_id"], "token": token}).encode()
+        request = Request(
+            self.base + "/approve/grant",
+            data=form,
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+            method="POST",
+        )
+        with urlopen(request) as response:
+            self.assertEqual(response.status, 200)
+
+        task = self.api(
+            "POST",
+            "/v1/tasks",
+            self.operator,
+            {
+                "node_id": node["node_id"],
+                "action": "repo.status",
+                "args": {"repo": "/srv/canger/project"},
+                "requester_id": "ICE-GL-ZY001",
+                "requester_name": "铸渊",
+            },
+        )
+        self.assertEqual(task["status"], "approved")
+        self.assertFalse(task["approval_required"])
+        self.assertEqual(task["grant_id"], grant["grant_id"])
+
+    def test_persona_can_request_but_cannot_approve_its_own_grant(self) -> None:
+        pairing = self.api("POST", "/v1/pairings", self.admin, {})
+        node = self.api(
+            "POST",
+            "/v1/pairings/claim",
+            body={
+                "code": pairing["code"],
+                "node_name": "canger-persona-request-test",
+                "agent_version": "test",
+                "policy": {
+                    "allowed_roots": ["/srv/canger"],
+                    "enabled_actions": ["system.status", "repo.status"],
+                },
+            },
+        )
+        grant = self.api(
+            "POST",
+            "/v1/grants",
+            self.persona,
+            {
+                "node_id": node["node_id"],
+                "duration_seconds": 4 * 3600,
+                "reason": "检查苍耳仓库状态",
+                "requester_id": "PTS-VA-001-EED",
+                "requester_name": "耳耳蛋",
+            },
+        )
+        self.assertEqual(grant["status"], "pending")
+        self.assertEqual(grant["requested_by"], "PTS-VA-001-EED")
+        self.assertEqual(grant["requester_name"], "耳耳蛋")
+        self.assertEqual(grant["notification"], "not_configured")
+        self.assertEqual(grant["roots"], ["/srv/canger"])
+
+        with self.assertRaises(HTTPError) as unauthorized:
+            self.api(
+                "POST",
+                f"/v1/tasks/{grant['grant_id']}/approve",
+                self.persona,
+                {},
+            )
+        self.assertEqual(unauthorized.exception.code, 401)
+
+        pending_task = self.api(
+            "POST",
+            "/v1/tasks",
+            self.persona,
+            {
+                "node_id": node["node_id"],
+                "action": "system.status",
+                "args": {},
+                "requester_id": "PTS-VA-001-EED",
+                "requester_name": "耳耳蛋",
+            },
+        )
+        self.assertEqual(pending_task["status"], "pending")
+        self.assertEqual(pending_task["requested_by"], "PTS-VA-001-EED")
+
+        link_expiry = control_plane.now_ts() + 1800
+        token = self.app.grant_token(grant["grant_id"], link_expiry)
+        form = urlencode({"id": grant["grant_id"], "token": token}).encode()
+        approval = Request(
+            self.base + "/approve/grant",
+            data=form,
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+            method="POST",
+        )
+        with urlopen(approval) as response:
+            self.assertEqual(response.status, 200)
+
+        approved_task = self.api(
+            "POST",
+            "/v1/tasks",
+            self.persona,
+            {
+                "node_id": node["node_id"],
+                "action": "repo.status",
+                "args": {"repo": "/srv/canger/project"},
+                "requester_id": "PTS-VA-001-EED",
+                "requester_name": "耳耳蛋",
+            },
+        )
+        self.assertEqual(approved_task["status"], "approved")
+        self.assertEqual(approved_task["requested_by"], "PTS-VA-001-EED")
+        self.assertEqual(approved_task["grant_id"], grant["grant_id"])
+
+
+if __name__ == "__main__":
+    unittest.main()

remote-node/tools/canger_persona.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""Repository entry for Canger's persona to request and use approved access."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import time
+from pathlib import Path
+from typing import Any
+from urllib.error import HTTPError, URLError
+
+from cangerctl import request
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_PROFILE = ROOT / "config" / "pufferfish-persona.json"
+
+
+def load_profile(path: str) -> dict[str, Any]:
+    with open(path, encoding="utf-8") as handle:
+        profile = json.load(handle)
+    required = {
+        "persona_id",
+        "human_owner",
+        "node_id",
+        "control_url",
+        "default_hours",
+        "minimum_hours",
+        "maximum_hours",
+    }
+    missing = sorted(required - set(profile))
+    if missing:
+        raise RuntimeError("persona profile is missing: " + ", ".join(missing))
+    return profile
+
+
+def persona_token() -> str:
+    value = os.environ.get("CANGER_PERSONA_TOKEN", "")
+    if len(value) < 24:
+        raise RuntimeError(
+            "CANGER_PERSONA_TOKEN is unavailable; run from the registered "
+            "Singapore persona environment"
+        )
+    return value
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="苍耳人格体的受控运维申请与执行入口"
+    )
+    parser.add_argument("--profile", default=str(DEFAULT_PROFILE))
+    parser.add_argument("--url")
+    parser.add_argument(
+        "--requester-name",
+        default=os.environ.get("GUANGHU_PERSONA_NAME", ""),
+        help="本次实际申请者名称；也可用 GUANGHU_PERSONA_NAME",
+    )
+    parser.add_argument(
+        "--requester-id",
+        default=os.environ.get("GUANGHU_PERSONA_ID", ""),
+        help="本次实际申请者正式编号；也可用 GUANGHU_PERSONA_ID",
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    request_access = sub.add_parser("request-access")
+    request_access.add_argument("--hours", type=int)
+    request_access.add_argument("--reason", required=True)
+
+    task = sub.add_parser("task")
+    task.add_argument("--action", required=True)
+    task.add_argument("--args", default="{}")
+
+    show = sub.add_parser("show")
+    show.add_argument("task_id")
+
+    watch = sub.add_parser("watch")
+    watch.add_argument("task_id")
+
+    sub.add_parser("grants")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    profile = load_profile(args.profile)
+    base = args.url or os.environ.get("CANGER_CONTROL_URL") or profile["control_url"]
+    token = persona_token()
+    node_id = profile["node_id"]
+    if args.command in {"request-access", "task"}:
+        if not args.requester_name or not args.requester_id:
+            raise RuntimeError(
+                "requester name and id are required; provide --requester-name "
+                "and --requester-id, or GUANGHU_PERSONA_NAME/GUANGHU_PERSONA_ID"
+            )
+
+    if args.command == "request-access":
+        hours = args.hours or int(profile["default_hours"])
+        minimum = int(profile["minimum_hours"])
+        maximum = int(profile["maximum_hours"])
+        if not minimum <= hours <= maximum:
+            raise RuntimeError(f"hours must be between {minimum} and {maximum}")
+        result = request(
+            base,
+            token,
+            "POST",
+            "/v1/grants",
+            {
+                "node_id": node_id,
+                "duration_seconds": hours * 3600,
+                "reason": args.reason,
+                "requester_name": args.requester_name,
+                "requester_id": args.requester_id,
+            },
+        )
+    elif args.command == "task":
+        result = request(
+            base,
+            token,
+            "POST",
+            "/v1/tasks",
+            {
+                "node_id": node_id,
+                "action": args.action,
+                "args": json.loads(args.args),
+                "requester_name": args.requester_name,
+                "requester_id": args.requester_id,
+            },
+        )
+    elif args.command == "show":
+        result = request(
+            base,
+            token,
+            "GET",
+            f"/v1/tasks/{args.task_id}",
+        )
+    elif args.command == "grants":
+        result = request(base, token, "GET", "/v1/grants")
+    else:
+        after = 0
+        while True:
+            events = request(
+                base,
+                token,
+                "GET",
+                f"/v1/tasks/{args.task_id}/events?after={after}",
+            )
+            for event in events["events"]:
+                after = max(after, event["seq"])
+                print(
+                    f"[{event['seq']}] {event['kind']}: "
+                    + json.dumps(event["payload"], ensure_ascii=False),
+                    flush=True,
+                )
+            state = request(
+                base,
+                token,
+                "GET",
+                f"/v1/tasks/{args.task_id}",
+            )
+            if state["status"] in {"completed", "rejected"}:
+                result = state
+                break
+            time.sleep(1)
+
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except HTTPError as exc:
+        print(f"error: control plane returned HTTP {exc.code}", file=sys.stderr)
+        raise SystemExit(2)
+    except URLError as exc:
+        print(f"error: cannot reach control plane: {exc.reason}", file=sys.stderr)
+        raise SystemExit(2)
+    except (RuntimeError, ValueError, json.JSONDecodeError, OSError) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        raise SystemExit(2)

remote-node/tools/cangerctl.py

@@ -0,0 +1,168 @@
+#!/usr/bin/env python3
+"""Small CLI used by the persona and owner on the Singapore server."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import time
+from typing import Any
+from urllib.error import HTTPError
+from urllib.request import Request, urlopen
+
+
+def request(
+    base: str, token: str, method: str, path: str, body: dict[str, Any] | None = None
+) -> dict[str, Any]:
+    data = None if body is None else json.dumps(body).encode("utf-8")
+    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = Request(base.rstrip("/") + path, data=data, headers=headers, method=method)
+    try:
+        with urlopen(req, timeout=30) as response:
+            return json.load(response)
+    except HTTPError as exc:
+        print(exc.read().decode("utf-8", errors="replace"), file=sys.stderr)
+        raise
+
+
+def token_for(role: str) -> str:
+    name = {
+        "admin": "CANGER_ADMIN_TOKEN",
+        "owner": "CANGER_OWNER_TOKEN",
+        "persona": "CANGER_PERSONA_TOKEN",
+        "operator": "CANGER_OPERATOR_TOKEN",
+    }[role]
+    value = os.environ.get(name, "")
+    if not value:
+        raise RuntimeError(f"{name} is not set")
+    return value
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--url", default=os.environ.get("CANGER_CONTROL_URL", "http://127.0.0.1:8787")
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+    sub.add_parser("pair-code")
+    sub.add_parser("nodes")
+    task = sub.add_parser("task")
+    task.add_argument("--node", required=True)
+    task.add_argument("--action", required=True)
+    task.add_argument("--args", default="{}")
+    approve = sub.add_parser("approve")
+    approve.add_argument("task_id")
+    reject = sub.add_parser("reject")
+    reject.add_argument("task_id")
+    show = sub.add_parser("show")
+    show.add_argument("task_id")
+    watch = sub.add_parser("watch")
+    watch.add_argument("task_id")
+    grant = sub.add_parser("grant")
+    grant.add_argument("--node", required=True)
+    grant.add_argument("--hours", type=int, default=4)
+    grant.add_argument("--reason", default="苍耳人格体编程、调试与诊断")
+    sub.add_parser("grants")
+    revoke = sub.add_parser("revoke")
+    revoke.add_argument("grant_id")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    if args.command == "pair-code":
+        result = request(args.url, token_for("admin"), "POST", "/v1/pairings", {})
+    elif args.command == "nodes":
+        result = request(args.url, token_for("operator"), "GET", "/v1/nodes")
+    elif args.command == "task":
+        result = request(
+            args.url,
+            token_for("operator"),
+            "POST",
+            "/v1/tasks",
+            {
+                "node_id": args.node,
+                "action": args.action,
+                "args": json.loads(args.args),
+            },
+        )
+    elif args.command == "approve":
+        result = request(
+            args.url,
+            token_for("owner"),
+            "POST",
+            f"/v1/tasks/{args.task_id}/approve",
+            {},
+        )
+    elif args.command == "reject":
+        result = request(
+            args.url,
+            token_for("owner"),
+            "POST",
+            f"/v1/tasks/{args.task_id}/reject",
+            {},
+        )
+    elif args.command == "show":
+        result = request(
+            args.url,
+            token_for("operator"),
+            "GET",
+            f"/v1/tasks/{args.task_id}",
+        )
+    elif args.command == "watch":
+        after = 0
+        while True:
+            events = request(
+                args.url,
+                token_for("operator"),
+                "GET",
+                f"/v1/tasks/{args.task_id}/events?after={after}",
+            )
+            for event in events["events"]:
+                after = max(after, event["seq"])
+                print(
+                    f"[{event['seq']}] {event['kind']}: "
+                    + json.dumps(event["payload"], ensure_ascii=False),
+                    flush=True,
+                )
+            state = request(
+                args.url,
+                token_for("operator"),
+                "GET",
+                f"/v1/tasks/{args.task_id}",
+            )
+            if state["status"] in {"completed", "rejected"}:
+                result = state
+                break
+            time.sleep(1)
+    elif args.command == "grant":
+        result = request(
+            args.url,
+            token_for("operator"),
+            "POST",
+            "/v1/grants",
+            {
+                "node_id": args.node,
+                "duration_seconds": args.hours * 3600,
+                "reason": args.reason,
+            },
+        )
+    elif args.command == "grants":
+        result = request(args.url, token_for("operator"), "GET", "/v1/grants")
+    else:
+        result = request(
+            args.url,
+            token_for("owner"),
+            "POST",
+            f"/v1/grants/{args.grant_id}/revoke",
+            {},
+        )
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()