refactor(audit): add retained-chain verifier

scripts/verify_audit_chain.py

@@ -0,0 +1,56 @@
+"""Verify the retained OpenClaw audit hash chain across active and rotated files."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+
+def _repo_root() -> Path:
+    return Path(__file__).resolve().parents[1]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="Verify the retained OpenClaw audit hash chain."
+    )
+    parser.add_argument(
+        "--path",
+        default="",
+        help="Explicit audit log path. Defaults to the configured OpenClaw audit path.",
+    )
+    parser.add_argument(
+        "--json",
+        action="store_true",
+        help="Emit machine-readable JSON instead of a human summary.",
+    )
+    args = parser.parse_args()
+
+    import sys
+
+    sys.path.insert(0, str(_repo_root()))
+    from services.audit import verify_audit_chain
+
+    result = verify_audit_chain(args.path or None)
+    if args.json:
+        print(json.dumps(result.to_dict(), indent=2))
+    else:
+        status = "PASS" if result.ok else "FAIL"
+        print(f"Audit Chain Verification: {status}")
+        print(f"Files checked: {len(result.files_checked)}")
+        print(f"Entries checked: {result.entries_checked}")
+        print(f"Window start prev_hash: {result.window_start_prev_hash}")
+        print(f"Terminal hash: {result.terminal_hash}")
+        print(f"Window truncated: {result.window_truncated}")
+        if result.issues:
+            print("Issues:")
+            for issue in result.issues:
+                print(
+                    f"- {issue.code} at {issue.file_path}:{issue.line_number} -> {issue.message}"
+                )
+    return 0 if result.ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/audit.py

@@ -13,6 +13,12 @@ import time
 import uuid
 from typing import Any, Dict, Optional, Tuple
 
+from .audit_pipeline import (
+    AuditVerificationResult,
+    LocalFileAuditSink,
+    read_last_entry_hash_from_chain,
+)
+from .audit_pipeline import verify_audit_chain as verify_audit_chain_impl
 from .redaction import redact_json, stable_redaction_tag
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.audit")
@@ -44,6 +50,15 @@ AUDIT_LOG_PATH = (
 )
 
 
+def _default_audit_chain_key_path() -> str:
+    override = os.environ.get("OPENCLAW_AUDIT_CHAIN_KEY_PATH") or os.environ.get(
+        "MOLTBOT_AUDIT_CHAIN_KEY_PATH"
+    )
+    if override:
+        return override
+    return f"{AUDIT_LOG_PATH}.key"
+
+
 def _env_int(primary: str, legacy: str, default: int) -> int:
     raw = os.environ.get(primary) or os.environ.get(legacy)
     if raw is None:
@@ -109,7 +124,41 @@ def _get_audit_chain_key() -> bytes:
         raw = os.environ.get("OPENCLAW_AUDIT_CHAIN_KEY") or os.environ.get(
             "MOLTBOT_AUDIT_CHAIN_KEY"
         )
-        _AUDIT_CHAIN_KEY = raw.encode("utf-8") if raw else secrets.token_bytes(32)
+        if raw:
+            _AUDIT_CHAIN_KEY = raw.encode("utf-8")
+            return _AUDIT_CHAIN_KEY
+
+        key_path = _default_audit_chain_key_path()
+        try:
+            with open(key_path, "r", encoding="utf-8") as handle:
+                persisted = handle.read().strip()
+            if persisted:
+                _AUDIT_CHAIN_KEY = bytes.fromhex(persisted)
+                return _AUDIT_CHAIN_KEY
+        except Exception:
+            pass
+
+        # CRITICAL: the audit chain key must stay stable across restarts or
+        # rotated-chain verification becomes impossible after the first reboot.
+        generated = secrets.token_bytes(32)
+        try:
+            os.makedirs(os.path.dirname(os.path.abspath(key_path)), exist_ok=True)
+        except Exception:
+            pass
+        try:
+            with open(key_path, "x", encoding="utf-8") as handle:
+                handle.write(generated.hex())
+            try:
+                os.chmod(key_path, 0o600)
+            except Exception:
+                pass
+            _AUDIT_CHAIN_KEY = generated
+        except FileExistsError:
+            with open(key_path, "r", encoding="utf-8") as handle:
+                persisted = handle.read().strip()
+            _AUDIT_CHAIN_KEY = bytes.fromhex(persisted) if persisted else generated
+        except Exception:
+            _AUDIT_CHAIN_KEY = generated
     return _AUDIT_CHAIN_KEY
 
 
@@ -126,54 +175,25 @@ def _chain_hash(prev_hash: str, entry: Dict[str, Any]) -> str:
     ).hex()
 
 
-def _rotate_if_needed(path: str) -> None:
+def _build_audit_sink(path: str) -> LocalFileAuditSink:
     max_bytes, backups = _audit_limits()
-    if max_bytes <= 0 or backups < 0:
-        return
-    if not os.path.exists(path):
-        return
+    return LocalFileAuditSink(
+        path=path,
+        max_bytes=max_bytes,
+        backups=backups,
+        chain_hash=_chain_hash,
+    )
+
+
+def _rotate_if_needed(path: str) -> None:
     try:
-        if os.path.getsize(path) < max_bytes:
-            return
-        if backups == 0:
-            os.remove(path)
-            return
-        for idx in range(backups, 0, -1):
-            src = f"{path}.{idx}"
-            dst = f"{path}.{idx + 1}"
-            if os.path.exists(src):
-                if idx == backups:
-                    os.remove(src)
-                else:
-                    os.replace(src, dst)
-        os.replace(path, f"{path}.1")
+        _build_audit_sink(path).rotate_if_needed()
     except Exception as exc:
         logger.error("Audit rotation failed: %s", exc)
 
 
 def _read_last_entry_hash(path: str) -> str:
-    # Best-effort bootstrap. Fall back to genesis if file is empty/unreadable.
-    if not os.path.exists(path):
-        return "GENESIS"
-    try:
-        with open(path, "rb") as f:
-            f.seek(0, os.SEEK_END)
-            size = f.tell()
-            if size <= 0:
-                return "GENESIS"
-            step = min(size, 8192)
-            f.seek(size - step)
-            data = f.read().decode("utf-8", errors="ignore")
-        lines = [line.strip() for line in data.splitlines() if line.strip()]
-        if not lines:
-            return "GENESIS"
-        last = json.loads(lines[-1])
-        last_hash = last.get("entry_hash")
-        if isinstance(last_hash, str) and last_hash:
-            return last_hash
-    except Exception:
-        pass
-    return "GENESIS"
+    return read_last_entry_hash_from_chain(path)
 
 
 _LAST_HASH: Optional[str] = None
@@ -182,29 +202,12 @@ _AUDIT_WRITE_LOCK = threading.Lock()
 
 def _write_audit_entry(entry: Dict[str, Any]) -> None:
     global _LAST_HASH
-    try:
-        os.makedirs(os.path.dirname(os.path.abspath(AUDIT_LOG_PATH)), exist_ok=True)
-    except Exception:
-        # Path may be relative to CWD with no parent folder.
-        pass
-
-    # CRITICAL: keep read->chain->append->state update atomic to avoid hash-chain forks.
+    sink = _build_audit_sink(AUDIT_LOG_PATH)
+    # CRITICAL: keep tail-resolution -> rotate -> append -> cache update atomic so
+    # restarts and file rotations cannot fork the retained audit chain.
     with _AUDIT_WRITE_LOCK:
-        _rotate_if_needed(AUDIT_LOG_PATH)
-        if _LAST_HASH is None:
-            _LAST_HASH = _read_last_entry_hash(AUDIT_LOG_PATH)
-
-        prev_hash = _LAST_HASH or "GENESIS"
-        event_hash = _chain_hash(prev_hash, entry)
-        wrapped = dict(entry)
-        wrapped["prev_hash"] = prev_hash
-        wrapped["entry_hash"] = event_hash
-
-        line = json.dumps(wrapped, sort_keys=True, ensure_ascii=True) + "\n"
         try:
-            with open(AUDIT_LOG_PATH, "a", encoding="utf-8") as f:
-                f.write(line)
-            _LAST_HASH = event_hash
+            _LAST_HASH = sink.append_entry(entry, last_hash=_LAST_HASH)
         except Exception as exc:
             logger.error("Failed to write audit entry: %s", exc)
 
@@ -395,3 +398,7 @@ def audit_llm_test(actor_ip: str, ok: bool, error: Optional[str] = None) -> None
             else {"actor_ip_tag": stable_redaction_tag(actor_ip, label="ip")}
         ),
     )
+
+
+def verify_audit_chain(path: Optional[str] = None) -> AuditVerificationResult:
+    return verify_audit_chain_impl(path or AUDIT_LOG_PATH, chain_hash=_chain_hash)

services/audit_pipeline.py

@@ -0,0 +1,306 @@
+"""Audit persistence sinks and chain verification helpers."""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Dict, List, Optional, Protocol
+
+
+class AuditSink(Protocol):
+    def append_entry(self, entry: Dict[str, Any], *, last_hash: Optional[str]) -> str: ...
+
+
+@dataclass
+class AuditVerificationIssue:
+    code: str
+    file_path: str
+    line_number: int
+    message: str
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "code": self.code,
+            "file_path": self.file_path,
+            "line_number": self.line_number,
+            "message": self.message,
+        }
+
+
+@dataclass
+class AuditVerificationResult:
+    ok: bool
+    files_checked: List[str]
+    entries_checked: int
+    window_start_prev_hash: str
+    terminal_hash: str
+    window_truncated: bool
+    issues: List[AuditVerificationIssue] = field(default_factory=list)
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "ok": self.ok,
+            "files_checked": list(self.files_checked),
+            "entries_checked": self.entries_checked,
+            "window_start_prev_hash": self.window_start_prev_hash,
+            "terminal_hash": self.terminal_hash,
+            "window_truncated": self.window_truncated,
+            "issues": [issue.to_dict() for issue in self.issues],
+        }
+
+
+def _rotated_chain_paths(base_path: str) -> List[str]:
+    base = Path(base_path)
+    parent = base.parent if str(base.parent) else Path(".")
+    if not parent.exists():
+        return []
+    pattern = re.compile(rf"^{re.escape(base.name)}\.(\d+)$")
+    matches = []
+    for child in parent.iterdir():
+        if not child.is_file():
+            continue
+        match = pattern.match(child.name)
+        if not match:
+            continue
+        matches.append((int(match.group(1)), str(child)))
+    matches.sort(reverse=True)
+    return [path for _, path in matches]
+
+
+def iter_audit_chain_paths(base_path: str) -> List[str]:
+    paths = _rotated_chain_paths(base_path)
+    if os.path.exists(base_path):
+        paths.append(base_path)
+    return paths
+
+
+def _tail_hash_from_file(path: str) -> Optional[str]:
+    if not os.path.exists(path):
+        return None
+    try:
+        with open(path, "rb") as handle:
+            handle.seek(0, os.SEEK_END)
+            size = handle.tell()
+            if size <= 0:
+                return None
+            step = min(size, 8192)
+            handle.seek(size - step)
+            data = handle.read().decode("utf-8", errors="ignore")
+    except Exception:
+        return None
+
+    lines = [line.strip() for line in data.splitlines() if line.strip()]
+    if not lines:
+        return None
+    try:
+        payload = json.loads(lines[-1])
+    except Exception:
+        return None
+    value = payload.get("entry_hash")
+    return value if isinstance(value, str) and value else None
+
+
+def read_last_entry_hash_from_chain(base_path: str) -> str:
+    newest_first = [base_path] if os.path.exists(base_path) else []
+    newest_first.extend(reversed(_rotated_chain_paths(base_path)))
+    for path in newest_first:
+        value = _tail_hash_from_file(path)
+        if value:
+            return value
+    return "GENESIS"
+
+
+class LocalFileAuditSink:
+    def __init__(
+        self,
+        *,
+        path: str,
+        max_bytes: int,
+        backups: int,
+        chain_hash: Callable[[str, Dict[str, Any]], str],
+    ) -> None:
+        self.path = path
+        self.max_bytes = max_bytes
+        self.backups = backups
+        self._chain_hash = chain_hash
+
+    def _ensure_parent_dir(self) -> None:
+        try:
+            os.makedirs(os.path.dirname(os.path.abspath(self.path)), exist_ok=True)
+        except Exception:
+            pass
+
+    def rotate_if_needed(self) -> None:
+        if self.max_bytes <= 0 or self.backups < 0:
+            return
+        if not os.path.exists(self.path):
+            return
+        if os.path.getsize(self.path) < self.max_bytes:
+            return
+        if self.backups == 0:
+            os.remove(self.path)
+            return
+        for idx in range(self.backups, 0, -1):
+            src = f"{self.path}.{idx}"
+            dst = f"{self.path}.{idx + 1}"
+            if os.path.exists(src):
+                if idx == self.backups:
+                    os.remove(src)
+                else:
+                    os.replace(src, dst)
+        os.replace(self.path, f"{self.path}.1")
+
+    def append_entry(self, entry: Dict[str, Any], *, last_hash: Optional[str]) -> str:
+        self._ensure_parent_dir()
+        self.rotate_if_needed()
+        prev_hash = last_hash or read_last_entry_hash_from_chain(self.path)
+        event_hash = self._chain_hash(prev_hash, entry)
+        wrapped = dict(entry)
+        wrapped["prev_hash"] = prev_hash
+        wrapped["entry_hash"] = event_hash
+        line = json.dumps(wrapped, sort_keys=True, ensure_ascii=True) + "\n"
+        with open(self.path, "a", encoding="utf-8") as handle:
+            handle.write(line)
+        return event_hash
+
+
+def verify_audit_chain(
+    base_path: str,
+    *,
+    chain_hash: Callable[[str, Dict[str, Any]], str],
+) -> AuditVerificationResult:
+    files = iter_audit_chain_paths(base_path)
+    issues: List[AuditVerificationIssue] = []
+    if not files:
+        issues.append(
+            AuditVerificationIssue(
+                code="missing_chain",
+                file_path=base_path,
+                line_number=0,
+                message="No audit log files found for verification.",
+            )
+        )
+        return AuditVerificationResult(
+            ok=False,
+            files_checked=[],
+            entries_checked=0,
+            window_start_prev_hash="GENESIS",
+            terminal_hash="GENESIS",
+            window_truncated=False,
+            issues=issues,
+        )
+
+    previous_entry_hash: Optional[str] = None
+    window_start_prev_hash = "GENESIS"
+    entries_checked = 0
+
+    for file_path in files:
+        with open(file_path, "r", encoding="utf-8") as handle:
+            for line_number, raw_line in enumerate(handle, start=1):
+                line = raw_line.strip()
+                if not line:
+                    continue
+                try:
+                    wrapped = json.loads(line)
+                except Exception as exc:
+                    issues.append(
+                        AuditVerificationIssue(
+                            code="invalid_json",
+                            file_path=file_path,
+                            line_number=line_number,
+                            message=f"Invalid JSON entry: {exc}",
+                        )
+                    )
+                    return AuditVerificationResult(
+                        ok=False,
+                        files_checked=files,
+                        entries_checked=entries_checked,
+                        window_start_prev_hash=window_start_prev_hash,
+                        terminal_hash=previous_entry_hash or "GENESIS",
+                        window_truncated=window_start_prev_hash != "GENESIS",
+                        issues=issues,
+                    )
+
+                prev_hash = wrapped.get("prev_hash")
+                entry_hash = wrapped.get("entry_hash")
+                if not isinstance(prev_hash, str) or not isinstance(entry_hash, str):
+                    issues.append(
+                        AuditVerificationIssue(
+                            code="missing_hash_fields",
+                            file_path=file_path,
+                            line_number=line_number,
+                            message="Audit entry is missing string prev_hash/entry_hash fields.",
+                        )
+                    )
+                    return AuditVerificationResult(
+                        ok=False,
+                        files_checked=files,
+                        entries_checked=entries_checked,
+                        window_start_prev_hash=window_start_prev_hash,
+                        terminal_hash=previous_entry_hash or "GENESIS",
+                        window_truncated=window_start_prev_hash != "GENESIS",
+                        issues=issues,
+                    )
+
+                if previous_entry_hash is None:
+                    window_start_prev_hash = prev_hash
+                elif prev_hash != previous_entry_hash:
+                    issues.append(
+                        AuditVerificationIssue(
+                            code="prev_hash_mismatch",
+                            file_path=file_path,
+                            line_number=line_number,
+                            message=(
+                                "Audit chain continuity failed: prev_hash does not match "
+                                "the preceding entry_hash."
+                            ),
+                        )
+                    )
+                    return AuditVerificationResult(
+                        ok=False,
+                        files_checked=files,
+                        entries_checked=entries_checked,
+                        window_start_prev_hash=window_start_prev_hash,
+                        terminal_hash=previous_entry_hash or "GENESIS",
+                        window_truncated=window_start_prev_hash != "GENESIS",
+                        issues=issues,
+                    )
+
+                payload = dict(wrapped)
+                payload.pop("prev_hash", None)
+                payload.pop("entry_hash", None)
+                expected_hash = chain_hash(prev_hash, payload)
+                if entry_hash != expected_hash:
+                    issues.append(
+                        AuditVerificationIssue(
+                            code="entry_hash_mismatch",
+                            file_path=file_path,
+                            line_number=line_number,
+                            message="Audit entry_hash does not match the persisted payload.",
+                        )
+                    )
+                    return AuditVerificationResult(
+                        ok=False,
+                        files_checked=files,
+                        entries_checked=entries_checked,
+                        window_start_prev_hash=window_start_prev_hash,
+                        terminal_hash=previous_entry_hash or "GENESIS",
+                        window_truncated=window_start_prev_hash != "GENESIS",
+                        issues=issues,
+                    )
+
+                previous_entry_hash = entry_hash
+                entries_checked += 1
+
+    return AuditVerificationResult(
+        ok=True,
+        files_checked=files,
+        entries_checked=entries_checked,
+        window_start_prev_hash=window_start_prev_hash,
+        terminal_hash=previous_entry_hash or "GENESIS",
+        window_truncated=window_start_prev_hash != "GENESIS",
+        issues=[],
+    )

services/security_doctor_connector_checks.py

@@ -4,7 +4,11 @@ from __future__ import annotations
 
 import os
 
-from .security_doctor_report import SecurityCheckResult, SecurityReport, SecuritySeverity
+from .security_doctor_report import (
+    SecurityCheckResult,
+    SecurityReport,
+    SecuritySeverity,
+)
 
 try:
     from .connector_allowlist_posture import (

services/security_doctor_endpoint_checks.py

@@ -7,7 +7,11 @@ import json
 import os
 from typing import Dict
 
-from .security_doctor_report import SecurityCheckResult, SecurityReport, SecuritySeverity
+from .security_doctor_report import (
+    SecurityCheckResult,
+    SecurityReport,
+    SecuritySeverity,
+)
 
 try:
     from ..config import PACK_VERSION

services/security_doctor_runtime_checks.py

@@ -8,7 +8,11 @@ import stat
 import sys
 from pathlib import Path
 
-from .security_doctor_report import SecurityCheckResult, SecurityReport, SecuritySeverity
+from .security_doctor_report import (
+    SecurityCheckResult,
+    SecurityReport,
+    SecuritySeverity,
+)
 
 
 def check_state_dir_permissions(report: SecurityReport) -> None:

tests/security/test_audit.py

@@ -17,23 +17,31 @@ from services.audit import (
 class TestAudit(unittest.TestCase):
     def setUp(self):
         self.test_log = "test_audit.log"
+        self.test_key = f"{self.test_log}.key"
         self.path_patcher = patch("services.audit.AUDIT_LOG_PATH", self.test_log)
         self.path_patcher.start()
         self.hash_patcher = patch("services.audit._LAST_HASH", None)
         self.hash_patcher.start()
+        self.chain_key_patcher = patch("services.audit._AUDIT_CHAIN_KEY", None)
+        self.chain_key_patcher.start()
         self.tag_key_patcher = patch.object(
             redaction_module, "_REDACTION_TAG_KEY", b"audit-test-redaction-key"
         )
         self.tag_key_patcher.start()
         if os.path.exists(self.test_log):
             os.remove(self.test_log)
+        if os.path.exists(self.test_key):
+            os.remove(self.test_key)
 
     def tearDown(self):
         self.path_patcher.stop()
         self.hash_patcher.stop()
+        self.chain_key_patcher.stop()
         self.tag_key_patcher.stop()
         if os.path.exists(self.test_log):
             os.remove(self.test_log)
+        if os.path.exists(self.test_key):
+            os.remove(self.test_key)
 
     def _read_entries(self):
         with open(self.test_log, "r", encoding="utf-8") as f:

tests/security/test_s78_redaction.py

@@ -153,23 +153,31 @@ class TestS78BridgeWorkerRedaction(unittest.TestCase):
 class TestS78AuditRedaction(unittest.TestCase):
     def setUp(self):
         self.test_log = "test_s78_audit.log"
+        self.test_key = f"{self.test_log}.key"
         self.path_patcher = patch("services.audit.AUDIT_LOG_PATH", self.test_log)
         self.hash_patcher = patch("services.audit._LAST_HASH", None)
+        self.chain_key_patcher = patch("services.audit._AUDIT_CHAIN_KEY", None)
         self.tag_key_patcher = patch.object(
             redaction_module, "_REDACTION_TAG_KEY", b"s78-test-redaction-key"
         )
         self.path_patcher.start()
         self.hash_patcher.start()
+        self.chain_key_patcher.start()
         self.tag_key_patcher.start()
         if os.path.exists(self.test_log):
             os.remove(self.test_log)
+        if os.path.exists(self.test_key):
+            os.remove(self.test_key)
 
     def tearDown(self):
         self.path_patcher.stop()
         self.hash_patcher.stop()
+        self.chain_key_patcher.stop()
         self.tag_key_patcher.stop()
         if os.path.exists(self.test_log):
             os.remove(self.test_log)
+        if os.path.exists(self.test_key):
+            os.remove(self.test_key)
 
     def _read_entries(self):
         with open(self.test_log, "r", encoding="utf-8") as handle:

tests/test_r159_audit_pipeline.py

@@ -0,0 +1,165 @@
+import importlib.util
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import services.audit as audit_module
+
+
+def _load_verify_script_module():
+    root = Path(__file__).resolve().parents[1]
+    module_path = root / "scripts" / "verify_audit_chain.py"
+    spec = importlib.util.spec_from_file_location("verify_audit_chain_script", module_path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError("Failed to load verify_audit_chain.py")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class TestR159AuditPipeline(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp_dir.cleanup)
+        self.audit_log = os.path.join(self.temp_dir.name, "audit.log")
+        self.audit_key = f"{self.audit_log}.key"
+        self.env_patcher = patch.dict(
+            os.environ,
+            {
+                "OPENCLAW_AUDIT_CHAIN_KEY": "",
+                "MOLTBOT_AUDIT_CHAIN_KEY": "",
+                "OPENCLAW_AUDIT_CHAIN_KEY_PATH": "",
+                "MOLTBOT_AUDIT_CHAIN_KEY_PATH": "",
+            },
+            clear=False,
+        )
+        self.env_patcher.start()
+        self.addCleanup(self.env_patcher.stop)
+        self.path_patcher = patch("services.audit.AUDIT_LOG_PATH", self.audit_log)
+        self.hash_patcher = patch("services.audit._LAST_HASH", None)
+        self.chain_key_patcher = patch("services.audit._AUDIT_CHAIN_KEY", None)
+        self.path_patcher.start()
+        self.hash_patcher.start()
+        self.chain_key_patcher.start()
+        self.addCleanup(self.path_patcher.stop)
+        self.addCleanup(self.hash_patcher.stop)
+        self.addCleanup(self.chain_key_patcher.stop)
+
+    def _read_chain_entries(self):
+        payloads = []
+        for path in audit_module.verify_audit_chain().files_checked:
+            with open(path, "r", encoding="utf-8") as handle:
+                for line in handle:
+                    if line.strip():
+                        payloads.append(json.loads(line))
+        return payloads
+
+    def test_restart_after_rotation_preserves_chain_continuity(self):
+        with patch("services.audit._audit_limits", return_value=(1, 3)):
+            audit_module.emit_audit_event(
+                action="config.update",
+                target="settings.json",
+                outcome="allow",
+                status_code=200,
+                details={"seq": 1},
+            )
+            audit_module.emit_audit_event(
+                action="config.update",
+                target="settings.json",
+                outcome="allow",
+                status_code=200,
+                details={"seq": 2},
+            )
+            audit_module._LAST_HASH = None
+            audit_module._AUDIT_CHAIN_KEY = None
+            audit_module.emit_audit_event(
+                action="config.update",
+                target="settings.json",
+                outcome="allow",
+                status_code=200,
+                details={"seq": 3},
+            )
+
+        result = audit_module.verify_audit_chain()
+        self.assertTrue(result.ok, result.to_dict())
+        self.assertEqual(result.entries_checked, 3)
+        self.assertEqual(len(result.files_checked), 3)
+        self.assertEqual(result.window_start_prev_hash, "GENESIS")
+        entries = self._read_chain_entries()
+        self.assertEqual(entries[1]["prev_hash"], entries[0]["entry_hash"])
+        self.assertEqual(entries[2]["prev_hash"], entries[1]["entry_hash"])
+        self.assertTrue(os.path.exists(self.audit_key))
+
+    def test_verify_detects_tampered_entry_hash(self):
+        audit_module.emit_audit_event(
+            action="config.update",
+            target="settings.json",
+            outcome="allow",
+            status_code=200,
+            details={"seq": 1},
+        )
+        with open(self.audit_log, "r", encoding="utf-8") as handle:
+            entry = json.loads(handle.readline())
+        entry["action"] = "config.tampered"
+        with open(self.audit_log, "w", encoding="utf-8") as handle:
+            handle.write(json.dumps(entry, sort_keys=True, ensure_ascii=True) + "\n")
+
+        result = audit_module.verify_audit_chain()
+        self.assertFalse(result.ok)
+        self.assertEqual(result.issues[0].code, "entry_hash_mismatch")
+
+    def test_emit_surfaces_sink_failure_but_does_not_raise(self):
+        with patch(
+            "services.audit.LocalFileAuditSink.append_entry",
+            side_effect=OSError("disk full"),
+        ):
+            with self.assertLogs("ComfyUI-OpenClaw.services.audit", level="ERROR") as logs:
+                audit_module.emit_audit_event(
+                    action="config.update",
+                    target="settings.json",
+                    outcome="allow",
+                    status_code=200,
+                    details={"seq": 1},
+                )
+
+        output = "\n".join(logs.output)
+        self.assertIn("Failed to write audit entry", output)
+        self.assertFalse(os.path.exists(self.audit_log))
+
+    def test_verify_script_reports_pass_and_fail(self):
+        script = _load_verify_script_module()
+        audit_module.emit_audit_event(
+            action="config.update",
+            target="settings.json",
+            outcome="allow",
+            status_code=200,
+            details={"seq": 1},
+        )
+
+        with patch.object(
+            sys,
+            "argv",
+            ["verify_audit_chain.py", "--path", self.audit_log, "--json"],
+        ):
+            self.assertEqual(script.main(), 0)
+
+        with open(self.audit_log, "r", encoding="utf-8") as handle:
+            entry = json.loads(handle.readline())
+        entry["entry_hash"] = "deadbeef"
+        with open(self.audit_log, "w", encoding="utf-8") as handle:
+            handle.write(json.dumps(entry, sort_keys=True, ensure_ascii=True) + "\n")
+
+        with patch.object(
+            sys,
+            "argv",
+            ["verify_audit_chain.py", "--path", self.audit_log],
+        ):
+            self.assertEqual(script.main(), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()