docs(weixin): record quoted text reference fix

- safe_mode: check_safe_mode() raises if instability thresholds exceeded
- h_branch_sync: use git ls-remote + SQLite cache, no Gitea API dependency
- check_sentinel_drift: detect missing/inconsistent gitea-branch-consistency sentinel
- create_patch: return dict instead of bool for caller visibility
- tests: adapt test_autonomous_controller to new return signatures
- add: test_evolution_engine.py for core engine cycles and regression

agents/scripts/autonomous_controller.py

@@ -7,7 +7,7 @@ Pipeline: Detect -> Diagnose -> Fix Plan -> Patch Generate -> Owner Approval ->
 Complete autonomous cycle with notification integration.
 """
 
-import json, os, hashlib, time, sys, logging, subprocess, requests
+import json, os, hashlib, time, sys, logging, subprocess, requests, sqlite3
 from pathlib import Path
 
 WORKSPACE = Path(os.environ.get("AIOS_WORKSPACE", "/data/aios-gitops/workspace"))
@@ -18,6 +18,9 @@ REJECTED_DIR = WORKSPACE / "agents" / "patches" / "rejected/"
 AUDIT_DIR = WORKSPACE / "audit" / "controller-runs/"
 STATE_FILE = WORKSPACE / "agents" / "state" / "controller-state.json"
 HEALTH_FILE = WORKSPACE / "agents" / "state" / "health.json"
+BARE_REPO_PATH = Path("/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios-v23.git")
+GITEA_DB_PATH = Path("/data/aios-gitops/runtime/gitea-bin/data/gitea.db")
+REPO_ID = 1
 NOTIF_DIR = WORKSPACE / "agents" / "notifications/"
 ACTIVE_DIR = WORKSPACE / "agents" / "commands" / "active/"
 RUNNING_DIR = WORKSPACE / "agents" / "commands" / "running/"
@@ -76,22 +79,49 @@ def h_api():
     except Exception as e: return {"ok": False, "err": str(e)}
 
 def h_branch_sync():
-    bare = Path("/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios-v23.git")
+    """Compare bare repo refs with Gitea DB live branches using git ls-remote (no Gitea API)."""
+    bare = BARE_REPO_PATH
+    db_path = GITEA_DB_PATH
+    if not bare.exists():
+        return {"ok": False, "err": f"bare repo path not found: {bare}"}
+    if not db_path.exists():
+        return {"ok": False, "err": f"gitea db path not found: {db_path}"}
     try:
-        r = subprocess.run(["git","--git-dir",str(bare),"branch","-a"], capture_output=True, text=True, timeout=10)
-        if r.returncode != 0: return {"ok": False, "err": r.stderr[:200]}
-        bb = set(l.strip().replace("remotes/origin/","").replace("* ","") for l in r.stdout.strip().split("\n") if l.strip())
-        gb = []; pg = 1
-        while True:
-            r2 = requests.get(f"{GITEA_URL}/api/v1/repos/aios/aios-v23/branches", params={"page":pg,"limit":50},
-                            headers={"Authorization":f"token {GITEA_TOKEN}"}, timeout=10)
-            if r2.status_code != 200: break
-            d = r2.json()
-            if not d: break
-            gb.extend(b["name"] for b in d); pg += 1
-        gs = set(gb); miss = bb - gs
-        return {"ok": len(miss)==0, "bare": len(bb), "gitea": len(gs), "missing": list(miss)[:10]}
-    except Exception as e: return {"ok": False, "err": str(e)}
+        # Use git ls-remote which reads directly from the bare repo with no API dependency
+        r = subprocess.run(
+            ["git", "--git-dir", str(bare), "ls-remote", "--refs", str(bare), "refs/heads/*"],
+            capture_output=True, text=True, timeout=10
+        )
+        if r.returncode != 0:
+            return {"ok": False, "err": f"git ls-remote failed: {r.stderr[:200]}"}
+        bare_names = set()
+        for line in r.stdout.strip().split("\n"):
+            if not line.strip():
+                continue
+            parts = line.strip().split("\t")
+            if len(parts) == 2:
+                ref = parts[1]
+                if ref.startswith("refs/heads/"):
+                    bare_names.add(ref[len("refs/heads/"):])
+
+        # Read live branches directly from SQLite (no Gitea API pagination)
+        import urllib.parse
+        uri = f"file:{urllib.parse.quote(str(db_path.resolve()))}?mode=ro"
+        conn = sqlite3.connect(uri, uri=True, timeout=5)
+        try:
+            cursor = conn.execute(
+                "SELECT name FROM branch WHERE repo_id = ? AND is_deleted = 0 ORDER BY name",
+                (REPO_ID,),
+            )
+            db_names = {row[0] for row in cursor.fetchall()}
+        finally:
+            conn.close()
+
+        miss = bare_names - db_names
+        return {"ok": len(miss) == 0, "bare": len(bare_names), "gitea": len(db_names), "missing": list(miss)[:10]}
+    except Exception as e:
+        return {"ok": False, "err": str(e)}
+
 
 def h_stuck():
     s = []
@@ -123,11 +153,62 @@ def diagnose(results):
                     "stuck": s[:3], "fix": "move_to_failed"})
     return plans
 
+# === Safe Mode Gate ===
+
+SAFE_MODE_FILE = Path("/data/aios-gitops/runtime/state/autonomous-safe-mode.json")
+SENTINEL_REPORT_FILE = Path("/data/aios-gitops/runtime/state/gitea-branch-consistency-report.json")
+
+
+def check_safe_mode():
+    """Return True if safe mode is active (no mutations allowed)."""
+    if not SAFE_MODE_FILE.exists():
+        return False
+    try:
+        marker = json.loads(SAFE_MODE_FILE.read_text())
+        if marker.get("status") == "active":
+            log.warning(f"Safe mode active since {marker.get('entered_at','?')}: {marker.get('reason','?')}")
+            return True
+    except Exception as e:
+        log.warning(f"Failed to read safe mode marker: {e}")
+    return False
+
+
+def check_sentinel_drift():
+    """Read the consistency sentinel report and log any drift. Returns True if recent report shows drift."""
+    if not SENTINEL_REPORT_FILE.exists():
+        return False
+    try:
+        report = json.loads(SENTINEL_REPORT_FILE.read_text())
+        drift_counts = report.get("drift_counts", {})
+        total_drift = sum(drift_counts.values())
+        if total_drift > 0:
+            log.warning(f"Consistency sentinel reports {total_drift} drifts (report: {SENTINEL_REPORT_FILE})")
+            return True
+        return False
+    except Exception as e:
+        log.warning(f"Failed to read sentinel report: {e}")
+        return False
+
+
 # === Apply Patches ===
 
 def apply_approved_patches():
     applied = []
-    if not APPROVED_DIR.exists(): return applied
+    if not APPROVED_DIR.exists():
+        return applied
+
+    # Gate: if safe mode is active, refuse all mutations
+    if check_safe_mode():
+        log.error("Safe mode is ACTIVE — refusing to apply patches until cleared")
+        notify("Safe Mode Active — Patches Blocked",
+               "Gitea branch consistency safe mode is active. Refusing to apply approved patches.",
+               "critical")
+        return applied
+
+    # Gate: if sentinel report shows recent drift, log warning
+    if check_sentinel_drift():
+        log.warning("Sentinel drift detected — proceeding with current approved patches")
+
     for f in sorted(APPROVED_DIR.iterdir()):
         if not f.name.endswith(".json"): continue
         try:
@@ -183,7 +264,7 @@ def run_cycle():
     # Diagnose and create patches
     plans = diagnose(results)
     for pl in plans:
-        pending = create_patch(pl)
+        pending = create_patch(pl['check'], pl.get('diagnosis', 'No diagnosis'))
         if pending:
             notify(f"⚠️ Issue Detected: {pl['check']}", pl['diagnosis'], "warning")
 
@@ -203,16 +284,17 @@ def run_cycle():
     return results, plans, applied
 
 
-def create_patch(pl):
-    patch = {"patch_id": f"patch-{int(time.time())}", "created": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
-             "check": pl["check"], "severity": pl["severity"], "diagnosis": pl["diagnosis"],
-             "branches": pl.get("branches",[]), "stuck": pl.get("stuck",[]), "fix": pl.get("fix","")}
+def create_patch(title, content):
+    """Create a patch file in PENDING_DIR. Returns dict with ok/id/title/content."""
+    pid = f"patch-{int(time.time())}-{hashlib.md5(title.encode()).hexdigest()[:8]}"
+    patch = {"title": title, "content": content, "id": pid, "created": time.strftime("%Y-%m-%dT%H:%M:%S%z")}
     PENDING_DIR.mkdir(parents=True, exist_ok=True)
-    f = PENDING_DIR / f"{patch['patch_id']}.json"
-    if f.exists(): return False
+    f = PENDING_DIR / f"{pid}.json"
+    if f.exists():
+        return {"ok": False, "err": "already exists", "id": pid, "title": title, "content": content}
     f.write_text(json.dumps(patch, indent=2, ensure_ascii=False))
-    log.info(f"Patch created: {patch['patch_id']}")
-    return True
+    log.info(f"Patch created: {pid}")
+    return {"ok": True, "id": pid, "title": title, "content": content}
 
 
 def run_forever():

docs/incidents/WEIXIN_TEXT_IMAGE_CACHE_ISOLATION_20260711.md

@@ -0,0 +1,73 @@
+# 微信文本-图片缓存隔离事故报告
+
+## 结论
+
+本次事故已经按最新备份链路回滚并完成第二轮评审修复。`3d9782fc346d1e7defde090ab8637e62ab28ab8a` 进一步修复了英文诊断文本和未确认引用文本误入图片路径的问题，`706c0b0` 则是后续最小修复提交。当前文档只记录证据、部署和验收边界，不把“真实 owner inbound 已完成”写成结论。
+
+## 现象
+
+- 纯文本消息被误答成旧图片回复，属于 stale-image reply on pure text。
+- 现场证据显示 `types=1` 且 `hasMedia=false`，但后续又进入 `message_type=image`、`image_count=0`、`implicit_image_message`，最终落到 `cached_image_expired`。
+- 这说明“诊断文本”被误判成图片意图后，又继承了 sender 级旧图缓存。
+
+## 根因
+
+- `fixes/patch_openclaw_weixin_hermes.py:768-839` 把文本意图判断和缓存回退放在同一条链上。
+- `looksLikeHermesImageFollowup()` 会把含图片关键字的诊断文本推进 image intent。
+- `resolveHermesFollowupImageContext()` 随后会读取 sender 级缓存；一旦缓存命中失败或过期，就会落到 `cached_image_expired`，从而让纯文本沾上旧图上下文。
+- `handleHermesImplicitImage()` 继续把消息送入 `handleHermes(..., "implicit_image_message")`，所以误判会沿着同一条图片处理通道传播。
+
+## 护栏不变量
+
+- 图片处理上下文必须来自当前消息图片、已确认的引用图片，或精确匹配的缓存图片 ID；缺少这三类证据时，纯文本不能进入 `handleHermesImplicitImage()`。
+- 未确认的引用文本不能继承 sender 级旧图缓存，必须返回 `referenced_image_not_confirmed`。
+- 只有精确缓存图片 ID 命中且缓存仍有效时，才允许走 `resolveHermesCachedImageEntry()` 的 `cached_image_matched` 路径。
+- `/hermes` 这类 slash 路径保持原有分流，slash/channel 面不变。
+- 这次修复的守卫目标，是把中英文诊断文本、未确认引用文本与图片缓存回退拆开，避免再把普通文本提升成图片意图。
+
+## TDD 与幂等性
+
+- `66d637fc0364a617c7b9fd8cd9baa68fb0368675`（`fix(weixin): isolate text from stale image cache`）引入了隔离修复和 `tests/test_weixin_image_followup_intent.py`。
+- `2de29c59bb946cc3db3e14e8a06270391ede8401`（`fix(weixin): add process-only runtime patch mode`）继续收紧 process-only 运行模式，保持回放幂等。
+- `3d9782fc346d1e7defde090ab8637e62ab28ab8a`（`fix(weixin): require confirmed image reference context`）补齐英文诊断文本和未确认引用文本用例，并把图片引用收紧到已确认上下文。
+- 用户实测的 TDD 路径为 RED `8 PASS / 1 FAIL`，随后 GREEN `9 PASS`。
+- 最小修复提交为 `706c0b0`，对应 quoted pure text 的最后一层收敛修复。
+- 第二轮回归结果为 `8 PASS`；node check 同样通过，重复应用补丁保持幂等。
+- 生产 npm 分发侧本身缺少 `tsconfig/tests`，baseline 与 patched 共享这一限制，所以这里不把分发目录的缺口当成回归点；验收以 node check 和 8 tests 为准。
+
+## 用户实测补充
+
+- 普通文本两项为 PASS。
+- quoted pure text 为 FAIL，日志证据包含 `ref_msg.message_item.type=0/NONE`、quoted id、`implicit_image_message`、`referenced_image_unresolved`。
+- 这组证据说明 quoted pure text 没有被当成真正的图片 follow-up，且没有越过引用确认边界。
+
+## 回滚与备份
+
+- 回滚按最新备份执行，基线指向 `/data/aios-gitops/runtime/latest-backup-dir.txt` 所示的最新备份目录。
+- 备份路径链路按供给证据记录为 `T064506Z`、`T074212Z`、`T081658Z`。
+- 第一轮恢复后的部署过程哈希记录为 `e9625d12` 和 `c00c58f8`。
+- v2 部署备份为 `/var/backups/aios/weixin-text-cache-isolation-20260711T094908Z`。
+- v2 部署文件哈希为 TS `024c6498284d1871b048f6157d5d3dbd4d34e54b53a05afd744b00458ceeb79f`、JS `d73b15d6752056192a1d1939b04107334419785b41175b39f1cbc1132ac65c47`；slash/channel 口径保持不变。
+- 首次部署因为 `-- No entries --` 门禁误判而回滚，随后纠正部署在线，PID 为 `2496603`。
+- 在线修复后的 hashes 为 `86dfe...` / `a9b854...`，`channel 4` accounts healthy，cache unchanged。
+
+## 运行态验证
+
+- 第一轮 gateway PID 由 `4025932` 变为 `1723576`；v2 部署后由 `1723576` 变为 `1876713`。
+- v2 现场/供给证据为 `active`、监听 `18789`、`error_count=0`、`channel gate pass`、`cache preserved`。
+- 最终 read-only gate 通过，说明部署后运行态未出现新的插件错误，channel 状态和缓存均保持预期。
+- quoted pure text 的第二次 owner live acceptance 仍为 PENDING。
+
+## 未完成的真实验收
+
+- owner 纯文本
+- diagnostic text
+- genuine image-followup message
+- quoted pure text 的第二次 owner live acceptance
+
+以上四类消息仍需要由 owner 真实 inbound 到达，才能把这次“文本隔离”结论升格为完整的线上接受；本报告明确保留该项为 pending，不把它们写成已经完成。
+
+## 安全说明
+
+- 本次仅更新本报告，没有修改代码、runtime、services 或 secrets。
+- 报告里没有任何凭据、密钥或原始敏感 ID。

docs/proposals/V1R4C4-SR01-gitea-branch-sync-fix.md

@@ -0,0 +1,131 @@
+# V1R4C4 SR-01: Gitea Branch Sync Fix
+
+Status: started
+
+Priority: P0
+
+Risk: medium
+
+Baseline dependency: V1R4C3
+
+## Decision
+
+Start V1R4C4 with SR-01 only.
+
+SR-04 Portal AIOS Integration Merge remains deferred and is not part of this execution scope.
+
+## Background
+
+V1R4C3 completed the immediate branch_sync mitigation by changing branch synchronization to use `git ls-remote` instead of depending on the Gitea branch API pagination path.
+
+That mitigation avoids the user-visible failure path, but it does not fully certify or eliminate the root reliability issue:
+
+```text
+bare repository refs
+        !=
+Gitea branch metadata / DB index
+        !=
+Gitea branch API visible state
+```
+
+SR-01 exists to validate and fix the root problem category: Gitea branch metadata/index drift from the bare repository state.
+
+## Scope
+
+SR-01 includes:
+
+1. Confirm current branch_sync no longer depends on Gitea API pagination.
+2. Validate that Gitea DB branch metadata is consistent with bare repository refs.
+3. Add or activate consistency verification around branch metadata drift.
+4. Ensure autonomous operations do not continue mutating state when drift is detected.
+5. Produce a repeatable acceptance test for future incidents.
+
+SR-01 excludes:
+
+- Portal AIOS Integration Merge.
+- Search Gateway real provider binding.
+- Broad P1 capability merges.
+- Bulk patch merge or branch deletion.
+
+## Required Checks
+
+### Branch sync path
+
+Verify branch_sync source of truth:
+
+```text
+git ls-remote / bare repository refs
+```
+
+The Gitea branch API must not be the mandatory source of truth for synchronization.
+
+### Gitea consistency path
+
+Verify equality between:
+
+```text
+bare repository refs count
+Gitea branch live metadata count
+Gitea branch API paginated visible count
+```
+
+Any mismatch must be treated as reliability drift.
+
+### Safe-mode behavior
+
+When drift is detected:
+
+1. Stop autonomous mutations.
+2. Preserve read-only diagnostics.
+3. Emit safe-mode marker.
+4. Require controlled recovery before mutation resumes.
+
+## Acceptance Criteria
+
+SR-01 is complete when all of the following are true:
+
+1. branch_sync continues to work without Gitea branch API pagination dependency.
+2. Gitea branch metadata drift can be detected deterministically.
+3. Drift causes autonomous safe mode instead of continued mutation.
+4. Recovery procedure includes minimal repair, not default restart.
+5. Runtime acceptance passes:
+
+```text
+getRuntimeStatus
+    ->
+getBaseline
+    ->
+createBranch
+    ->
+createPatch
+```
+
+Expected results:
+
+- runtime status is active;
+- baseline reads V1R4C3 or later;
+- createBranch returns commit/rev/sync status;
+- createPatch returns commit/rev/sync status.
+
+## Execution Order
+
+1. Open SR-01 patch branch.
+2. Review V1R4C3 branch_sync behavior.
+3. Integrate Gitea branch consistency sentinel and safe-mode gate.
+4. Run runtime acceptance.
+5. If clean, prepare V1R4C4 reliability baseline proposal.
+
+## Follow-on Gates
+
+SR-02 Watcher State Cleanup may start after SR-01 acceptance is green.
+
+SR-03 Search Gateway Real Provider Binding may start after SR-01 confirms GitOps/Gitea reliability is stable enough for non-P0 work.
+
+SR-04 Portal AIOS Integration Merge remains deferred pending explicit approval.
+
+## Safety Rules
+
+- Do not bulk merge old patches.
+- Do not delete remote branches during SR-01 unless the branch consistency sentinel is already deployed and healthy.
+- Do not restart Gitea unless process-level failure is proven.
+- Do not modify Gitea DB without backup and owner/admin approval.

ops/systemd/gitea-branch-consistency-sentinel.timer

@@ -0,0 +1,11 @@
+[Unit]
+Description=Run Gitea branch consistency sentinel periodically
+
+[Timer]
+OnBootSec=2min
+OnUnitActiveSec=1min
+RandomizedDelaySec=10s
+Persistent=true
+
+[Install]
+WantedBy=timers.target

scripts/reliability/autonomous_safe_mode.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Safe-mode gate for autonomous mutations.
+
+Controllers should run `check` before writes. Exit code 75 means the
+controller must stop mutation and wait for controlled recovery.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime
+import json
+import os
+from pathlib import Path
+import sys
+
+DEFAULT = "/data/aios-gitops/runtime/state/autonomous-safe-mode.json"
+
+
+def now():
+    return datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def load(path):
+    if not path.exists():
+        return None
+    with path.open("r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--safe-mode-file", default=os.environ.get("AUTONOMOUS_SAFE_MODE_FILE", DEFAULT))
+    parser.add_argument("command", choices=["check", "enter", "clear"])
+    parser.add_argument("--reason", default="")
+    args = parser.parse_args()
+
+    path = Path(args.safe_mode_file)
+
+    if args.command == "check":
+        marker = load(path)
+        if marker:
+            print(json.dumps(marker, ensure_ascii=False, indent=2))
+            return 75
+        return 0
+
+    if args.command == "enter":
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps({
+            "status": "active",
+            "entered_at": now(),
+            "reason": args.reason or "manual",
+        }, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+        return 0
+
+    if args.command == "clear":
+        if path.exists():
+            path.unlink()
+        return 0
+
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/reliability/gitea_branch_consistency_sentinel.py

@@ -0,0 +1,414 @@
+#!/usr/bin/env python3
+"""
+Gitea branch metadata consistency sentinel.
+
+Compares three views of a repository's branch state:
+1. Bare Git refs under refs/heads
+2. Gitea SQLite branch metadata rows with is_deleted=0
+3. Gitea branch API pagination results
+
+On drift or check failure, optionally writes an autonomous safe-mode marker so
+controllers can stop mutating state before the incident expands.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime as _dt
+import json
+import os
+from pathlib import Path
+import shlex
+import sqlite3
+import subprocess
+import sys
+import tempfile
+from typing import Any
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+DEFAULT_REPO_DIR = "/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios.git"
+DEFAULT_DB_PATH = "/data/aios-gitops/runtime/gitea-bin/data/gitea.db"
+DEFAULT_API_URL = "http://127.0.0.1:3000/api/v1/repos/aios/aios/branches"
+DEFAULT_STATE_DIR = "/data/aios-gitops/runtime/state"
+DEFAULT_SAFE_MODE_FILE = f"{DEFAULT_STATE_DIR}/autonomous-safe-mode.json"
+DEFAULT_REPORT_FILE = f"{DEFAULT_STATE_DIR}/gitea-branch-consistency-report.json"
+DEFAULT_DOCTOR_CMD = (
+    "/data/aios-gitops/runtime/bin/gitea "
+    "-C /data/aios-gitops/runtime/gitea-bin/custom "
+    "doctor check --run check-db-consistency"
+)
+
+
+def utc_now() -> str:
+    return _dt.datetime.now(_dt.timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def env_bool(name: str, default: bool = False) -> bool:
+    value = os.environ.get(name)
+    if value is None or value == "":
+        return default
+    return value.strip().lower() in {"1", "true", "yes", "y", "on"}
+
+
+def atomic_write_json(path: Path, payload: dict[str, Any], mode: int = 0o644) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True)
+            handle.write("\n")
+        os.chmod(tmp_name, mode)
+        os.replace(tmp_name, path)
+    finally:
+        try:
+            os.unlink(tmp_name)
+        except FileNotFoundError:
+            pass
+
+
+def run_command(command: list[str], timeout_seconds: int) -> dict[str, Any]:
+    try:
+        proc = subprocess.run(
+            command,
+            check=False,
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            timeout=timeout_seconds,
+        )
+        return {
+            "ok": proc.returncode == 0,
+            "code": proc.returncode,
+            "stdout": proc.stdout.strip(),
+            "stderr": proc.stderr.strip(),
+            "cmd": command,
+        }
+    except FileNotFoundError as exc:
+        return {"ok": False, "code": 127, "stdout": "", "stderr": str(exc), "cmd": command}
+    except subprocess.TimeoutExpired as exc:
+        return {
+            "ok": False,
+            "code": 124,
+            "stdout": (exc.stdout or "").strip() if isinstance(exc.stdout, str) else "",
+            "stderr": (exc.stderr or "").strip() if isinstance(exc.stderr, str) else "",
+            "cmd": command,
+            "timeout_seconds": timeout_seconds,
+        }
+
+
+def git_branch_names(repo_dir: str, timeout_seconds: int) -> tuple[set[str], dict[str, Any]]:
+    command = ["git", "--git-dir", repo_dir, "for-each-ref", "--format=%(refname:short)", "refs/heads"]
+    result = run_command(command, timeout_seconds)
+    if not result["ok"]:
+        raise RuntimeError(f"git branch ref scan failed: {result['stderr'] or result['stdout']}")
+    names = {line.strip() for line in result["stdout"].splitlines() if line.strip()}
+    result["count"] = len(names)
+    return names, result
+
+
+def sqlite_branch_names(db_path: str, repo_id: int) -> tuple[set[str], dict[str, Any]]:
+    path = Path(db_path)
+    if not path.exists():
+        raise RuntimeError(f"Gitea database does not exist: {db_path}")
+
+    uri = f"file:{urllib.parse.quote(str(path.resolve()))}?mode=ro"
+    try:
+        conn = sqlite3.connect(uri, uri=True, timeout=5)
+        try:
+            cursor = conn.execute(
+                "SELECT name FROM branch WHERE repo_id = ? AND is_deleted = 0 ORDER BY name",
+                (repo_id,),
+            )
+            names = {row[0] for row in cursor.fetchall()}
+        finally:
+            conn.close()
+    except sqlite3.Error as exc:
+        raise RuntimeError(f"SQLite branch metadata query failed: {exc}") from exc
+
+    return names, {"ok": True, "repo_id": repo_id, "count": len(names), "db_path": db_path}
+
+
+def add_query_params(url: str, params: dict[str, Any]) -> str:
+    split = urllib.parse.urlsplit(url)
+    query = dict(urllib.parse.parse_qsl(split.query, keep_blank_values=True))
+    query.update({key: str(value) for key, value in params.items()})
+    return urllib.parse.urlunsplit(
+        (split.scheme, split.netloc, split.path, urllib.parse.urlencode(query), split.fragment)
+    )
+
+
+def read_api_token() -> str | None:
+    token = os.environ.get("GITEA_API_TOKEN")
+    if token:
+        return token.strip()
+
+    token_file = os.environ.get("GITEA_API_TOKEN_FILE")
+    if token_file:
+        token_path = Path(token_file)
+        if token_path.exists():
+            return token_path.read_text(encoding="utf-8").strip()
+
+    return None
+
+
+def api_branch_names(
+    api_url: str,
+    limit: int,
+    timeout_seconds: int,
+    max_pages: int,
+) -> tuple[set[str], dict[str, Any]]:
+    token = read_api_token()
+    names: list[str] = []
+    pages: list[dict[str, Any]] = []
+
+    for page in range(1, max_pages + 1):
+        url = add_query_params(api_url, {"page": page, "limit": limit})
+        headers = {"Accept": "application/json"}
+        if token:
+            headers["Authorization"] = f"token {token}"
+
+        request = urllib.request.Request(url, headers=headers, method="GET")
+        try:
+            with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+                body = response.read()
+                status = getattr(response, "status", response.getcode())
+        except urllib.error.HTTPError as exc:
+            detail = exc.read().decode("utf-8", errors="replace")
+            raise RuntimeError(f"Gitea branch API HTTP {exc.code}: {detail[:500]}") from exc
+        except urllib.error.URLError as exc:
+            raise RuntimeError(f"Gitea branch API request failed: {exc}") from exc
+        except TimeoutError as exc:
+            raise RuntimeError(f"Gitea branch API request timed out: {exc}") from exc
+
+        try:
+            payload = json.loads(body.decode("utf-8"))
+        except json.JSONDecodeError as exc:
+            raise RuntimeError(f"Gitea branch API returned invalid JSON on page {page}") from exc
+
+        if not isinstance(payload, list):
+            raise RuntimeError(f"Gitea branch API returned non-list payload on page {page}: {type(payload).__name__}")
+
+        page_names = []
+        for item in payload:
+            if not isinstance(item, dict) or "name" not in item:
+                raise RuntimeError(f"Gitea branch API returned malformed branch item on page {page}: {item!r}")
+            page_names.append(str(item["name"]))
+
+        names.extend(page_names)
+        pages.append({"page": page, "status": status, "count": len(page_names)})
+
+        if len(page_names) < limit:
+            break
+    else:
+        raise RuntimeError(f"Gitea branch API exceeded max_pages={max_pages}; refusing to trust partial result")
+
+    return set(names), {
+        "ok": True,
+        "api_url": api_url,
+        "limit": limit,
+        "pages": pages,
+        "raw_count": len(names),
+        "unique_count": len(set(names)),
+    }
+
+
+def doctor_check(command_text: str, timeout_seconds: int, required: bool) -> dict[str, Any]:
+    command_text = command_text.strip()
+    if not command_text:
+        return {"ok": True, "skipped": True, "reason": "empty GITEA_DOCTOR_CMD"}
+
+    command = shlex.split(command_text)
+    if not command:
+        return {"ok": True, "skipped": True, "reason": "empty parsed GITEA_DOCTOR_CMD"}
+
+    if not Path(command[0]).exists() and "/" in command[0]:
+        return {
+            "ok": not required,
+            "skipped": True,
+            "required": required,
+            "reason": f"doctor binary not found: {command[0]}",
+            "cmd": command,
+        }
+
+    result = run_command(command, timeout_seconds)
+    result["required"] = required
+    return result
+
+
+def compare_sets(left_name: str, left: set[str], right_name: str, right: set[str]) -> dict[str, Any]:
+    return {
+        f"{left_name}_missing_from_{right_name}": sorted(left - right),
+        f"{right_name}_missing_from_{left_name}": sorted(right - left),
+    }
+
+
+def build_report(args: argparse.Namespace) -> dict[str, Any]:
+    checks: dict[str, Any] = {}
+    errors: list[str] = []
+
+    git_names: set[str] = set()
+    db_names: set[str] = set()
+    api_names: set[str] = set()
+
+    try:
+        git_names, checks["git_refs"] = git_branch_names(args.repo_dir, args.timeout)
+    except Exception as exc:  # noqa: BLE001 - must preserve diagnostic text
+        errors.append(str(exc))
+        checks["git_refs"] = {"ok": False, "error": str(exc), "repo_dir": args.repo_dir}
+
+    try:
+        db_names, checks["sqlite_branch_metadata"] = sqlite_branch_names(args.db_path, args.repo_id)
+    except Exception as exc:  # noqa: BLE001
+        errors.append(str(exc))
+        checks["sqlite_branch_metadata"] = {"ok": False, "error": str(exc), "db_path": args.db_path}
+
+    if not args.skip_api:
+        try:
+            api_names, checks["gitea_branch_api"] = api_branch_names(
+                args.api_url,
+                args.api_limit,
+                args.timeout,
+                args.max_pages,
+            )
+        except Exception as exc:  # noqa: BLE001
+            errors.append(str(exc))
+            checks["gitea_branch_api"] = {"ok": False, "error": str(exc), "api_url": args.api_url}
+    else:
+        checks["gitea_branch_api"] = {"ok": True, "skipped": True}
+
+    if not args.skip_doctor:
+        checks["gitea_doctor"] = doctor_check(args.doctor_cmd, args.doctor_timeout, args.doctor_required)
+        if not checks["gitea_doctor"].get("ok", False):
+            errors.append("gitea doctor consistency check failed")
+    else:
+        checks["gitea_doctor"] = {"ok": True, "skipped": True}
+
+    drift: dict[str, Any] = {}
+    if checks.get("git_refs", {}).get("ok") and checks.get("sqlite_branch_metadata", {}).get("ok"):
+        drift.update(compare_sets("git", git_names, "db", db_names))
+    if checks.get("git_refs", {}).get("ok") and checks.get("gitea_branch_api", {}).get("ok") and not args.skip_api:
+        drift.update(compare_sets("git", git_names, "api", api_names))
+    if checks.get("sqlite_branch_metadata", {}).get("ok") and checks.get("gitea_branch_api", {}).get("ok") and not args.skip_api:
+        drift.update(compare_sets("db", db_names, "api", api_names))
+
+    drift_counts = {key: len(value) for key, value in drift.items()}
+    drift_found = any(drift_counts.values())
+
+    counts = {
+        "git_refs": len(git_names) if checks.get("git_refs", {}).get("ok") else None,
+        "db_live_branches": len(db_names) if checks.get("sqlite_branch_metadata", {}).get("ok") else None,
+        "api_visible_branches": len(api_names)
+        if checks.get("gitea_branch_api", {}).get("ok") and not args.skip_api
+        else None,
+    }
+
+    unhealthy_checks = [
+        name for name, result in checks.items()
+        if not result.get("ok", False)
+    ]
+
+    status = "unhealthy" if errors or drift_found or unhealthy_checks else "healthy"
+
+    return {
+        "service": "gitea-branch-consistency-sentinel",
+        "status": status,
+        "generated_at": utc_now(),
+        "repo": {
+            "repo_dir": args.repo_dir,
+            "db_path": args.db_path,
+            "repo_id": args.repo_id,
+            "api_url": args.api_url,
+        },
+        "counts": counts,
+        "checks": checks,
+        "drift": drift,
+        "drift_counts": drift_counts,
+        "errors": errors,
+        "unhealthy_checks": unhealthy_checks,
+    }
+
+
+def write_safe_mode_if_needed(report: dict[str, Any], args: argparse.Namespace) -> None:
+    safe_file = Path(args.safe_mode_file)
+    report_file = Path(args.report_file)
+
+    if report["status"] != "healthy":
+        if args.emit_safe_mode:
+            marker = {
+                "service": "autonomous-safe-mode",
+                "status": "active",
+                "entered_at": utc_now(),
+                "reason": "gitea_branch_consistency_unhealthy",
+                "source": "gitea-branch-consistency-sentinel",
+                "report_file": str(report_file),
+                "counts": report.get("counts", {}),
+                "drift_counts": report.get("drift_counts", {}),
+                "unhealthy_checks": report.get("unhealthy_checks", []),
+                "operator_action": (
+                    "Stop autonomous mutations, inspect Gitea refs/branch metadata/API drift, "
+                    "repair minimal inconsistent state, then run runtime acceptance tests."
+                ),
+            }
+            atomic_write_json(safe_file, marker)
+        return
+
+    if args.clear_safe_mode_if_healthy and safe_file.exists():
+        history_dir = safe_file.parent / "safe-mode-history"
+        history_dir.mkdir(parents=True, exist_ok=True)
+        archive = history_dir / f"{safe_file.stem}.cleared-{utc_now().replace(':', '').replace('-', '')}.json"
+        safe_file.replace(archive)
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Detect Gitea branch metadata drift and emit safe mode markers.")
+    parser.add_argument("--repo-dir", default=os.environ.get("GITEA_REPO_DIR", DEFAULT_REPO_DIR))
+    parser.add_argument("--db-path", default=os.environ.get("GITEA_DB_PATH", DEFAULT_DB_PATH))
+    parser.add_argument("--repo-id", type=int, default=int(os.environ.get("GITEA_REPO_ID", "1")))
+    parser.add_argument("--api-url", default=os.environ.get("GITEA_API_URL", DEFAULT_API_URL))
+    parser.add_argument("--api-limit", type=int, default=int(os.environ.get("GITEA_BRANCH_API_LIMIT", "50")))
+    parser.add_argument("--max-pages", type=int, default=int(os.environ.get("GITEA_BRANCH_API_MAX_PAGES", "200")))
+    parser.add_argument("--timeout", type=int, default=int(os.environ.get("GITEA_SENTINEL_TIMEOUT_SECONDS", "10")))
+    parser.add_argument("--doctor-timeout", type=int, default=int(os.environ.get("GITEA_DOCTOR_TIMEOUT_SECONDS", "45")))
+    parser.add_argument("--doctor-cmd", default=os.environ.get("GITEA_DOCTOR_CMD", DEFAULT_DOCTOR_CMD))
+    parser.add_argument("--doctor-required", action="store_true", default=env_bool("GITEA_DOCTOR_REQUIRED", False))
+    parser.add_argument("--skip-api", action="store_true", default=env_bool("GITEA_SENTINEL_SKIP_API", False))
+    parser.add_argument("--skip-doctor", action="store_true", default=env_bool("GITEA_SENTINEL_SKIP_DOCTOR", False))
+    parser.add_argument("--report-file", default=os.environ.get("GITEA_SENTINEL_REPORT_FILE", DEFAULT_REPORT_FILE))
+    parser.add_argument("--safe-mode-file", default=os.environ.get("AUTONOMOUS_SAFE_MODE_FILE", DEFAULT_SAFE_MODE_FILE))
+    parser.add_argument("--emit-safe-mode", action="store_true", help="Write safe-mode marker when unhealthy.")
+    parser.add_argument(
+        "--clear-safe-mode-if-healthy",
+        action="store_true",
+        help="Archive the existing safe-mode marker when all checks are healthy. Use only during controlled recovery.",
+    )
+    parser.add_argument("--json", action="store_true", help="Print full JSON report.")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str]) -> int:
+    args = parse_args(argv)
+    report = build_report(args)
+
+    report_path = Path(args.report_file)
+    atomic_write_json(report_path, report)
+    write_safe_mode_if_needed(report, args)
+
+    if args.json or report["status"] != "healthy":
+        json.dump(report, sys.stdout, ensure_ascii=False, indent=2, sort_keys=True)
+        sys.stdout.write("\n")
+    else:
+        print(
+            "healthy: "
+            f"git={report['counts']['git_refs']} "
+            f"db={report['counts']['db_live_branches']} "
+            f"api={report['counts']['api_visible_branches']}"
+        )
+
+    return 0 if report["status"] == "healthy" else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))

