chore(worktree): ignore registered child worktrees

.agent/dispatch/V2R1-PERSONAL-WEIXIN-IMMEDIATE-WORK-UNITS.json

@@ -1,32 +0,0 @@
-{
-  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
-  "created_at": "2026-07-17T14:13:38+00:00",
-  "schema_version": 1,
-  "state_writer": "openclaw",
-  "work_units": [
-    {
-      "fallback": "codex",
-      "gate": "R0_RUNNER_LIVE_RECOVERY",
-      "id": "R0-DISCOVER-SECRET-PROVIDER",
-      "primary": "openclaw",
-      "status": "READY",
-      "track": "R_RUNNER"
-    },
-    {
-      "fallback": "codex",
-      "gate": "P2_REAL_B3_EVIDENCE",
-      "id": "P2-INVENTORY-REAL-EVIDENCE",
-      "primary": "openclaw",
-      "status": "READY",
-      "track": "P_PERSONAL_WEIXIN"
-    },
-    {
-      "fallback": "openclaw",
-      "gate": "P1_REPAIR3_CODE",
-      "id": "P1-POST-CONTRACT-REACK",
-      "primary": "codex",
-      "status": "READY",
-      "track": "P_PERSONAL_WEIXIN"
-    }
-  ]
-}

.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml

@@ -1,157 +0,0 @@
-{
-  "adaptive_dispatch": {
-    "codex_first": [
-      "complex_code",
-      "security_boundary",
-      "architecture",
-      "high_risk_review",
-      "strong_model_materially_better"
-    ],
-    "default_orchestrator": "openclaw",
-    "dual_required": [
-      "release_gate",
-      "crypto_or_identity_change",
-      "production_migration_or_rollback"
-    ],
-    "max_attempts_per_work_unit": 2,
-    "on_timeout": [
-      "mark_TIMEOUT",
-      "preserve_logs",
-      "switch_agent"
-    ],
-    "openclaw_first": [
-      "quality_close",
-      "large_or_repetitive",
-      "long_running",
-      "infra_or_git_inventory",
-      "cost_sensitive"
-    ],
-    "single_state_writer_lease_required": true,
-    "strict_role_partition": false,
-    "timeout_can_pass": false
-  },
-  "authoritative_state": {
-    "b3": "BLOCKED_EVIDENCE_REQUIRED",
-    "b4": "NOT_STARTED",
-    "baseline": "V1R4C6",
-    "production_release": "NOT_STARTED",
-    "release": "paused",
-    "repair3": {
-      "base": "8d9c6da90fc8caa69b691d243c6fefb40b6b0575",
-      "candidate": "PASS_CANDIDATE_PREPARED",
-      "commit": "e110cda9968cadc5c7b7179ec5003b05aad11ce4",
-      "exact4": "PASS",
-      "result": "49a9c2e0cc898ecf2bbd45f454e007644d7f3045"
-    },
-    "runner": {
-      "artifact_sha256": "85c472c05a1e55a5899b215d3cd41df53912787de529ac9cb3d5308da4158130",
-      "commit": "bbd27cf06acf1f589eef291eab54d4b127aa3aea",
-      "live": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
-      "offline": "PASS",
-      "result": "e789632e359e9232d8ae081e7c4f23ed7b615a9d"
-    }
-  },
-  "contract_id": "V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT",
-  "facts_policy": {
-    "accept": [
-      "committed_git_tree",
-      "full_oid",
-      "worktree_bare_gitea_oid_match",
-      "test_log_with_rc",
-      "immutable_digest",
-      "result_only_commit",
-      "live_evidence"
-    ],
-    "reject": [
-      "verbal_claim",
-      "readme_only",
-      "local_only_commit",
-      "environment_variable_as_secret_ref",
-      "compile_only",
-      "timeout_as_pass",
-      "unrelated_repo",
-      "fabricated_evidence"
-    ]
-  },
-  "forbidden_git": [
-    "reset",
-    "amend",
-    "rebase",
-    "force_push",
-    "direct_main_push",
-    "direct_bare_ref_mutation",
-    "unapproved_gitea_db_mutation",
-    "reuse_accident_worktree",
-    "delete_live_worktree_to_clear_collision"
-  ],
-  "generated_at": "2026-07-17T14:13:38+00:00",
-  "handoff": {
-    "ack": "CONTRACT_ACK=<full_contract_commit_oid>",
-    "advance": "independent_verification_required",
-    "all_agents_match": true
-  },
-  "join_gate": {
-    "requires": [
-      "RUNNER_RECOVERED",
-      "B3_PASS",
-      "B4_PASS",
-      "INTEGRATION_PASS"
-    ]
-  },
-  "model_policy": {
-    "codex": "strongest_model_actually_available_in_runtime",
-    "gpt_5_6": "use_only_if_exposed",
-    "no_unverified_model_claims": true,
-    "openclaw_default": "DeepSeek V4 Flash"
-  },
-  "owner_authorization": {
-    "adaptive_agent_assignment_authorized": true,
-    "agent_substitution_authorized": true,
-    "autonomous_execution": true,
-    "may_bypass_release_controls": false,
-    "may_bypass_secret_controls": false,
-    "may_fabricate_evidence": false,
-    "user_step_by_step_confirmation_required": false
-  },
-  "release_gates": [
-    "MAIN_MERGE",
-    "STAGING_AND_CANARY",
-    "PRODUCTION"
-  ],
-  "schema_version": 3,
-  "state_rules": [
-    "tracks_run_parallel",
-    "runner_block_does_not_block_offline_work_or_evidence_inventory",
-    "join_requires_both_tracks",
-    "failure_stops_dependents",
-    "production_remains_V1R4C6_paused_until_production_pass"
-  ],
-  "status": "ACTIVE",
-  "tracks": {
-    "P_PERSONAL_WEIXIN": {
-      "b3": "BLOCKED_EVIDENCE_REQUIRED",
-      "b4": "NOT_STARTED",
-      "parallel": true,
-      "repair3": "PASS_CANDIDATE_PREPARED",
-      "repair3_commit": "e110cda9968cadc5c7b7179ec5003b05aad11ce4"
-    },
-    "R_RUNNER": {
-      "commit": "bbd27cf06acf1f589eef291eab54d4b127aa3aea",
-      "current": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
-      "fallback": "codex",
-      "parallel": true,
-      "primary": "openclaw",
-      "required": [
-        "artifact_verified",
-        "approved_secret_provider_contract",
-        "owner_probe_verified",
-        "service_healthy",
-        "heartbeat_healthy",
-        "gateway_auth",
-        "durable_task_create_read_cancel",
-        "execution_channel",
-        "live_rollback"
-      ]
-    }
-  }
-}

.agent/state/V2R1-PERSONAL-WEIXIN-PRODUCTION-STATE.json

@@ -1,15 +0,0 @@
-{
-  "b3": "BLOCKED_EVIDENCE_REQUIRED",
-  "b4": "NOT_STARTED",
-  "baseline": "V1R4C6",
-  "contract_branch": "task/governance/v2r1-personal-weixin-production-contract-20260717",
-  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
-  "dispatch_commit": null,
-  "production_release": "NOT_STARTED",
-  "release": "paused",
-  "repair3": "PASS_CANDIDATE_PREPARED",
-  "runner": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
-  "schema_version": 1,
-  "single_state_writer": "openclaw",
-  "updated_at": "2026-07-17T14:13:38+00:00"
-}

.agent/tasks/codex/TASK-CODEX-V2R1-ADAPTIVE-PRODUCTION-EXECUTION-20260717.md

@@ -1,13 +0,0 @@
-# Codex V2R1 adaptive execution
-CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
-Read `.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml`.
-Execute P1-POST-CONTRACT-REACK now:
-- read-only verify Repair3 e110cda9968cadc5c7b7179ec5003b05aad11ce4
-- read-only verify RESULT 49a9c2e0cc898ecf2bbd45f454e007644d7f3045
-- verify lineage from 8d9c6da90fc8caa69b691d243c6fefb40b6b0575, exact4, tests, Gitleaks and RESULT-only scope
-- do not modify Repair3 code in this unit
-Act as fallback for OpenClaw only on timeout/unavailability. Use GPT-5.6 only if actually exposed; otherwise strongest visible model. Prefer OpenClaw when quality is comparable or volume is high; prefer Codex for complex/security-critical work.
-Forbidden: reset, amend, rebase, force push, direct main push, controller mutation, secret disclosure, fabricated evidence, skipped gates.
-Report:
-CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> TRACK=<id> GATE=<id> AGENT=codex STATUS=<status> IMPLEMENTATION=<oid-or-NONE> RESULT=<oid-or-NONE> OID_MATCH=<YES-or-NO> BLOCKER=<reason-or-NONE>
-Keep BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED until production gate passes.

.agent/tasks/openclaw/TASK-GITOPS-ISOLATED-WORKTREE-MIGRATION-20260714T233742UTC8.md

@@ -0,0 +1,267 @@
+# 执行 GitOps 独立 Worktree 与统一运行身份迁移
+
+```yaml
+task_id: TASK-GITOPS-ISOLATED-WORKTREE-MIGRATION-20260714T233742UTC8
+created_at: 2026-07-14T23:37:42+08:00
+timezone: Asia/Shanghai
+owner_approved: true
+user_action_required: false
+risk: high
+production_business_change: false
+```
+
+## 目标
+
+根治共享 worktree 与多身份写入导致的 ACL、checkout 和未跟踪文件覆盖问题。
+
+实施：
+
+1. 统一 GitOps/OpenClaw 写入身份为 `aiman:gitops`；
+2. 统一 `UMask=0002`；
+3. 为 GitOps 根目录设置 setgid 与默认 ACL；
+4. 建立 bare repo 和每任务独立 worktree；
+5. 启用任务启动前自检；
+6. 验证 OpenClaw 与 ChatGPT 并发任务互不干扰；
+7. 保留原共享 workspace 作为只读回滚源，禁止智能体继续在其中切换分支。
+
+## 强制通知
+
+收到任务后立即微信通知 Owner：
+
+```text
+【任务已接收】GitOps独立worktree与统一运行身份迁移开始。将暂停写入服务、建立独立任务工作树并验证并发执行；生产业务版本不变。
+```
+
+任务结束后无论 PASS、BLOCKED 或 ROLLED_BACK，立即微信汇报结果并记录 message_id。
+
+## 1. 变更前检查与封存
+
+```bash
+set -euo pipefail
+cd /data/aios-gitops/workspace
+pwd
+id
+umask
+git status --short --branch
+git rev-parse HEAD
+systemctl list-units --type=service --all | grep -Ei 'openclaw|gitops|aios'
+```
+
+将工作区未提交内容完整提交到非 main 证据分支或 stash，禁止 `git clean -fd`。
+
+记录当前运行服务、用户、组、umask、PID、工作区 HEAD 和未跟踪文件清单。
+
+## 2. 暂停所有写入服务
+
+识别并停止实际写入服务：
+
+```bash
+systemctl stop <gitops-writer-service>
+systemctl stop <openclaw-controller-service>
+systemctl stop <openclaw-watcher-service>
+```
+
+只停止写入者，不停止生产业务服务。记录每条命令和退出码。
+
+## 3. 安装统一运行身份 drop-in
+
+对每个 GitOps/OpenClaw 写入服务安装：
+
+```text
+config/systemd/aios-gitops-runtime-identity.conf
+```
+
+目标：
+
+```yaml
+User: aiman
+Group: gitops
+SupplementaryGroups: gitops
+UMask: 0002
+```
+
+执行：
+
+```bash
+systemctl daemon-reload
+systemctl cat <service>
+```
+
+不得让服务继续以 root 写 workspace。
+
+## 4. 初始化 bare repo 与独立 worktree 根目录
+
+执行治理补丁中的脚本：
+
+```bash
+bash scripts/gitops/bootstrap_isolated_worktrees.sh
+```
+
+验收：
+
+```yaml
+bare_repo: /data/aios-gitops/repo.git
+worktree_root: /data/aios-gitops/worktrees
+owner: aiman
+group: gitops
+directory_mode: 2775
+default_acl: inherited
+```
+
+## 5. 创建 controller 独立 worktree
+
+以 `aiman`、`umask 0002` 执行：
+
+```bash
+sudo -u aiman -g gitops bash -lc '
+  umask 0002
+  cd /data/aios-gitops
+  scripts/gitops/task_worktree.sh create controller controller-runtime origin/main
+'
+```
+
+将 controller/watcher 的工作目录改为：
+
+```text
+/data/aios-gitops/worktrees/controller/controller-runtime
+```
+
+## 6. 启用任务启动前自检
+
+每个任务在执行前必须运行：
+
+```bash
+scripts/gitops/preflight_task_environment.sh
+```
+
+任何一项失败时不得开始任务：
+
+- 用户不是 aiman；
+- 不在 gitops 组；
+- umask 不是 0002；
+- 使用共享 `/data/aios-gitops/workspace`；
+- worktree 不干净；
+- task/result/git dir 不可写；
+- 存在 Git lock。
+
+## 7. 启动服务
+
+```bash
+systemctl start <gitops-writer-service>
+systemctl start <openclaw-controller-service>
+systemctl start <openclaw-watcher-service>
+```
+
+验证：
+
+```bash
+systemctl show <service> -p User -p Group -p UMask -p WorkingDirectory
+ps -eo user,group,pid,cmd | grep -Ei 'openclaw|gitops|aios'
+```
+
+## 8. 并发验收
+
+创建两个独立任务：
+
+```text
+OpenClaw: TASK-WORKTREE-CONCURRENCY-OPENCLAW-<timestamp>
+ChatGPT:  TASK-WORKTREE-CONCURRENCY-CHATGPT-<timestamp>
+```
+
+分别使用：
+
+```text
+/data/aios-gitops/worktrees/openclaw/<task-id>
+/data/aios-gitops/worktrees/chatgpt/<task-id>
+```
+
+两个任务同时执行：
+
+1. 写各自 `.agent/tasks/<agent>/`；
+2. 写各自 `.agent/results/<agent>/`；
+3. 创建深层子目录，验证默认 ACL；
+4. commit；
+5. push；
+6. 不切换共享 workspace 分支。
+
+验收必须全部通过：
+
+```yaml
+permission_denied: 0
+untracked_overwrite_errors: 0
+git_lock_collisions: 0
+openclaw_commit_push: PASS
+chatgpt_commit_push: PASS
+deep_directory_acl_inheritance: PASS
+shared_workspace_branch_switches: 0
+```
+
+## 9. 回滚
+
+若迁移失败：
+
+1. 停止新服务；
+2. 保留失败 worktree 供审计；
+3. 恢复 systemd drop-in 备份；
+4. `daemon-reload`；
+5. 恢复原服务状态；
+6. 不删除 `/data/aios-gitops/workspace`；
+7. 生产业务保持不变。
+
+## 10. 结果文件
+
+提交：
+
+```text
+.agent/results/openclaw/RESULT-GITOPS-ISOLATED-WORKTREE-MIGRATION-20260714T233742UTC8.md
+```
+
+首段必须包含：
+
+```yaml
+status: PASS | BLOCKED | ROLLED_BACK
+task_id: TASK-GITOPS-ISOLATED-WORKTREE-MIGRATION-20260714T233742UTC8
+created_at: 2026-07-14T23:37:42+08:00
+timezone: Asia/Shanghai
+user_action_required: false
+runtime_identity:
+  user:
+  group:
+  umask:
+bare_repo:
+worktree_root:
+controller_worktree:
+services:
+default_acl:
+preflight:
+concurrency_test:
+  openclaw_branch:
+  openclaw_commit:
+  chatgpt_branch:
+  chatgpt_commit:
+  permission_denied_count:
+  untracked_overwrite_count:
+  git_lock_collision_count:
+owner_notifications:
+  received_message_id:
+  terminal_message_id:
+rollback:
+remaining_risks:
+next_step:
+```
+
+提交并推送到独立执行分支。禁止直接推 main。
+
+## 完成微信模板
+
+PASS：
+
+```text
+【任务完成】GitOps独立worktree迁移已通过。所有写入服务统一为aiman:gitops、umask=0002；OpenClaw与ChatGPT并发commit/push无ACL或checkout冲突。生产业务版本未变化。
+```
+
+阻塞：
+
+```text
+【任务阻塞】GitOps独立worktree迁移未通过。原因=<具体服务/权限/并发验收>；已保留原workspace和回滚能力，生产业务未变化。
+```

.agent/tasks/openclaw/TASK-OPENCLAW-V2R1-ADAPTIVE-PRODUCTION-ORCHESTRATION-20260717.md

@@ -1,13 +0,0 @@
-# OpenClaw V2R1 adaptive orchestration
-CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
-Read `.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml`.
-You are default orchestrator and sole authoritative state writer.
-Execute immediately:
-1. R0-DISCOVER-SECRET-PROVIDER: discover approved provider metadata and owner-probe source without reading or outputting secrets. Environment variables alone are not approved secret refs.
-2. P2-INVENTORY-REAL-EVIDENCE: inventory approved restricted B3 evidence locations; validate existing facts; never synthesize evidence.
-3. Independently verify Codex results before advancing.
-Prefer OpenClaw for comparable-quality, large, repetitive, long-running, infrastructure and cost-sensitive work. Prefer Codex strongest actually available model for complex code, security boundaries, architecture and high-risk review. On timeout mark TIMEOUT and switch agent; never infer PASS.
-Forbidden: reset, amend, rebase, force push, direct main push, controller mutation, secret disclosure, fabricated evidence, skipped gates.
-Report:
-CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> TRACK=<id> GATE=<id> AGENT=openclaw STATUS=<status> COMMIT=<oid-or-NONE> BLOCKER=<reason-or-NONE>
-Keep BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED until production gate passes.

.gitignore

@@ -11,3 +11,8 @@ runtime/tmp/
 runtime/cache/
 logs/
 config/test_write_perms
+
+# BEGIN AIOS REGISTERED CHILD WORKTREES (V2R1C1 2026-07-21)
+/TASK-WORKTREE-CONCURRENCY-OPENCLAW-20260715T060254UTC8/
+/TASK-WORKTREE-CONCURRENCY-OPENCLAW-20260715T061051UTC8/
+# END AIOS REGISTERED CHILD WORKTREES (V2R1C1 2026-07-21)

config/gitops/worktree-governance.yaml