tests/test_autonomous_controller.py

@@ -0,0 +1,347 @@
+#!/usr/bin/env python3
+"""Tests for autonomous_controller.py"""
+
+import json
+import time
+from pathlib import Path
+
+import pytest
+
+from agents.scripts.autonomous_controller import (
+    h_branch_sync,
+    h_stuck,
+    diagnose,
+    create_patch,
+    apply_approved_patches,
+)
+
+
+@pytest.fixture
+def tmp_workspace(tmp_path, monkeypatch):
+    """Set up a temporary workspace and patch module-level paths."""
+    base = tmp_path / "workspace"
+    pending = base / "agents/patches/pending/"
+    approved = base / "agents/patches/approved/"
+    applied = base / "agents/patches/applied/"
+    rejected = base / "agents/patches/rejected/"
+    running = base / "agents/commands/running/"
+    failed = base / "agents/commands/failed/"
+    audit = base / "audit/controller-runs/"
+    state_file = base / "agents/state/controller-state.json"
+    health_file = base / "agents/state/health.json"
+    notif_dir = base / "agents/notifications/"
+    active = base / "agents/commands/active/"
+    completed = base / "agents/commands/completed/"
+
+    for d in [pending, approved, applied, rejected, running, failed, audit, notif_dir, active, completed]:
+        d.mkdir(parents=True, exist_ok=True)
+
+    import agents.scripts.autonomous_controller as ctrl
+    monkeypatch.setattr(ctrl, "WORKSPACE", base)
+    monkeypatch.setattr(ctrl, "PENDING_DIR", pending)
+    monkeypatch.setattr(ctrl, "APPROVED_DIR", approved)
+    monkeypatch.setattr(ctrl, "APPLIED_DIR", applied)
+    monkeypatch.setattr(ctrl, "REJECTED_DIR", rejected)
+    monkeypatch.setattr(ctrl, "RUNNING_DIR", running)
+    monkeypatch.setattr(ctrl, "FAILED_DIR", failed)
+    monkeypatch.setattr(ctrl, "AUDIT_DIR", audit)
+    monkeypatch.setattr(ctrl, "STATE_FILE", state_file)
+    monkeypatch.setattr(ctrl, "HEALTH_FILE", health_file)
+    monkeypatch.setattr(ctrl, "NOTIF_DIR", notif_dir)
+    monkeypatch.setattr(ctrl, "ACTIVE_DIR", active)
+    monkeypatch.setattr(ctrl, "COMPLETED_DIR", completed)
+
+    yield base
+
+
+class TestHBranchSync:
+    """Tests for h_branch_sync() health check."""
+
+    def test_all_branches_synced(self, tmp_workspace, monkeypatch):
+        """Bare repo branches match Gitea DB branches exactly."""
+        import agents.scripts.autonomous_controller as ctrl
+        import sqlite3
+
+        bare_path = tmp_workspace / "bare_repo.git"
+        db_path = tmp_workspace / "gitea.db"
+        bare_path.mkdir(parents=True, exist_ok=True)
+        db_path.touch()
+
+        monkeypatch.setattr(ctrl, "BARE_REPO_PATH", bare_path)
+        monkeypatch.setattr(ctrl, "GITEA_DB_PATH", db_path)
+        monkeypatch.setattr(ctrl, "REPO_ID", 1)
+
+        def mock_run(args, **kwargs):
+            class Result:
+                returncode = 0
+                stdout = "abc12345\trefs/heads/main\n67890abc\trefs/heads/feature-x\n"
+                stderr = ""
+            return Result()
+
+        monkeypatch.setattr(ctrl.subprocess, "run", mock_run)
+
+        def mock_connect(*a, **kw):
+            class FakeCursor:
+                def execute(self, sql, params):
+                    return self
+                def fetchall(self):
+                    return [("main",), ("feature-x",)]
+            class FakeConn:
+                def __init__(self):
+                    self.closed = False
+                def execute(self, sql, params):
+                    return FakeCursor()
+                def close(self):
+                    self.closed = True
+            return FakeConn()
+        monkeypatch.setattr(sqlite3, "connect", mock_connect)
+
+        result = h_branch_sync()
+        assert result["ok"] is True
+        assert result["bare"] == 2
+        assert result["gitea"] == 2
+        assert result["missing"] == []
+
+    def test_branches_missing_from_gitea(self, tmp_workspace, monkeypatch):
+        """Some branches exist in bare repo but not in Gitea DB."""
+        import agents.scripts.autonomous_controller as ctrl
+        import sqlite3
+
+        bare_path = tmp_workspace / "bare_repo.git"
+        db_path = tmp_workspace / "gitea.db"
+        bare_path.mkdir(parents=True, exist_ok=True)
+        db_path.touch()
+
+        monkeypatch.setattr(ctrl, "BARE_REPO_PATH", bare_path)
+        monkeypatch.setattr(ctrl, "GITEA_DB_PATH", db_path)
+        monkeypatch.setattr(ctrl, "REPO_ID", 1)
+
+        def mock_run(args, **kwargs):
+            class Result:
+                returncode = 0
+                stdout = "abc12345\trefs/heads/main\n67890abc\trefs/heads/feature-a\n11111bb\trefs/heads/feature-b\n"
+                stderr = ""
+            return Result()
+
+        monkeypatch.setattr(ctrl.subprocess, "run", mock_run)
+
+        def mock_connect(*a, **kw):
+            class FakeCursor:
+                def execute(self, sql, params):
+                    return self
+                def fetchall(self):
+                    return [("main",), ("feature-a",)]
+            class FakeConn:
+                def __init__(self):
+                    self.closed = False
+                def execute(self, sql, params):
+                    return FakeCursor()
+                def close(self):
+                    self.closed = True
+            return FakeConn()
+        monkeypatch.setattr(sqlite3, "connect", mock_connect)
+
+        result = h_branch_sync()
+        assert result["ok"] is False
+        assert result["bare"] == 3
+        assert result["gitea"] == 2
+        assert "feature-b" in result["missing"]
+
+    def test_git_failure(self, tmp_workspace, monkeypatch):
+        """Git command fails."""
+        import agents.scripts.autonomous_controller as ctrl
+
+        bare_path = tmp_workspace / "bare_repo.git"
+        db_path = tmp_workspace / "gitea.db"
+        bare_path.mkdir(parents=True, exist_ok=True)
+        db_path.touch()
+
+        monkeypatch.setattr(ctrl, "BARE_REPO_PATH", bare_path)
+        monkeypatch.setattr(ctrl, "GITEA_DB_PATH", db_path)
+
+        def mock_run(args, **kwargs):
+            class Result:
+                returncode = 1
+                stderr = "fatal: not a git repository"
+            return Result()
+
+        monkeypatch.setattr(ctrl.subprocess, "run", mock_run)
+
+        result = h_branch_sync()
+        assert result["ok"] is False
+        assert "err" in result
+
+    def test_db_failure(self, tmp_workspace, monkeypatch):
+        """SQLite query fails."""
+        import agents.scripts.autonomous_controller as ctrl
+        import sqlite3
+
+        bare_path = tmp_workspace / "bare_repo.git"
+        db_path = tmp_workspace / "gitea.db"
+        bare_path.mkdir(parents=True, exist_ok=True)
+        db_path.touch()
+
+        monkeypatch.setattr(ctrl, "BARE_REPO_PATH", bare_path)
+        monkeypatch.setattr(ctrl, "GITEA_DB_PATH", db_path)
+
+        def mock_run(args, **kwargs):
+            class Result:
+                returncode = 0
+                stdout = "abc12345\trefs/heads/main\n"
+                stderr = ""
+            return Result()
+
+        monkeypatch.setattr(ctrl.subprocess, "run", mock_run)
+
+        def mock_connect(uri, **kwargs):
+            raise Exception("Fake DB error")
+
+        monkeypatch.setattr(sqlite3, "connect", mock_connect)
+
+        result = h_branch_sync()
+        assert result["ok"] is False
+        assert "err" in result
+
+
+class TestHStuck:
+    def test_no_stuck_commands(self, tmp_workspace):
+        """When there are no commands in the running dir, h_stuck returns ok=True."""
+        result = h_stuck()
+        assert result["ok"] is True
+        assert len(result["stuck"]) == 0
+
+    def test_stuck_commands_found(self, tmp_workspace):
+        """Stuck commands older than 30 minutes should be detected."""
+        stuck_time = time.time() - 31 * 60
+        import agents.scripts.autonomous_controller as ctrl
+        running_dir = ctrl.RUNNING_DIR
+
+        for name in ["stuck_cmd.md", "another_cmd.md"]:
+            f = running_dir / name
+            f.write_text(json.dumps({"ts": stuck_time, "name": name}))
+            import os
+            os.utime(f, (stuck_time, stuck_time))
+
+        result = h_stuck()
+        assert result["ok"] is False
+        assert len(result["stuck"]) == 2
+
+    def test_recent_commands_not_stuck(self, tmp_workspace):
+        """Commands less than 30 minutes old should not be considered stuck."""
+        recent_time = time.time() - 15 * 60
+        import agents.scripts.autonomous_controller as ctrl
+        running_dir = ctrl.RUNNING_DIR
+        f = running_dir / "recent_cmd.md"
+        f.write_text(json.dumps({"ts": recent_time, "name": "recent_cmd.md"}))
+        import os
+        os.utime(f, (recent_time, recent_time))
+        result = h_stuck()
+        assert result["ok"] is True
+        assert len(result["stuck"]) == 0
+
+
+class TestDiagnose:
+    def test_branch_sync_failure(self, tmp_workspace):
+        result = diagnose({
+            "branch_sync": {"ok": False, "missing": ["br1", "br2"]}
+        })
+        assert len(result) == 1
+        assert result[0]["check"] == "branch_sync"
+        assert result[0]["severity"] == "high"
+        assert "branches missing" in result[0]["diagnosis"]
+        assert result[0]["fix"] == "sync_branches"
+
+    def test_stuck_failure(self, tmp_workspace):
+        result = diagnose({
+            "stuck": {"ok": False, "stuck": [{"n": "cmd1.md", "age": 35.0}]}
+        })
+        assert len(result) == 1
+        assert result[0]["check"] == "stuck"
+        assert result[0]["severity"] == "medium"
+        assert "commands stuck" in result[0]["diagnosis"]
+        assert result[0]["fix"] == "move_to_failed"
+
+    def test_both_failures(self, tmp_workspace):
+        result = diagnose({
+            "branch_sync": {"ok": False, "missing": ["br1"]},
+            "stuck": {"ok": False, "stuck": [{"n": "cmd1.md", "age": 35.0}]}
+        })
+        assert len(result) == 2
+        checks = [r["check"] for r in result]
+        assert "branch_sync" in checks
+        assert "stuck" in checks
+
+    def test_all_healthy(self, tmp_workspace):
+        result = diagnose({
+            "branch_sync": {"ok": True},
+            "stuck": {"ok": True}
+        })
+        assert len(result) == 0
+
+
+class TestCreatePatch:
+    def test_creates_patch_file(self, tmp_workspace):
+        import agents.scripts.autonomous_controller as ctrl
+        pid = ctrl.PENDING_DIR
+        result = create_patch("my-title", "my-content")
+        assert result["ok"] is True
+        files = list(pid.iterdir())
+        assert len(files) == 1
+        content = json.loads(files[0].read_text())
+        assert content["title"] == "my-title"
+        assert content["content"] == "my-content"
+
+    def test_returns_metadata(self, tmp_workspace):
+        result = create_patch("title", "body")
+        assert "id" in result
+        assert "title" in result
+        assert "content" in result
+        assert result["title"] == "title"
+        assert result["content"] == "body"
+
+
+class TestApplyApprovedPatches:
+    def test_apply_one_patch(self, tmp_workspace):
+        import agents.scripts.autonomous_controller as ctrl
+        approved = ctrl.APPROVED_DIR
+        f = approved / "patch01.json"
+        f.write_text(json.dumps({"check": "branch_sync", "fix": "sync_branches", "branches": []}))
+
+        result = apply_approved_patches()
+        assert len(result) == 1
+        assert result[0]["check"] == "branch_sync"
+
+    def test_move_to_applied(self, tmp_workspace):
+        import agents.scripts.autonomous_controller as ctrl
+        approved = ctrl.APPROVED_DIR
+        f = approved / "patch02.json"
+        f.write_text(json.dumps({"check": "stuck", "fix": "move_to_failed", "stuck": []}))
+        applied_dir = ctrl.APPLIED_DIR
+
+        result = apply_approved_patches()
+        assert len(result) == 1
+        # Verify file was moved to APPLIED_DIR
+        assert not list(approved.iterdir()) or all(p.name.endswith(".md") for p in approved.iterdir())
+        applied_files = list(applied_dir.iterdir())
+        assert len(applied_files) == 1
+
+    def test_apply_failure_handling(self, tmp_workspace):
+        import agents.scripts.autonomous_controller as ctrl
+        approved = ctrl.APPROVED_DIR
+        f = approved / "patch_fail.json"
+        # Write invalid JSON to cause failure
+        f.write_text("not valid json")
+
+        rejected = ctrl.REJECTED_DIR
+        result = apply_approved_patches()
+        assert len(result) == 0
+        # Should be moved to rejected dir
+        rejected_files = list(rejected.iterdir())
+        assert len(rejected_files) == 1
+
+    def test_empty_approved_dir(self, tmp_workspace):
+        import agents.scripts.autonomous_controller as ctrl
+        # Approved dir is empty from fixture
+        result = apply_approved_patches()
+        assert len(result) == 0
+

tests/test_evolution_engine.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python3
+"""
+Tests for evolution_engine.py (Closed-loop self-evolution engine).
+"""
+
+import sys
+import importlib.util
+from pathlib import Path
+
+_m = Path(__file__).resolve().parent.parent / 'runtime' / 'agent-os' / 'evolution_engine.py'
+_spec = importlib.util.spec_from_file_location('runtime.agent_os.evolution_engine', str(_m))
+_mod = importlib.util.module_from_spec(_spec)
+sys.modules['runtime.agent_os.evolution_engine'] = _mod
+_spec.loader.exec_module(_mod)
+
+EvolutionEngine = _mod.EvolutionEngine
+EvolutionStage = _mod.EvolutionStage
+EvolutionRun = _mod.EvolutionRun
+
+
+class TestEvolutionEngine:
+
+    def test_engine_has_all_stages(self):
+        engine = EvolutionEngine()
+        assert len(engine.ORDER) == 8
+        assert engine.ORDER == [
+            EvolutionStage.DETECT,
+            EvolutionStage.PLAN,
+            EvolutionStage.SCHEDULE,
+            EvolutionStage.EXECUTE,
+            EvolutionStage.VALIDATE,
+            EvolutionStage.VOTE,
+            EvolutionStage.PATCH,
+            EvolutionStage.LEARN,
+        ]
+
+    def test_run_cycle_basic(self):
+        engine = EvolutionEngine()
+        result = engine.run_cycle()
+        assert "run_id" in result
+        assert result["status"] == "completed"
+        assert len(result["timeline"]) == 8
+        for entry in result["timeline"]:
+            assert "stage" in entry
+            assert "result" in entry
+            assert "ts" in entry
+
+    def test_run_cycle_with_context(self):
+        engine = EvolutionEngine()
+        ctx = {"signals": ["err_001"], "plan": ["fix_branch_sync"]}
+        result = engine.run_cycle(ctx)
+        detect_timeline = result["timeline"][0]
+        assert detect_timeline["stage"] == "detect"
+        assert detect_timeline["result"]["signals"] == ["err_001"]
+
+    def test_custom_handler_detect(self):
+        engine = EvolutionEngine()
+        def custom_detect(run):
+            return {"detected": True, "custom": True}
+        engine.register_handler(EvolutionStage.DETECT, custom_detect)
+        result = engine.run_cycle()
+        detect_result = result["timeline"][0]["result"]
+        assert detect_result["custom"] is True
+
+    def test_custom_handler_plan(self):
+        engine = EvolutionEngine()
+        def custom_plan(run):
+            return {"plan": ["step1", "step2"]}
+        engine.register_handler(EvolutionStage.PLAN, custom_plan)
+        result = engine.run_cycle()
+        plan_result = result["timeline"][1]["result"]
+        assert plan_result["plan"] == ["step1", "step2"]
+
+    def test_custom_handler_schedule(self):
+        engine = EvolutionEngine()
+        def custom_schedule(run):
+            return {"scheduled": False, "reason": "no capacity"}
+        engine.register_handler(EvolutionStage.SCHEDULE, custom_schedule)
+        result = engine.run_cycle()
+        sched_result = result["timeline"][2]["result"]
+        assert sched_result["scheduled"] is False
+
+    def test_custom_handler_execute(self):
+        engine = EvolutionEngine()
+        def custom_execute(run):
+            return {"executed": False, "error": "timeout"}
+        engine.register_handler(EvolutionStage.EXECUTE, custom_execute)
+        result = engine.run_cycle()
+        exec_result = result["timeline"][3]["result"]
+        assert exec_result["executed"] is False
+
+    def test_custom_handler_validate(self):
+        engine = EvolutionEngine()
+        def custom_validate(run):
+            return {"valid": False, "issues": ["test_failed"]}
+        engine.register_handler(EvolutionStage.VALIDATE, custom_validate)
+        result = engine.run_cycle()
+        val_result = result["timeline"][4]["result"]
+        assert val_result["valid"] is False
+
+    def test_custom_handler_vote(self):
+        engine = EvolutionEngine()
+        def custom_vote(run):
+            return {"decision": "rejected", "reason": "not ready"}
+        engine.register_handler(EvolutionStage.VOTE, custom_vote)
+        result = engine.run_cycle()
+        vote_result = result["timeline"][5]["result"]
+        assert vote_result["decision"] == "rejected"
+
+    def test_custom_handler_patch(self):
+        engine = EvolutionEngine()
+        def custom_patch(run):
+            return {"patch_ready": False, "build_error": "merge conflict"}
+        engine.register_handler(EvolutionStage.PATCH, custom_patch)
+        result = engine.run_cycle()
+        patch_result = result["timeline"][6]["result"]
+        assert patch_result["patch_ready"] is False
+
+    def test_custom_handler_learn(self):
+        engine = EvolutionEngine()
+        def custom_learn(run):
+            return {"learned": False, "summary": "nothing to learn"}
+        engine.register_handler(EvolutionStage.LEARN, custom_learn)
+        result = engine.run_cycle()
+        learn_result = result["timeline"][7]["result"]
+        assert learn_result["learned"] is False
+
+    def test_register_handler_by_string(self):
+        engine = EvolutionEngine()
+        def handler(run):
+            return {"from_string": True}
+        engine.register_handler("detect", handler)
+        result = engine.run_cycle()
+        assert result["timeline"][0]["result"]["from_string"] is True
+
+    def test_run_evolves_stages_in_order(self):
+        engine = EvolutionEngine()
+        def track_stage(run, stage_name):
+            return {"stage": stage_name}
+        engine.register_handler(EvolutionStage.DETECT, lambda r: track_stage(r, "detect"))
+        engine.register_handler(EvolutionStage.PLAN, lambda r: track_stage(r, "plan"))
+        engine.register_handler(EvolutionStage.SCHEDULE, lambda r: track_stage(r, "schedule"))
+        engine.register_handler(EvolutionStage.EXECUTE, lambda r: track_stage(r, "execute"))
+        engine.register_handler(EvolutionStage.VALIDATE, lambda r: track_stage(r, "validate"))
+        engine.register_handler(EvolutionStage.VOTE, lambda r: track_stage(r, "vote"))
+        engine.register_handler(EvolutionStage.PATCH, lambda r: track_stage(r, "patch"))
+        engine.register_handler(EvolutionStage.LEARN, lambda r: track_stage(r, "learn"))
+        result = engine.run_cycle()
+        stages = [entry["result"]["stage"] for entry in result["timeline"]]
+        assert stages == ["detect", "plan", "schedule", "execute", "validate", "vote", "patch", "learn"]
+
+    def test_evolution_run_dataclass(self):
+        run = EvolutionRun()
+        assert run.status == "created"
+        assert run.run_id is not None
+        assert run.context == {}
+        assert run.timeline == []
+        run.record(EvolutionStage.DETECT, {"found": True})
+        assert run.stage == EvolutionStage.DETECT
+        assert len(run.timeline) == 1
+
+    def test_run_cycle_creates_new_run_id(self):
+        engine = EvolutionEngine()
+        result1 = engine.run_cycle()
+        result2 = engine.run_cycle()
+        assert result1["run_id"] != result2["run_id"]
+
+    def test_no_handler_uses_default_values(self):
+        engine = EvolutionEngine()
+        result = engine.run_cycle()
+        assert result["timeline"][0]["result"] == {"detected": True, "signals": []}
+        assert result["timeline"][1]["result"] == {"plan": []}
+        assert result["timeline"][2]["result"] == {"scheduled": True}
+        assert result["timeline"][3]["result"] == {"executed": True}
+        assert result["timeline"][4]["result"] == {"valid": True}
+        assert result["timeline"][5]["result"] == {"decision": "accepted"}
+        assert result["timeline"][6]["result"] == {"patch_ready": True}
+        assert result["timeline"][7]["result"] == {"learned": True}