@@ -0,0 +1,59 @@
+version: 1
+policy_id: gitops-isolated-worktree-governance-20260714T232948UTC8
+
+runtime_identity:
+  user: aiman
+  group: gitops
+  umask: "0002"
+  prohibit_root_workspace_writes: true
+
+repository:
+  source_worktree: /data/aios-gitops/workspace
+  bare_repo: /data/aios-gitops/repo.git
+  worktree_root: /data/aios-gitops/worktrees
+  remote: origin
+
+worktrees:
+  controller: /data/aios-gitops/worktrees/controller
+  task_pattern: /data/aios-gitops/worktrees/{agent}/{task_id}
+  allowed_agents:
+    - openclaw
+    - chatgpt
+    - controller
+  require_clean_start: true
+  remove_after_successful_push: true
+  retain_on_failure: true
+
+permissions:
+  owner: aiman
+  group: gitops
+  directory_mode: "2775"
+  file_mode: "0664"
+  default_acl:
+    user:aiman: rwx
+    group:gitops: rwx
+  apply_to:
+    - /data/aios-gitops
+    - /data/aios-gitops/worktrees
+
+preflight:
+  required_user: aiman
+  required_group: gitops
+  required_umask: "0002"
+  require_writable_git_dir: true
+  require_writable_task_dir: true
+  require_writable_result_dir: true
+  reject_shared_source_worktree: true
+  reject_existing_git_lock: true
+
+notifications:
+  owner_on_received: required
+  owner_on_terminal: required
+  preferred_channel: weixin
+  redact_secrets: true
+
+rollback:
+  preserve_source_worktree: true
+  remove_created_worktrees: true
+  restore_service_dropins: true
+  restore_service_state: true

config/systemd/aios-gitops-runtime-identity.conf

@@ -0,0 +1,14 @@
+# Install as a systemd drop-in for GitOps/OpenClaw writer services.
+# Example: /etc/systemd/system/<service>.service.d/10-runtime-identity.conf
+
+[Service]
+User=aiman
+Group=gitops
+SupplementaryGroups=gitops
+UMask=0002
+Environment=AIOS_GITOPS_ROOT=/data/aios-gitops
+Environment=AIOS_BARE_REPO=/data/aios-gitops/repo.git
+Environment=AIOS_WORKTREE_ROOT=/data/aios-gitops/worktrees
+Environment=AIOS_RUNTIME_USER=aiman
+Environment=AIOS_RUNTIME_GROUP=gitops
+Environment=AIOS_RUNTIME_UMASK=0002

evidence/OPENCLAW_TASK-CODEX-GITOPS-SECRET-ACL-AND-IDENTITY-HARDENING-20260715T004818UTC8-FINAL.md

@@ -0,0 +1 @@
+TASK-CODEX-GITOPS-SECRET-ACL-AND-IDENTITY-HARDENING-20260715T004818UTC8-FINAL

evidence/OPENCLAW_TASK-CODEX-GITOPS-SECRET-ACL-AND-IDENTITY-HARDENING-20260715T004818UTC8.md

@@ -0,0 +1 @@
+TASK-CODEX-GITOPS-SECRET-ACL-AND-IDENTITY-HARDENING-20260715T004818UTC8

patches/gitops-isolated-worktree-governance-20260714T232948UTC8.json

@@ -0,0 +1,9 @@
+{
+  "id": "patch-1784043007770-e97206c3",
+  "patch_name": "gitops-isolated-worktree-governance-20260714T232948UTC8",
+  "branch": "patch/gitops-isolated-worktree-governance-20260714T232948UTC8",
+  "risk": "high",
+  "description": "Root-cause remediation for recurring ACL and shared-worktree failures: unified aiman:gitops runtime identity, UMask=0002, inherited default ACLs, per-task isolated Git worktrees, startup preflight, migration/rollback scripts, systemd drop-in templates, audit evidence, and an OpenClaw execution task. No production business deployment in this patch.",
+  "created_by": "chatgpt-gitops-bot",
+  "created_at": "2026-07-14T15:30:07Z"
+}

patches/task_codex_gitops-isolated-worktree-migration-20260715T001652UTC8.branch.json

@@ -0,0 +1,9 @@
+{
+  "type": "branch",
+  "branch": "task/codex/gitops-isolated-worktree-migration-20260715T001652UTC8",
+  "base": "patch/gitops-isolated-worktree-governance-20260714T232948UTC8",
+  "created_by": "chatgpt-gitops-bot",
+  "created_at": "2026-07-14T16:17:28Z",
+  "direct_main_push": false,
+  "owner_approval_required": true
+}

reports/migration-openclaw-isolated-worktree-acceptance.md

@@ -0,0 +1,5 @@
+# OpenClaw isolated-worktree acceptance
+
+branch: `agent/openclaw/gitops-isolated-worktree-migration-20260715T001652UTC8`
+
+base: `b05eb1babe9c657e8776c09eec71302166902cc3`

scripts/gitops/bootstrap_isolated_worktrees.sh

@@ -0,0 +1,47 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT="${AIOS_GITOPS_ROOT:-/data/aios-gitops}"
+SOURCE="${AIOS_SOURCE_WORKTREE:-$ROOT/workspace}"
+BARE="${AIOS_BARE_REPO:-$ROOT/repo.git}"
+WORKTREES="${AIOS_WORKTREE_ROOT:-$ROOT/worktrees}"
+RUNTIME_USER="${AIOS_RUNTIME_USER:-aiman}"
+RUNTIME_GROUP="${AIOS_RUNTIME_GROUP:-gitops}"
+
+if [[ "$(id -u)" -ne 0 ]]; then
+  echo "ERROR: bootstrap must run as root" >&2
+  exit 2
+fi
+
+getent passwd "$RUNTIME_USER" >/dev/null
+getent group "$RUNTIME_GROUP" >/dev/null
+[[ -d "$SOURCE/.git" || -f "$SOURCE/.git" ]] || {
+  echo "ERROR: source worktree not found: $SOURCE" >&2
+  exit 3
+}
+
+install -d -o "$RUNTIME_USER" -g "$RUNTIME_GROUP" -m 2775 "$ROOT" "$WORKTREES"
+install -d -o "$RUNTIME_USER" -g "$RUNTIME_GROUP" -m 2775 \
+  "$WORKTREES/openclaw" "$WORKTREES/chatgpt" "$WORKTREES/controller"
+
+if [[ ! -d "$BARE" ]]; then
+  sudo -u "$RUNTIME_USER" -g "$RUNTIME_GROUP" git clone --bare "$SOURCE" "$BARE"
+fi
+
+chown -R "$RUNTIME_USER:$RUNTIME_GROUP" "$BARE" "$WORKTREES"
+find "$BARE" "$WORKTREES" -type d -exec chmod 2775 {} +
+find "$BARE" "$WORKTREES" -type f -exec chmod ug+rw {} +
+
+setfacl -R -m "u:${RUNTIME_USER}:rwX,g:${RUNTIME_GROUP}:rwX,m::rwX" "$BARE" "$WORKTREES"
+find "$BARE" "$WORKTREES" -type d -exec \
+  setfacl -m "d:u:${RUNTIME_USER}:rwx,d:g:${RUNTIME_GROUP}:rwx,d:m::rwx" {} +
+
+cat <<EOF
+bootstrap_status: PASS
+root: $ROOT
+source_worktree: $SOURCE
+bare_repo: $BARE
+worktree_root: $WORKTREES
+runtime_user: $RUNTIME_USER
+runtime_group: $RUNTIME_GROUP
+EOF

scripts/gitops/preflight_task_environment.sh