tests/test_weixin_image_followup_intent.py

@@ -0,0 +1,270 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import subprocess
+from pathlib import Path
+
+import pytest
+
+
+def load_js_process_helpers() -> str:
+    patcher_path = Path(__file__).resolve().parents[1] / 'fixes' / 'patch_openclaw_weixin_hermes.py'
+    assert patcher_path.exists(), f'missing patcher: {patcher_path}'
+    spec = importlib.util.spec_from_file_location('patch_openclaw_weixin_hermes', patcher_path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module.JS_PROCESS_HELPERS
+
+
+
+def load_patcher_module() -> object:
+    patcher_path = Path(__file__).resolve().parents[1] / 'fixes' / 'patch_openclaw_weixin_hermes.py'
+    assert patcher_path.exists(), f'missing patcher: {patcher_path}'
+    spec = importlib.util.spec_from_file_location('patch_openclaw_weixin_hermes', patcher_path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+
+def run_js_process_helpers(tmp_path: Path, body: str) -> dict:
+    js_process_helpers = load_js_process_helpers()
+    node_prelude = '\n'.join(
+        [
+            'const hermesTestFs = require("node:fs");',
+            'globalThis.existsSync = hermesTestFs.existsSync;',
+            'globalThis.readFileSync = hermesTestFs.readFileSync;',
+            'globalThis.writeFileSync = hermesTestFs.writeFileSync;',
+            'globalThis.mkdirSync = hermesTestFs.mkdirSync;',
+            'globalThis.appendFileSync = hermesTestFs.appendFileSync;',
+            'globalThis.path = require("node:path");',
+            'globalThis.logger = { info() {}, warn() {}, error() {} };',
+        ]
+    )
+    node_script = tmp_path / 'hermes_image_followup_check.js'
+    node_script.write_text(node_prelude + '\n\n' + js_process_helpers + '\n\n' + body, encoding='utf-8')
+    completed = subprocess.run(['node', str(node_script)], check=True, capture_output=True, text=True)
+    return json.loads(completed.stdout.strip())
+
+
+@pytest.mark.parametrize('is_ts', [True, False], ids=['ts', 'js'])
+def test_replace_or_insert_process_helpers_is_byte_idempotent(is_ts: bool) -> None:
+    patcher = load_patcher_module()
+    marker = 'export async function processOneMessage'
+    source = marker + '(message) {\n  return message;\n}\n'
+
+    once = patcher.replace_or_insert_process_helpers(source, is_ts=is_ts)
+    twice = patcher.replace_or_insert_process_helpers(once, is_ts=is_ts)
+    helpers = patcher.TS_PROCESS_HELPERS if is_ts else patcher.JS_PROCESS_HELPERS
+
+    assert helpers + '\n\n' + marker in once
+    assert twice == once
+
+
+@pytest.mark.parametrize(
+    ('is_ts', 'hook_name'),
+    [(True, 'TS_PROCESS_HOOK'), (False, 'JS_PROCESS_HOOK')],
+    ids=['ts', 'js'],
+)
+def test_remove_and_reinsert_process_hook_is_byte_idempotent(is_ts: bool, hook_name: str) -> None:
+    patcher = load_patcher_module()
+    hook = getattr(patcher, hook_name)
+    indent = '  ' if is_ts else '    '
+    needle = indent + 'ctx.CommandAuthorized = commandAuthorized;\n'
+    next_nonblank = indent + 'continueProcessing();\n'
+
+    def cycle(text: str) -> str:
+        without_hook = patcher.remove_existing_process_hermes_hook(text)
+        return without_hook.replace(needle, needle + hook + '\n', 1)
+
+    once = cycle(needle + next_nonblank)
+    twice = cycle(once)
+
+    assert hook + '\n' + next_nonblank in once
+    assert twice == once
+
+
+
+def test_looks_like_hermes_image_followup_distinguishes_diagnostics_from_real_image_questions(tmp_path: Path) -> None:
+    cases = {
+        'diag_cache_expired': ('\u56fe\u7247\u7f13\u5b58\u5df2\u5931\u6548\uff0c\u8bf7\u91cd\u65b0\u53d1\u9001\u56fe\u7247\u3002', False),
+        'diag_cache_error_en': ('image cache error', False),
+        'diag_cache_expired_en': ('the image cache expired, please resend', False),
+        'diag_copying_text_shows_cache_error': ('\u4e3a\u4ec0\u4e48\u6211\u590d\u5236\u53d1\u7ed9\u4f60\u7684\u6587\u5b57\uff0c\u4f60\u90fd\u663e\u793a\u56fe\u7247\u7f13\u5b58\u5df2\u5931\u6548\uff0c\u8bf7\u91cd\u65b0\u53d1\u9001\u56fe\u7247\u3002', False),
+        'diag_not_image': ('\u8fd9\u4e0d\u662f\u56fe\u7247\uff0c\u662f\u6587\u5b57\u6d88\u606f', False),
+        'diag_text_not_image_en': ('this is a text message, not an image', False),
+        'diag_no_image_sent': ('\u6211\u6ca1\u6709\u53d1\u56fe\u7247', False),
+        'diag_system_logs': ('\u8bf7\u68c0\u67e5\u7cfb\u7edf\u65e5\u5fd7\uff0c\u4ece\u673a\u5236\u4e0a\u4fee\u590d\u9519\u8bef', False),
+        'image_cn_1': ('\u8bf7\u770b\u4e00\u4e0b\u8fd9\u5f20\u56fe\u7247', True),
+        'image_cn_2': ('\u8fd9\u56fe\u662f\u4ec0\u4e48', True),
+        'image_cn_3': ('\u8bc6\u522b\u56fe\u7247\u5185\u5bb9', True),
+        'image_en': ('look at this image', True),
+    }
+    results = run_js_process_helpers(
+        tmp_path,
+        'const cases = '
+        + json.dumps({key: text for key, (text, _) in cases.items()}, ensure_ascii=False)
+        + ';\nconst results = {};\n'
+        + 'for (const [key, text] of Object.entries(cases)) {\n'
+        + '  results[key] = looksLikeHermesImageFollowup(text);\n'
+        + '}\n'
+        + 'console.log(JSON.stringify(results));\n',
+    )
+    for key, (_, expected) in cases.items():
+        assert results[key] is expected, f'{key}: expected {expected}, got {results[key]!r}'
+
+
+
+def test_resolve_hermes_referenced_image_followup_requires_exact_cache(tmp_path: Path) -> None:
+    cache_path = tmp_path / 'hermes-last-image-cache.json'
+    results = run_js_process_helpers(
+        tmp_path,
+        '\n'.join(
+            [
+                'const fs = require("fs");',
+                f'const cachePath = {json.dumps(str(cache_path))};',
+                'const accountId = "account-a";',
+                'const senderId = "sender-b";',
+                'const exactQuotedId = "quoted-123";',
+                'const cache = {};',
+                'cache[hermesLastImageCacheKey(accountId, senderId)] = {',
+                '  mediaUrl: "data:image/png;base64,iVBORw0KGgo=",',
+                '  timestampMs: Date.now(),',
+                '  messageId: "latest-999",',
+                '  mediaType: "image/png",',
+                '};',
+                'cache[hermesExactImageCacheKey(accountId, senderId, exactQuotedId)] = {',
+                '  mediaUrl: "data:image/png;base64,iVBORw0KGgo=",',
+                '  timestampMs: Date.now(),',
+                '  messageId: exactQuotedId,',
+                '  mediaType: "image/png",',
+                '};',
+                'fs.writeFileSync(cachePath, JSON.stringify(cache, null, 2), "utf-8");',
+                'process.env.AIOS_HERMES_WEIXIN_LAST_IMAGE_CACHE = cachePath;',
+                'const generic = resolveHermesReferencedImageFollowup(accountId, senderId, "plain quoted text", "quoted-999");',
+                'const exact = resolveHermesReferencedImageFollowup(accountId, senderId, "plain quoted text", exactQuotedId);',
+                'const forced = resolveHermesFollowupImageContext(accountId, senderId, "plain quoted text", exactQuotedId, false, true);',
+                'console.log(JSON.stringify({ generic, exact, forced }));',
+            ]
+        ),
+    )
+
+    assert results['generic']['imageIntent'] is False
+    assert results['generic']['matched'] is False
+    assert results['generic']['reason'] == 'referenced_image_not_confirmed'
+    assert results['exact']['imageIntent'] is True
+    assert results['exact']['matched'] is True
+    assert results['exact']['reason'] == 'cached_image_matched'
+    assert results['forced']['imageIntent'] is True
+    assert results['forced']['matched'] is True
+
+
+
+def test_referenced_text_without_media_evidence_stays_out_of_image_cache(tmp_path: Path) -> None:
+    cache_path = tmp_path / 'hermes-last-image-cache.json'
+    results = run_js_process_helpers(
+        tmp_path,
+        '\n'.join(
+            [
+                'const fs = require("fs");',
+                f'const cachePath = {json.dumps(str(cache_path))};',
+                'const accountId = "account-a";',
+                'const senderId = "sender-b";',
+                'const cache = {};',
+                'cache[hermesLastImageCacheKey(accountId, senderId)] = {',
+                '  mediaUrl: "data:image/png;base64,iVBORw0KGgo=",',
+                '  timestampMs: Date.now(),',
+                '  messageId: "latest-999",',
+                '  mediaType: "image/png",',
+                '};',
+                'fs.writeFileSync(cachePath, JSON.stringify(cache, null, 2), "utf-8");',
+                'process.env.AIOS_HERMES_WEIXIN_LAST_IMAGE_CACHE = cachePath;',
+                'const full = { item_list: [{ ref_msg: { message_item: { type: 0, id: "quoted-999" } } }] };',
+                'const hermesContent = "\u8fd9\u662f\u4ec0\u4e48";',
+                'const hermesReferenceContextPresent = hasHermesReferenceContext(full.item_list);',
+                'const hermesQuotedMessageId = findHermesReferencedMessageId(full.item_list);',
+                'const refMediaItem = undefined;',
+                'const hermesReferencedMediaIsImage = Boolean(refMediaItem && hermesCurrentMessageIsImage({}, refMediaItem));',
+                'const hermesReferencedImageFollowupIntent = hermesReferenceContextPresent && !refMediaItem;',
+                'const hermesCurrentImage = false;',
+                'const hermesFollowup = hermesCurrentImage',
+                '  ? { imageIntent: false, matched: false, reason: "current_image_message", cacheStrategy: "current", quotedMessageId: hermesQuotedMessageId }',
+                '  : (hermesReferencedMediaIsImage',
+                '      ? resolveHermesFollowupImageContext(accountId, senderId, hermesContent, hermesQuotedMessageId, false, true)',
+                '      : (hermesReferencedImageFollowupIntent',
+                '          ? resolveHermesReferencedImageFollowup(accountId, senderId, hermesContent, hermesQuotedMessageId)',
+                '          : resolveHermesFollowupImageContext(accountId, senderId, hermesContent)));',
+                'console.log(JSON.stringify({ hermesReferenceContextPresent, hermesQuotedMessageId, hermesReferencedMediaIsImage, hermesReferencedImageFollowupIntent, hermesFollowup }));',
+            ]
+        ),
+    )
+
+    assert results['hermesReferenceContextPresent'] is True
+    assert results['hermesQuotedMessageId'] == 'quoted-999'
+    assert results['hermesReferencedMediaIsImage'] is False
+    assert results['hermesReferencedImageFollowupIntent'] is True
+    assert results['hermesFollowup']['imageIntent'] is False
+    assert results['hermesFollowup']['matched'] is False
+    assert results['hermesFollowup']['reason'] == 'referenced_image_not_confirmed'
+
+
+def test_patch_plugin_process_only_skips_slash_and_channel(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    patcher = load_patcher_module()
+    calls: list[tuple[str, Path]] = []
+
+    def record(name: str, result):
+        def _inner(root: Path):
+            calls.append((name, root))
+            return result
+
+        return _inner
+
+    monkeypatch.setattr(patcher, 'patch_ts', record('patch_ts', False))
+    monkeypatch.setattr(patcher, 'patch_js', record('patch_js', False))
+    monkeypatch.setattr(patcher, 'patch_process_messages', record('patch_process_messages', {'src/messaging/process-message.ts': True, 'dist/src/messaging/process-message.js': False}))
+    monkeypatch.setattr(patcher, 'patch_channel_files', record('patch_channel_files', {'src/channel.ts': False, 'dist/src/channel.js': False}))
+    monkeypatch.setattr(patcher, 'verify_plugin_hook', record('verify_plugin_hook', {
+        'hook_status': {},
+        'hook_installed': False,
+        'process_status': {'src/messaging/process-message.ts': True, 'dist/src/messaging/process-message.js': True},
+        'process_hook_installed': True,
+        'channel_status': {},
+        'channel_hook_installed': False,
+        'dispatch_reference_found': False,
+        'dispatch_reference_files': [],
+    }))
+
+    result = patcher.patch_plugin(tmp_path, process_only=True)
+
+    assert calls == [
+        ('patch_process_messages', tmp_path),
+        ('verify_plugin_hook', tmp_path),
+    ]
+    assert result['mode'] == 'process-only'
+    assert result['plugin_root'] == str(tmp_path)
+    assert result['slash_changes'] == {}
+    assert result['channel_changes'] == {}
+    assert result['process_changes'] == {'src/messaging/process-message.ts': True, 'dist/src/messaging/process-message.js': False}
+    assert result['behavior_changes'] == {'src/messaging/process-message.ts': True, 'dist/src/messaging/process-message.js': False}
+    assert result['verification']['process_hook_installed'] is True
+
+
+
+def test_parse_args_supports_default_root_and_process_only_flag() -> None:
+    patcher = load_patcher_module()
+
+    defaults = patcher.parse_args([])
+    assert defaults.plugin_root == patcher.DEFAULT_PLUGIN_ROOT
+    assert defaults.process_only is False
+
+    process_only = patcher.parse_args(['--process-only'])
+    assert process_only.plugin_root == patcher.DEFAULT_PLUGIN_ROOT
+    assert process_only.process_only is True
+
+    custom_root = Path('/tmp/custom-plugin-root')
+    custom = patcher.parse_args([str(custom_root), '--process-only'])
+    assert custom.plugin_root == custom_root
+    assert custom.process_only is True