@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+EXPECTED_USER="${AIOS_RUNTIME_USER:-aiman}"
+EXPECTED_GROUP="${AIOS_RUNTIME_GROUP:-gitops}"
+EXPECTED_UMASK="${AIOS_RUNTIME_UMASK:-0002}"
+TASK_DIR="${AIOS_TASK_DIR:-.agent/tasks}"
+RESULT_DIR="${AIOS_RESULT_DIR:-.agent/results}"
+
+fail() {
+  echo "preflight_status: FAIL"
+  echo "reason: $1"
+  exit 1
+}
+
+USER_NOW="$(id -un)"
+GROUPS_NOW="$(id -Gn)"
+UMASK_NOW="$(umask)"
+ROOT="$(git rev-parse --show-toplevel 2>/dev/null || true)"
+GIT_DIR="$(git rev-parse --git-dir 2>/dev/null || true)"
+BRANCH="$(git branch --show-current 2>/dev/null || true)"
+
+[[ "$USER_NOW" == "$EXPECTED_USER" ]] || fail "runtime user is $USER_NOW, expected $EXPECTED_USER"
+[[ " $GROUPS_NOW " == *" $EXPECTED_GROUP "* ]] || fail "runtime user is not in $EXPECTED_GROUP"
+[[ "$UMASK_NOW" == "$EXPECTED_UMASK" || "$UMASK_NOW" == "002" ]] || fail "umask is $UMASK_NOW, expected $EXPECTED_UMASK"
+[[ -n "$ROOT" ]] || fail "not inside a git worktree"
+[[ "$ROOT" == /data/aios-gitops/worktrees/* ]] || fail "shared source worktree is prohibited: $ROOT"
+[[ -w "$ROOT" ]] || fail "worktree root is not writable"
+[[ -w "$GIT_DIR" ]] || fail "git dir is not writable: $GIT_DIR"
+[[ ! -e "$GIT_DIR/index.lock" ]] || fail "git index lock exists"
+install -d -m 2775 "$TASK_DIR" "$RESULT_DIR"
+[[ -w "$TASK_DIR" ]] || fail "task directory is not writable: $TASK_DIR"
+[[ -w "$RESULT_DIR" ]] || fail "result directory is not writable: $RESULT_DIR"
+[[ -z "$(git status --porcelain)" ]] || fail "worktree is not clean at task start"
+
+cat <<EOF
+preflight_status: PASS
+runtime_user: $USER_NOW
+runtime_groups: $GROUPS_NOW
+umask: $UMASK_NOW
+worktree: $ROOT
+git_dir: $GIT_DIR
+branch: $BRANCH
+workspace_clean: true
+task_path_writable: true
+result_path_writable: true
+git_lock_present: false
+EOF

scripts/gitops/task_worktree.sh

@@ -0,0 +1,73 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ACTION="${1:-}"
+AGENT="${2:-}"
+TASK_ID="${3:-}"
+BASE_REF="${4:-origin/main}"
+
+ROOT="${AIOS_GITOPS_ROOT:-/data/aios-gitops}"
+BARE="${AIOS_BARE_REPO:-$ROOT/repo.git}"
+WORKTREES="${AIOS_WORKTREE_ROOT:-$ROOT/worktrees}"
+RUNTIME_USER="${AIOS_RUNTIME_USER:-aiman}"
+RUNTIME_GROUP="${AIOS_RUNTIME_GROUP:-gitops}"
+
+usage() {
+  echo "usage: $0 create|remove <openclaw|chatgpt|controller> <task-id> [base-ref]" >&2
+  exit 64
+}
+
+[[ "$ACTION" =~ ^(create|remove)$ ]] || usage
+[[ "$AGENT" =~ ^(openclaw|chatgpt|controller)$ ]] || usage
+[[ "$TASK_ID" =~ ^[A-Za-z0-9._-]+$ ]] || {
+  echo "ERROR: invalid task id" >&2
+  exit 65
+}
+[[ "$(id -un)" == "$RUNTIME_USER" ]] || {
+  echo "ERROR: must run as $RUNTIME_USER" >&2
+  exit 66
+}
+[[ "$(id -gn)" == "$RUNTIME_GROUP" || " $(id -Gn) " == *" $RUNTIME_GROUP "* ]] || {
+  echo "ERROR: user must be in $RUNTIME_GROUP" >&2
+  exit 67
+}
+[[ "$(umask)" == "0002" || "$(umask)" == "002" ]] || {
+  echo "ERROR: umask must be 0002" >&2
+  exit 68
+}
+[[ -d "$BARE" ]] || {
+  echo "ERROR: bare repository missing: $BARE" >&2
+  exit 69
+}
+
+PATHNAME="$WORKTREES/$AGENT/$TASK_ID"
+BRANCH="task/$AGENT/$TASK_ID"
+
+case "$ACTION" in
+  create)
+    [[ ! -e "$PATHNAME" ]] || {
+      echo "ERROR: worktree already exists: $PATHNAME" >&2
+      exit 70
+    }
+    git --git-dir="$BARE" fetch --all --prune
+    git --git-dir="$BARE" worktree add -b "$BRANCH" "$PATHNAME" "$BASE_REF"
+    install -d -m 2775 "$PATHNAME/.agent/tasks/$AGENT" "$PATHNAME/.agent/results/$AGENT"
+    setfacl -m "u:${RUNTIME_USER}:rwx,g:${RUNTIME_GROUP}:rwx,m::rwx" "$PATHNAME"
+    find "$PATHNAME" -type d -exec chmod g+s {} +
+    find "$PATHNAME" -type d -exec \
+      setfacl -m "d:u:${RUNTIME_USER}:rwx,d:g:${RUNTIME_GROUP}:rwx,d:m::rwx" {} +
+    printf 'worktree: %s\nbranch: %s\nbase: %s\n' "$PATHNAME" "$BRANCH" "$BASE_REF"
+    ;;
+  remove)
+    [[ -d "$PATHNAME" ]] || {
+      echo "ERROR: worktree not found: $PATHNAME" >&2
+      exit 71
+    }
+    [[ -z "$(git -C "$PATHNAME" status --porcelain)" ]] || {
+      echo "ERROR: worktree is not clean; retain for investigation" >&2
+      exit 72
+    }
+    git --git-dir="$BARE" worktree remove "$PATHNAME"
+    printf 'removed: %s\n' "$PATHNAME"
+    ;;
+esac

scripts/wechat_send_backend.py

@@ -1,652 +0,0 @@
-#!/usr/bin/env python3
-"""Hardened WeChat outbound backend for AIOS.
-
-Only the dedicated UCC adapter is a supported send backend.  OpenClaw channel
-discovery is retained for diagnostics, but generic OpenClaw chat or agent
-delivery is never promoted to WeChat delivery.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import re
-import selectors
-import signal
-import subprocess
-import time
-import urllib.error
-import urllib.parse
-import urllib.request
-from typing import Any, Dict, List, Optional, Tuple
-
-
-OPENCLAW_BIN = os.environ.get("AIOS_OPENCLAW_BIN", "openclaw")
-WECHAT_ACCOUNT_ID = os.environ.get("AIOS_WECHAT_ACCOUNT_ID", "")
-WECHAT_CHANNEL = os.environ.get("AIOS_WECHAT_CHANNEL", "")
-SEND_BACKEND = os.environ.get("AIOS_WECHAT_SEND_BACKEND", "")
-UCC_ADAPTER_URL = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_URL", "http://127.0.0.1:18800")
-UCC_ADAPTER_TOKEN = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_TOKEN") or os.environ.get(
-    "AIOS_WECHAT_QR_ADAPTER_TOKEN", ""
-)
-UCC_SEND_PATH = os.environ.get("AIOS_UCC_WECHAT_SEND_PATH", "/api/internal/wechat/send")
-_allowed_origins = os.environ.get("AIOS_UCC_WECHAT_ALLOWED_ORIGINS", "").strip()
-UCC_ADAPTER_ALLOWED_ORIGINS = tuple(
-    origin.strip() for origin in _allowed_origins.split(",") if origin.strip()
-) or (
-    "http://127.0.0.1:18800",
-    "http://[::1]:18800",
-)
-
-MAX_TARGET_BYTES = 512
-MAX_CONTENT_BYTES = 32 * 1024
-MAX_IDENTIFIER_BYTES = 512
-MAX_METADATA_BYTES = 8 * 1024
-MAX_METADATA_DEPTH = 32
-MAX_METADATA_NODES = 1024
-MAX_RESPONSE_BYTES = 64 * 1024
-MAX_ENDPOINT_BYTES = 256
-_DEFAULT_TIMEOUT_SECONDS = 10
-_MAX_TIMEOUT_SECONDS = 30
-_LEGACY_SEND_PATH = "/internal/channels/wechat/send"
-_SAFE_PATH = re.compile(r"^/[A-Za-z0-9._~!$&'()*+,;=:@%/-]+$")
-_MAX_PATH_DECODE_ROUNDS = 4
-_OPENCLAW_TIMEOUT_RETURN_CODE = 124
-_OPENCLAW_OUTPUT_LIMIT_RETURN_CODE = 125
-_OPENCLAW_SPAWN_ERROR_RETURN_CODE = 126
-
-
-def _bounded_timeout(value: str) -> int:
-    try:
-        parsed = int(value)
-    except (TypeError, ValueError):
-        return _DEFAULT_TIMEOUT_SECONDS
-    return min(max(parsed, 1), _MAX_TIMEOUT_SECONDS)
-
-
-REQUEST_TIMEOUT_SECONDS = _bounded_timeout(os.environ.get("AIOS_UCC_WECHAT_TIMEOUT_SECONDS", "10"))
-
-
-class BackendResult:
-    def __init__(self, ok: bool, backend: str, status: str, data: Dict[str, Any]) -> None:
-        self.ok = ok
-        self.backend = backend
-        self.status = status
-        self.data = data
-
-    def to_dict(self) -> Dict[str, Any]:
-        return {"ok": self.ok, "backend": self.backend, "status": self.status, **self.data}
-
-
-class _BackendProtocolError(Exception):
-    """Internal marker whose details must never cross the module boundary."""
-
-
-def _openclaw_base_args() -> List[str]:
-    """Return the OpenClaw 2026.6.11-compatible command prefix.
-
-    Gateway/auth global flags are not supported by the installed CLI and would
-    also expose credentials through the process argument vector.
-    """
-
-    return [OPENCLAW_BIN]
-
-
-def _terminate_process_tree(process: Any) -> None:
-    if os.name == "posix":
-        try:
-            os.killpg(process.pid, signal.SIGTERM)
-        except OSError:
-            pass
-        try:
-            process.wait(timeout=0.5)
-        except (OSError, subprocess.TimeoutExpired):
-            pass
-        time.sleep(0.05)
-        try:
-            os.killpg(process.pid, signal.SIGKILL)
-        except OSError:
-            pass
-        try:
-            process.wait(timeout=0.5)
-        except (OSError, subprocess.TimeoutExpired):
-            pass
-        return
-    if process.poll() is not None:
-        return
-    try:
-        process.terminate()
-        process.wait(timeout=1)
-        return
-    except (OSError, subprocess.TimeoutExpired):
-        pass
-    try:
-        process.kill()
-        process.wait(timeout=1)
-    except (OSError, subprocess.TimeoutExpired):
-        pass
-
-
-def _run_openclaw(
-    args: List[str], timeout: int = 15
-) -> subprocess.CompletedProcess[str]:
-    bounded_timeout = min(max(int(timeout), 1), _MAX_TIMEOUT_SECONDS)
-    popen_options: Dict[str, Any] = {
-        "stdout": subprocess.PIPE,
-        "stderr": subprocess.PIPE,
-    }
-    if os.name == "posix":
-        popen_options["start_new_session"] = True
-    elif hasattr(subprocess, "CREATE_NEW_PROCESS_GROUP"):
-        popen_options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
-    try:
-        process = subprocess.Popen(args, **popen_options)
-    except (OSError, ValueError):
-        return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
-
-    selector = selectors.DefaultSelector()
-    streams = {"stdout": process.stdout, "stderr": process.stderr}
-    buffers = {"stdout": bytearray(), "stderr": bytearray()}
-    try:
-        for name, stream in streams.items():
-            if stream is not None:
-                selector.register(stream, selectors.EVENT_READ, name)
-    except (OSError, ValueError):
-        selector.close()
-        _terminate_process_tree(process)
-        for stream in streams.values():
-            if stream is not None:
-                stream.close()
-        return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
-
-    deadline = time.monotonic() + bounded_timeout
-    total_bytes = 0
-    return_code: Optional[int] = None
-    try:
-        while selector.get_map():
-            remaining = deadline - time.monotonic()
-            if remaining <= 0:
-                return_code = _OPENCLAW_TIMEOUT_RETURN_CODE
-                break
-            events = selector.select(min(remaining, 0.1))
-            for key, _mask in events:
-                allowance = MAX_RESPONSE_BYTES - total_bytes
-                try:
-                    chunk = os.read(key.fileobj.fileno(), min(8192, allowance + 1))
-                except OSError:
-                    return_code = _OPENCLAW_SPAWN_ERROR_RETURN_CODE
-                    break
-                if not chunk:
-                    selector.unregister(key.fileobj)
-                    key.fileobj.close()
-                    continue
-                if len(chunk) > allowance:
-                    return_code = _OPENCLAW_OUTPUT_LIMIT_RETURN_CODE
-                    break
-                buffers[key.data].extend(chunk)
-                total_bytes += len(chunk)
-            if return_code is not None:
-                break
-    finally:
-        selector.close()
-
-    if return_code is not None:
-        _terminate_process_tree(process)
-        for stream in streams.values():
-            if stream is not None and not stream.closed:
-                stream.close()
-        return subprocess.CompletedProcess(args, return_code, "", "")
-    try:
-        completed_return_code = process.wait(timeout=1)
-    except subprocess.TimeoutExpired:
-        _terminate_process_tree(process)
-        return subprocess.CompletedProcess(args, _OPENCLAW_TIMEOUT_RETURN_CODE, "", "")
-    stdout = buffers["stdout"].decode("utf-8", errors="replace")
-    stderr = buffers["stderr"].decode("utf-8", errors="replace")
-    return subprocess.CompletedProcess(args, completed_return_code, stdout, stderr)
-
-
-def _json_dict(text: str) -> Dict[str, Any]:
-    try:
-        value = json.loads(text or "{}")
-    except (RecursionError, TypeError, ValueError, json.JSONDecodeError):
-        return {}
-    return value if isinstance(value, dict) else {}
-
-
-def discover_openclaw_wechat_channel() -> Optional[str]:
-    """Discover an explicitly typed WeChat/Weixin channel for diagnostics."""
-
-    if WECHAT_CHANNEL:
-        return WECHAT_CHANNEL
-    try:
-        proc = _run_openclaw(_openclaw_base_args() + ["channels", "list", "--json"], timeout=15)
-    except (OSError, subprocess.SubprocessError, ValueError):
-        return None
-    if proc.returncode != 0:
-        return None
-    data = _json_dict(proc.stdout[:MAX_RESPONSE_BYTES])
-    candidates: List[str] = []
-
-    def walk(value: Any) -> None:
-        if isinstance(value, dict):
-            name = value.get("name") or value.get("channel") or value.get("id")
-            kind = value.get("type") or value.get("kind")
-            if isinstance(name, str) and isinstance(kind, str):
-                label = f"{name} {kind}".lower()
-                if "wechat" in label or "weixin" in label:
-                    candidates.append(name)
-            for child in value.values():
-                walk(child)
-        elif isinstance(value, list):
-            for child in value:
-                walk(child)
-
-    walk(data)
-    return candidates[0] if candidates else None
-
-
-def _auth_headers() -> Dict[str, str]:
-    headers = {"Content-Type": "application/json", "Accept": "application/json"}
-    if UCC_ADAPTER_TOKEN:
-        headers["Authorization"] = "Bearer " + UCC_ADAPTER_TOKEN
-    return headers
-
-
-def _safe_path(value: Any) -> Optional[str]:
-    if not isinstance(value, str):
-        return None
-    path = value if value.startswith("/") else "/" + value
-    try:
-        encoded_path = path.encode("utf-8")
-    except UnicodeEncodeError:
-        return None
-    if len(encoded_path) > MAX_ENDPOINT_BYTES:
-        return None
-    if not _SAFE_PATH.fullmatch(path):
-        return None
-    decoded = path
-    for _ in range(_MAX_PATH_DECODE_ROUNDS):
-        try:
-            next_decoded = urllib.parse.unquote_to_bytes(decoded).decode("utf-8")
-        except UnicodeDecodeError:
-            return None
-        if next_decoded == decoded:
-            break
-        decoded = next_decoded
-    if "%" in decoded or "\\" in decoded or "\x00" in decoded:
-        return None
-    if not _SAFE_PATH.fullmatch(decoded):
-        return None
-    if any(segment in {".", ".."} for segment in decoded.split("/")):
-        return None
-    return path
-
-
-def _unique_paths(paths: List[str]) -> List[str]:
-    out: List[str] = []
-    for candidate in paths:
-        path = _safe_path(candidate)
-        if path and path not in out:
-            out.append(path)
-        if len(out) == 2:
-            break
-    return out
-
-
-def _ucc_send_paths() -> List[str]:
-    canonical = _safe_path(UCC_SEND_PATH) or "/api/internal/wechat/send"
-    return _unique_paths([canonical, _LEGACY_SEND_PATH])
-
-
-def _canonical_origin(value: Any) -> Optional[str]:
-    if not isinstance(value, str) or not value or len(value) > 2048:
-        return None
-    if "\\" in value or any(ord(character) < 32 for character in value):
-        return None
-    try:
-        parsed = urllib.parse.urlsplit(value)
-        port = parsed.port
-    except (UnicodeError, ValueError):
-        return None
-    scheme = parsed.scheme.lower()
-    if scheme not in {"http", "https"} or not parsed.hostname:
-        return None
-    if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:
-        return None
-    try:
-        host = parsed.hostname.rstrip(".").encode("idna").decode("ascii").lower()
-    except UnicodeError:
-        return None
-    if not host:
-        return None
-    if port is None:
-        port = 80 if scheme == "http" else 443
-    rendered_host = f"[{host}]" if ":" in host else host
-    return f"{scheme}://{rendered_host}:{port}"
-
-
-def _adapter_base_url() -> Optional[str]:
-    origin = _canonical_origin(UCC_ADAPTER_URL)
-    allowed = {
-        canonical
-        for canonical in (_canonical_origin(value) for value in UCC_ADAPTER_ALLOWED_ORIGINS)
-        if canonical
-    }
-    return origin if origin and origin in allowed else None
-
-
-class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
-    def redirect_request(
-        self,
-        req: Any,
-        fp: Any,
-        code: int,
-        msg: str,
-        headers: Any,
-        newurl: str,
-    ) -> None:
-        del req, fp, code, msg, headers, newurl
-        return None
-
-
-def _open_no_redirect(request: urllib.request.Request, timeout: int) -> Any:
-    opener = urllib.request.build_opener(_NoRedirectHandler())
-    return opener.open(request, timeout=timeout)
-
-
-def ucc_adapter_alive() -> bool:
-    base_url = _adapter_base_url()
-    if not base_url:
-        return False
-    request = urllib.request.Request(
-        base_url + "/health", headers=_auth_headers(), method="GET"
-    )
-    try:
-        with _open_no_redirect(request, timeout=min(5, REQUEST_TIMEOUT_SECONDS)) as response:
-            return 200 <= int(response.status) < 300
-    except (OSError, ValueError, urllib.error.URLError):
-        return False
-
-
-def select_backend() -> Dict[str, Any]:
-    explicit = SEND_BACKEND.strip().lower()
-    if explicit:
-        if explicit == "ucc-adapter":
-            return {"backend": "ucc-adapter", "reason": "explicit"}
-        return {"backend": "unavailable", "reason": "unsupported_backend"}
-    if ucc_adapter_alive():
-        return {"backend": "ucc-adapter", "reason": "ucc_adapter_alive"}
-    return {"backend": "unavailable", "reason": "ucc_adapter_unavailable"}
-
-
-def _post_ucc(path: str, payload: Dict[str, Any]) -> Tuple[int, Dict[str, Any]]:
-    base_url = _adapter_base_url()
-    safe_path = _safe_path(path)
-    if not base_url or not safe_path:
-        raise _BackendProtocolError("invalid adapter endpoint")
-    try:
-        body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
-    except (TypeError, ValueError) as exc:
-        raise _BackendProtocolError("invalid request payload") from exc
-    request = urllib.request.Request(
-        base_url + safe_path,
-        data=body,
-        headers=_auth_headers(),
-        method="POST",
-    )
-    with _open_no_redirect(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
-        raw = response.read(MAX_RESPONSE_BYTES + 1)
-        if len(raw) > MAX_RESPONSE_BYTES:
-            raise _BackendProtocolError("adapter response too large")
-        try:
-            text = raw.decode("utf-8")
-        except UnicodeDecodeError as exc:
-            raise _BackendProtocolError("adapter response encoding invalid") from exc
-        data = _json_dict(text)
-        if not data:
-            raise _BackendProtocolError("adapter response invalid")
-        return int(response.status), data
-
-
-def _ucc_response_sent(data: Dict[str, Any]) -> bool:
-    return isinstance(data, dict) and data.get("ok") is True and data.get("status") == "sent"
-
-
-def _ucc_result_is_ok(data: Dict[str, Any], dry_run: bool = False) -> bool:
-    expected_status = "dry_run" if dry_run else "sent"
-    return isinstance(data, dict) and data.get("ok") is True and data.get("status") == expected_status
-
-
-def _utf8_size(value: str) -> Optional[int]:
-    try:
-        return len(value.encode("utf-8"))
-    except UnicodeEncodeError:
-        return None
-
-
-def _valid_identifier(value: Any, required: bool = False) -> bool:
-    if not isinstance(value, str):
-        return False
-    if required and not value.strip():
-        return False
-    if not value:
-        return not required
-    if len(value) > MAX_IDENTIFIER_BYTES:
-        return False
-    size = _utf8_size(value)
-    return size is not None and size <= MAX_IDENTIFIER_BYTES and not any(
-        ord(character) < 32 or ord(character) == 127 for character in value
-    )
-
-
-def _validate_metadata(metadata: Any) -> Optional[str]:
-    if not isinstance(metadata, dict):
-        return "invalid_input"
-    nodes = 0
-    stack: List[Tuple[Any, int]] = [(metadata, 0)]
-    while stack:
-        value, depth = stack.pop()
-        nodes += 1
-        if depth > MAX_METADATA_DEPTH or nodes > MAX_METADATA_NODES:
-            return "input_too_large"
-        if isinstance(value, dict):
-            if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
-                return "input_too_large"
-            for key, child in value.items():
-                if (
-                    not isinstance(key, str)
-                    or len(key) > MAX_METADATA_BYTES
-                    or _utf8_size(key) is None
-                ):
-                    return "invalid_input"
-                stack.append((child, depth + 1))
-        elif isinstance(value, list):
-            if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
-                return "input_too_large"
-            stack.extend((child, depth + 1) for child in value)
-        elif isinstance(value, str):
-            if len(value) > MAX_METADATA_BYTES:
-                return "input_too_large"
-            if _utf8_size(value) is None:
-                return "invalid_input"
-        elif value is not None and not isinstance(value, (bool, int, float)):
-            return "invalid_input"
-    try:
-        encoded = json.dumps(
-            metadata,
-            ensure_ascii=False,
-            allow_nan=False,
-            separators=(",", ":"),
-        ).encode("utf-8")
-    except (RecursionError, TypeError, UnicodeEncodeError, ValueError):
-        return "invalid_input"
-    return "input_too_large" if len(encoded) > MAX_METADATA_BYTES else None
-
-
-def _validate_payload(
-    target: Any,
-    content: Any,
-    account_id: Any,
-    message_id: Any,
-    metadata: Any,
-    dry_run: Any,
-) -> Optional[str]:
-    if not isinstance(target, str) or not isinstance(content, str):
-        return "invalid_input"
-    if not target.strip() or not content.strip() or not isinstance(dry_run, bool):
-        return "invalid_input"
-    if any(ord(character) < 32 or ord(character) == 127 for character in target):
-        return "invalid_input"
-    if "\x00" in content or not _valid_identifier(account_id) or not _valid_identifier(message_id):
-        return "invalid_input"
-    if len(target) > MAX_TARGET_BYTES or len(content) > MAX_CONTENT_BYTES:
-        return "input_too_large"
-    target_size = _utf8_size(target)
-    content_size = _utf8_size(content)
-    if target_size is None or content_size is None:
-        return "invalid_input"
-    if target_size > MAX_TARGET_BYTES or content_size > MAX_CONTENT_BYTES:
-        return "input_too_large"
-    return _validate_metadata(metadata)
-
-
-def _safe_provider_message_id(value: Any) -> Optional[str]:
-    if not isinstance(value, str) or not value:
-        return None
-    if len(value) > MAX_IDENTIFIER_BYTES:
-        return None
-    size = _utf8_size(value)
-    if size is None or size > MAX_IDENTIFIER_BYTES:
-        return None
-    if any(ord(character) < 32 or ord(character) == 127 for character in value):
-        return None
-    return value
-
-
-def send_ucc_adapter(
-    target: str,
-    content: str,
-    account_id: str = "",
-    message_id: str = "",
-    metadata: Optional[Dict[str, Any]] = None,
-    dry_run: bool = False,
-) -> BackendResult:
-    effective_account = account_id or WECHAT_ACCOUNT_ID
-    effective_metadata: Any = {} if metadata is None else metadata
-    error_code = _validate_payload(
-        target,
-        content,
-        effective_account,
-        message_id,
-        effective_metadata,
-        dry_run,
-    )
-    if error_code:
-        return BackendResult(False, "ucc-adapter", "failed", {"code": error_code})
-
-    payload = {
-        "target": target,
-        "content": content,
-        "account_id": effective_account,
-        "message_id": message_id,
-        "metadata": effective_metadata,
-        "dry_run": dry_run,
-    }
-    last_endpoint: Optional[str] = None
-    for path in _ucc_send_paths():
-        last_endpoint = path
-        try:
-            http_status, data = _post_ucc(path, payload)
-        except urllib.error.HTTPError as exc:
-            if exc.code in {404, 405}:
-                continue
-            return BackendResult(
-                False,
-                "ucc-adapter",
-                "failed",
-                {"code": "ucc_adapter_http_error", "http_status": int(exc.code), "endpoint": path},
-            )
-        except (OSError, ValueError, urllib.error.URLError, _BackendProtocolError):
-            return BackendResult(
-                False,
-                "ucc-adapter",
-                "failed",
-                {"code": "ucc_adapter_unavailable", "endpoint": path},
-            )
-        except Exception:
-            # Fail closed without returning exception text, type, or chained cause.
-            return BackendResult(
-                False,
-                "ucc-adapter",
-                "failed",
-                {"code": "ucc_adapter_unavailable", "endpoint": path},
-            )
-
-        if not (200 <= http_status < 300):
-            return BackendResult(
-                False,
-                "ucc-adapter",
-                "failed",
-                {"code": "ucc_adapter_http_error", "http_status": http_status, "endpoint": path},
-            )
-        if not _ucc_result_is_ok(data, dry_run):
-            return BackendResult(
-                False,
-                "ucc-adapter",
-                "failed",
-                {"code": "ucc_adapter_rejected", "http_status": http_status, "endpoint": path},
-            )
-        confirmed_status = "dry_run" if dry_run else "sent"
-        result_data: Dict[str, Any] = {
-            "code": confirmed_status,
-            "http_status": http_status,
-            "endpoint": path,
-        }
-        provider_message_id = _safe_provider_message_id(data.get("provider_message_id"))
-        if provider_message_id:
-            result_data["provider_message_id"] = provider_message_id
-        return BackendResult(True, "ucc-adapter", confirmed_status, result_data)
-
-    data: Dict[str, Any] = {"code": "ucc_adapter_unavailable"}
-    if last_endpoint:
-        data["endpoint"] = last_endpoint
-    return BackendResult(False, "ucc-adapter", "failed", data)
-
-
-def send_openclaw_channel(
-    target: str,
-    content: str,
-    account_id: str = "",
-    message_id: str = "",
-    metadata: Optional[Dict[str, Any]] = None,
-    dry_run: bool = False,
-) -> BackendResult:
-    del target, content, account_id, message_id, metadata, dry_run
-    return BackendResult(
-        False,
-        "openclaw-channel",
-        "failed",
-        {"code": "unsupported_backend"},
-    )
-
-
-def send_message(
-    target: str,
-    content: str,
-    account_id: str = "",
-    message_id: str = "",
-    metadata: Optional[Dict[str, Any]] = None,
-    dry_run: bool = False,
-) -> Dict[str, Any]:
-    selected = select_backend()
-    if selected.get("backend") == "ucc-adapter":
-        return send_ucc_adapter(
-            target, content, account_id, message_id, metadata, dry_run
-        ).to_dict()
-    code = "unsupported_backend" if selected.get("reason") == "unsupported_backend" else "wechat_outbound_backend_unavailable"
-    return BackendResult(False, "unavailable", "failed", {"code": code}).to_dict()
-
-
-if __name__ == "__main__":
-    print(json.dumps(select_backend(), ensure_ascii=False, separators=(",", ":")))

tests/test_wechat_send_backend.py

@@ -1,20 +1,11 @@
 from __future__ import annotations
 
 import importlib.util
-import io
-import os
-import signal
-import subprocess
 import sys
-import tempfile
-import time
-import urllib.error
-import urllib.parse
-from contextlib import redirect_stderr, redirect_stdout
 from pathlib import Path
 from types import SimpleNamespace
 from unittest import TestCase
-from unittest.mock import Mock, call, patch
+from unittest.mock import patch
 
 
 BACKEND_PATH = Path(__file__).resolve().parents[1] / "scripts" / "wechat_send_backend.py"
@@ -89,450 +80,6 @@ class WechatSendBackendTests(TestCase):
         self.assertEqual(args[1]["content"], "hello")
         self.assertFalse(args[1]["dry_run"])
 
-    def test_ucc_auth_is_authorization_header_not_argv_or_output(self):
-        secret = "ucc-super-secret-token"
-
-        class Response:
-            status = 200
-
-            def __enter__(self):
-                return self
-
-            def __exit__(self, *_args):
-                return False
-
-            def read(self, _limit=-1):
-                return b'{"ok":true,"status":"sent"}'
-
-        urlopen = Mock(return_value=Response())
-        output = io.StringIO()
-        with (
-            patch.object(self.backend, "UCC_ADAPTER_TOKEN", secret),
-            patch.object(self.backend, "_open_no_redirect", urlopen),
-            redirect_stdout(output),
-            redirect_stderr(output),
-        ):
-            status, payload = self.backend._post_ucc(
-                "/api/internal/wechat/send", {"target": "chat-001", "content": "hello"}
-            )
-
-        request = urlopen.call_args.args[0]
-        self.assertEqual(request.get_header("Authorization"), f"Bearer {secret}")
-        self.assertEqual(status, 200)
-        self.assertEqual(payload, {"ok": True, "status": "sent"})
-        self.assertNotIn(secret, " ".join(self.backend._openclaw_base_args()))
-        self.assertNotIn(secret, output.getvalue())
-        self.assertFalse({"--gateway", "--token", "--password"} & set(self.backend._openclaw_base_args()))
-
-    def test_ucc_paths_are_canonical_first_unique_and_bounded(self):
-        with patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"):
-            paths = self.backend._ucc_send_paths()
-
-        self.assertEqual(paths[0], "/api/internal/wechat/send")
-        self.assertLessEqual(len(paths), 2)
-        self.assertEqual(len(paths), len(set(paths)))
-        self.assertTrue(all(path.startswith("/") for path in paths))
-
-    def test_ucc_accepts_only_explicit_true_and_sent_when_not_dry_run(self):
-        self.assertTrue(self.backend._ucc_response_sent({"ok": True, "status": "sent"}))
-        for response in (
-            {"ok": 1, "status": "sent"},
-            {"ok": True, "status": "accepted"},
-            {"ok": False, "status": "sent"},
-            {"status": "sent"},
-        ):
-            with self.subTest(response=response):
-                self.assertFalse(self.backend._ucc_response_sent(response))
-
-    def test_ambiguous_canonical_failures_never_fallback(self):
-        secret = "raw-upstream-secret"
-        failures = (
-            TimeoutError(secret),
-            ConnectionError(secret),
-            self.backend._BackendProtocolError(secret),
-            RuntimeError(secret),
-        )
-        for failure in failures:
-            with self.subTest(failure=type(failure).__name__):
-                with (
-                    patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
-                    patch.object(self.backend, "_post_ucc", side_effect=failure) as post,
-                ):
-                    result = self.backend.send_ucc_adapter(
-                        target="chat-001", content="hello", message_id="msg-001"
-                    ).to_dict()
-
-                self.assertFalse(result["ok"])
-                self.assertEqual(result["code"], "ucc_adapter_unavailable")
-                self.assertEqual(post.call_count, 1)
-                self.assertNotIn(secret, repr(result))
-
-    def test_only_explicit_http_404_or_405_uses_legacy_fallback(self):
-        for http_status in (404, 405):
-            with self.subTest(http_status=http_status):
-                missing = urllib.error.HTTPError(
-                    "http://127.0.0.1/api/internal/wechat/send",
-                    http_status,
-                    "not found",
-                    {},
-                    None,
-                )
-                with (
-                    patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
-                    patch.object(
-                        self.backend,
-                        "_post_ucc",
-                        side_effect=[missing, (200, {"ok": True, "status": "sent"})],
-                    ) as post,
-                ):
-                    result = self.backend.send_ucc_adapter(
-                        target="chat-001", content="hello", message_id="msg-001"
-                    ).to_dict()
-
-                self.assertTrue(result["ok"])
-                self.assertEqual(result["endpoint"], "/internal/channels/wechat/send")
-                self.assertEqual(post.call_count, 2)
-
-    def test_non_not_found_http_error_never_falls_back(self):
-        denied = urllib.error.HTTPError(
-            "http://127.0.0.1/api/internal/wechat/send", 401, "denied", {}, None
-        )
-        with patch.object(self.backend, "_post_ucc", side_effect=denied) as post:
-            result = self.backend.send_ucc_adapter(target="chat-001", content="hello").to_dict()
-
-        self.assertFalse(result["ok"])
-        self.assertEqual(result["code"], "ucc_adapter_http_error")
-        post.assert_called_once()
-
-    def test_ucc_rejection_does_not_expose_raw_error_or_secret(self):
-        secret = "provider-password=secret-value"
-        with patch.object(
-            self.backend,
-            "_post_ucc",
-            return_value=(200, {"ok": False, "status": "failed", "error": secret}),
-        ):
-            result = self.backend.send_ucc_adapter(
-                target="chat-001", content="hello", message_id="msg-001"
-            ).to_dict()
-
-        self.assertFalse(result["ok"])
-        self.assertEqual(result["code"], "ucc_adapter_rejected")
-        self.assertNotIn(secret, repr(result))
-
-    def test_dry_run_requires_dry_run_status_and_never_reports_sent(self):
-        with patch.object(
-            self.backend,
-            "_post_ucc",
-            return_value=(200, {"ok": True, "status": "dry_run"}),
-        ):
-            result = self.backend.send_ucc_adapter(
-                target="chat-001", content="hello", dry_run=True
-            ).to_dict()
-
-        self.assertTrue(result["ok"])
-        self.assertEqual(result["status"], "dry_run")
-        self.assertNotEqual(result["status"], "sent")
-
-        for dry_run, upstream_status in ((True, "sent"), (False, "dry_run")):
-            with self.subTest(dry_run=dry_run, upstream_status=upstream_status):
-                with patch.object(
-                    self.backend,
-                    "_post_ucc",
-                    return_value=(200, {"ok": True, "status": upstream_status}),
-                ):
-                    rejected = self.backend.send_ucc_adapter(
-                        target="chat-001", content="hello", dry_run=dry_run
-                    ).to_dict()
-                self.assertFalse(rejected["ok"])
-                self.assertEqual(rejected["status"], "failed")
-
-    def test_invalid_provider_message_id_never_reverses_confirmed_send(self):
-        invalid_values = (
-            "\ud800",
-            "contains\x00control",
-            "x" * (self.backend.MAX_IDENTIFIER_BYTES + 1),
-            {"not": "a string"},
-        )
-        for provider_message_id in invalid_values:
-            with self.subTest(value=repr(provider_message_id)[:40]):
-                with patch.object(
-                    self.backend,
-                    "_post_ucc",
-                    return_value=(
-                        200,
-                        {
-                            "ok": True,
-                            "status": "sent",
-                            "provider_message_id": provider_message_id,
-                        },
-                    ),
-                ):
-                    result = self.backend.send_ucc_adapter(
-                        target="chat-001", content="hello"
-                    ).to_dict()
-
-                self.assertTrue(result["ok"])
-                self.assertEqual(result["status"], "sent")
-                self.assertNotIn("provider_message_id", result)
-
-    def test_openclaw_2026_6_11_commands_have_no_unsupported_global_gateway(self):
-        completed = SimpleNamespace(returncode=0, stdout='{"channels": []}')
-        with (
-            patch.object(self.backend, "OPENCLAW_BIN", "/usr/local/bin/openclaw"),
-            patch.object(self.backend, "_run_openclaw", return_value=completed) as run,
-        ):
-            self.assertIsNone(self.backend.discover_openclaw_wechat_channel())
-
-        command = run.call_args.args[0]
-        self.assertEqual(command, ["/usr/local/bin/openclaw", "channels", "list", "--json"])
-        self.assertFalse({"--gateway", "--token", "--password"} & set(command))
-
-    def test_openclaw_output_overflow_terminates_while_process_is_running(self):
-        writer = (
-            "import os,time;"
-            f"os.write(1,b'x'*{self.backend.MAX_RESPONSE_BYTES + 4096});"
-            "time.sleep(10)"
-        )
-        started = time.monotonic()
-        completed = self.backend._run_openclaw([sys.executable, "-c", writer], timeout=4)
-        elapsed = time.monotonic() - started
-
-        self.assertLess(elapsed, 2.0)
-        self.assertNotEqual(completed.returncode, 0)
-        self.assertEqual(completed.stdout, "")
-        self.assertEqual(completed.stderr, "")
-
-    def test_openclaw_timeout_returns_failure_and_terminates_promptly(self):
-        started = time.monotonic()
-        completed = self.backend._run_openclaw(
-            [sys.executable, "-c", "import time;time.sleep(10)"], timeout=1
-        )
-        elapsed = time.monotonic() - started
-
-        self.assertLess(elapsed, 3.0)
-        self.assertNotEqual(completed.returncode, 0)
-        self.assertEqual(completed.stdout, "")
-        self.assertEqual(completed.stderr, "")
-
-    def test_timeout_kills_inherited_pipe_descendant_after_parent_exits(self):
-        if os.name != "posix" or not Path("/proc").is_dir():
-            self.skipTest("POSIX process-group and procfs contract")
-
-        def process_is_running(pid):
-            try:
-                state = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").split()[2]
-            except (FileNotFoundError, IndexError, OSError):
-                return False
-            return state not in {"X", "Z"}
-
-        with tempfile.TemporaryDirectory() as directory:
-            pid_path = Path(directory) / "descendant.pid"
-            descendant = "import time;time.sleep(30)"
-            launcher = (
-                "import pathlib,subprocess,sys;"
-                f"child=subprocess.Popen([sys.executable,'-c',{descendant!r}]);"
-                f"pathlib.Path({str(pid_path)!r}).write_text(str(child.pid),encoding='utf-8')"
-            )
-            started = time.monotonic()
-            completed = self.backend._run_openclaw(
-                [sys.executable, "-c", launcher], timeout=1
-            )
-            elapsed = time.monotonic() - started
-            self.assertTrue(pid_path.is_file())
-            descendant_pid = int(pid_path.read_text(encoding="utf-8"))
-            try:
-                deadline = time.monotonic() + 2
-                while process_is_running(descendant_pid) and time.monotonic() < deadline:
-                    time.sleep(0.02)
-                self.assertFalse(process_is_running(descendant_pid))
-            finally:
-                if process_is_running(descendant_pid):
-                    os.kill(descendant_pid, signal.SIGKILL)
-
-        self.assertLess(elapsed, 3.0)
-        self.assertNotEqual(completed.returncode, 0)
-
-    def test_process_tree_termination_uses_posix_process_group(self):
-        if os.name != "posix":
-            self.skipTest("POSIX process-group contract")
-
-        class Process:
-            pid = 43210
-
-            def __init__(self):
-                self.wait_count = 0
-
-            def poll(self):
-                return None
-
-            def wait(self, timeout):
-                self.wait_count += 1
-                if self.wait_count == 1:
-                    raise subprocess.TimeoutExpired(["openclaw"], timeout)
-                return -signal.SIGKILL
-
-        process = Process()
-        with patch.object(self.backend.os, "killpg") as killpg:
-            self.backend._terminate_process_tree(process)
-
-        self.assertEqual(
-            killpg.call_args_list,
-            [
-                call(process.pid, signal.SIGTERM),
-                call(process.pid, signal.SIGKILL),
-            ],
-        )
-
-    def test_unsupported_and_generic_chat_backends_fail_closed(self):
-        with (
-            patch.object(self.backend, "SEND_BACKEND", "openclaw-chat"),
-            patch.object(self.backend, "ucc_adapter_alive", return_value=False),
-        ):
-            selected = self.backend.select_backend()
-            result = self.backend.send_message(target="chat-001", content="hello")
-
-        self.assertEqual(selected["backend"], "unavailable")
-        self.assertEqual(selected["reason"], "unsupported_backend")
-        self.assertFalse(result["ok"])
-        self.assertEqual(result["code"], "unsupported_backend")
-
-    def test_invalid_or_oversized_input_is_rejected_before_network(self):
-        with patch.object(self.backend, "_post_ucc") as post:
-            empty = self.backend.send_ucc_adapter(target="", content="hello").to_dict()
-            oversized = self.backend.send_ucc_adapter(
-                target="chat-001", content="x" * (self.backend.MAX_CONTENT_BYTES + 1)
-            ).to_dict()
-
-        self.assertEqual(empty["code"], "invalid_input")
-        self.assertEqual(oversized["code"], "input_too_large")
-        post.assert_not_called()
-
-    def test_unpaired_surrogate_inputs_fail_closed_before_network(self):
-        cases = (
-            {"target": "\ud800", "content": "hello"},
-            {"target": "chat-001", "content": "\ud800"},
-            {"target": "chat-001", "content": "hello", "account_id": "\ud800"},
-            {"target": "chat-001", "content": "hello", "message_id": "\ud800"},
-            {"target": "chat-001", "content": "hello", "metadata": {"bad": "\ud800"}},
-        )
-        with patch.object(self.backend, "_post_ucc") as post:
-            for kwargs in cases:
-                with self.subTest(field=next(reversed(kwargs))):
-                    result = self.backend.send_ucc_adapter(**kwargs).to_dict()
-                    self.assertEqual(result["code"], "invalid_input")
-        post.assert_not_called()
-
-    def test_deeply_nested_metadata_is_bounded_before_network(self):
-        metadata = {}
-        for _ in range(1200):
-            metadata = {"nested": metadata}
-
-        with patch.object(self.backend, "_post_ucc") as post:
-            result = self.backend.send_ucc_adapter(
-                target="chat-001", content="hello", metadata=metadata
-            ).to_dict()
-
-        self.assertEqual(result["code"], "input_too_large")
-        post.assert_not_called()
-
-    def test_adapter_origin_must_be_allowlisted(self):
-        with (
-            patch.object(self.backend, "UCC_ADAPTER_URL", "https://evil.example:8443"),
-            patch.object(
-                self.backend,
-                "UCC_ADAPTER_ALLOWED_ORIGINS",
-                ("http://127.0.0.1:18800",),
-            ),
-        ):
-            self.assertIsNone(self.backend._adapter_base_url())
-
-        with (
-            patch.object(self.backend, "UCC_ADAPTER_URL", "https://adapter.example:8443"),
-            patch.object(
-                self.backend,
-                "UCC_ADAPTER_ALLOWED_ORIGINS",
-                ("https://adapter.example:8443",),
-            ),
-        ):
-            self.assertEqual(
-                self.backend._adapter_base_url(), "https://adapter.example:8443"
-            )
-
-    def test_adapter_origin_rejects_scheme_userinfo_fragment_and_path(self):
-        invalid_origins = (
-            "ftp://127.0.0.1:18800",
-            "http://user:password@127.0.0.1:18800",
-            "http://127.0.0.1:18800#fragment",
-            "http://127.0.0.1:18800/base-path",
-        )
-        for origin in invalid_origins:
-            with self.subTest(origin=origin):
-                with (
-                    patch.object(self.backend, "UCC_ADAPTER_URL", origin),
-                    patch.object(
-                        self.backend,
-                        "UCC_ADAPTER_ALLOWED_ORIGINS",
-                        ("http://127.0.0.1:18800",),
-                    ),
-                ):
-                    self.assertIsNone(self.backend._adapter_base_url())
-
-    def test_adapter_path_rejects_decoded_traversal(self):
-        for path in (
-            "/api/%2e%2e/send",
-            "/api/%2E%2E/send",
-            "/api/%2525252e%2525252e/send",
-            "/api/%252525252e%252525252e/send",
-            "/api/../send",
-            "/api/%zz/send",
-        ):
-            with self.subTest(path=path):
-                self.assertIsNone(self.backend._safe_path(path))
-
-    def test_adapter_path_keeps_intended_safe_canonical_paths(self):
-        for path in (
-            "/api/internal/wechat/send",
-            "/internal/channels/wechat/send",
-            "/api/%7Ehealth/status",
-        ):
-            with self.subTest(path=path):
-                self.assertEqual(self.backend._safe_path(path), path)
-
-    def test_redirects_are_refused_and_authorization_cannot_cross_origin(self):
-        handler = self.backend._NoRedirectHandler()
-        self.assertIsNone(
-            handler.redirect_request(
-                None,
-                None,
-                302,
-                "redirect",
-                {"Location": "https://evil.example/steal"},
-                "https://evil.example/steal",
-            )
-        )
-
-        redirected = urllib.error.HTTPError(
-            "http://127.0.0.1:18800/api/internal/wechat/send",
-            302,
-            "redirect",
-            {"Location": "https://evil.example/steal"},
-            None,
-        )
-        with (
-            patch.object(self.backend, "UCC_ADAPTER_TOKEN", "header-secret"),
-            patch.object(self.backend, "_open_no_redirect", side_effect=redirected) as opened,
-        ):
-            result = self.backend.send_ucc_adapter(
-                target="chat-001", content="hello"
-            ).to_dict()
-
-        request = opened.call_args.args[0]
-        self.assertEqual(urllib.parse.urlsplit(request.full_url).netloc, "127.0.0.1:18800")
-        self.assertEqual(request.get_header("Authorization"), "Bearer header-secret")
-        opened.assert_called_once()
-        self.assertFalse(result["ok"])
-        self.assertEqual(result["code"], "ucc_adapter_http_error")
-
 
 if __name__ == "__main__":
     import unittest