feat(security): implement S35/R77/S34 wave2 with process-isolated transforms, tamper-evident persistence, and observability tier enforcement

.github/workflows/ci.yml

@@ -69,7 +69,9 @@ jobs:
       - name: Install test deps
         run: |
           python -m pip install --upgrade pip
-          python -m pip install numpy pillow
+          # Keep aligned with local pre-push/full-test scripts.
+          # aiohttp is required by multiple unit-test import paths.
+          python -m pip install numpy pillow aiohttp
       - name: Run unit tests
 
         env:

api/routes.py

@@ -67,7 +67,10 @@ if web is not None:
         # CRITICAL: These imports MUST remain present.
         # If edited out, module-level placeholders stay as None and handlers raise at runtime
         # (e.g., TypeError: 'NoneType' object is not callable), producing noisy aiohttp tracebacks.
-        from ..services.access_control import require_observability_access
+        from ..services.access_control import (
+            require_admin_token,
+            require_observability_access,
+        )
         from ..services.log_tail import tail_log
         from ..services.metrics import metrics
         from ..services.rate_limit import check_rate_limit
@@ -109,6 +112,7 @@ if web is not None:
 
         # IMPORTANT: keep PACK_* imports aligned with config.py (VERSION/config_path do not exist).
         from config import LOG_FILE, PACK_NAME, PACK_START_TIME, PACK_VERSION
+        from services.access_control import require_admin_token  # type: ignore
         from services.access_control import require_observability_access  # type: ignore
         from services.log_tail import tail_log  # type: ignore
         from services.metrics import metrics  # type: ignore
@@ -261,8 +265,8 @@ async def logs_tail_handler(request: web.Request) -> web.Response:
     ok, init_error = _ensure_observability_deps_ready()
     if not ok:
         return web.json_response({"ok": False, "error": init_error}, status=500)
-    # S14: Access Control
-    allowed, error = require_observability_access(request)
+    # S34: Trace/Log data is high sensitivity -> Require Admin Token
+    allowed, error = require_admin_token(request)
     if not allowed:
         return web.json_response({"ok": False, "error": error}, status=403)
 
@@ -364,7 +368,8 @@ async def trace_handler(request: web.Request) -> web.Response:
     ok, init_error = _ensure_observability_deps_ready()
     if not ok:
         return web.json_response({"ok": False, "error": init_error}, status=500)
-    allowed, error = require_observability_access(request)
+    # S34: Trace/Log data is high sensitivity -> Require Admin Token
+    allowed, error = require_admin_token(request)
     if not allowed:
         return web.json_response({"ok": False, "error": error}, status=403)
 

connector/platforms/kakao_webhook.py

@@ -39,6 +39,41 @@ def _import_aiohttp_web():
     return aiohttp, web
 
 
+class _CompatResponse:
+    """Minimal response shim for unit tests when aiohttp is unavailable."""
+
+    def __init__(
+        self,
+        *,
+        status: int = 200,
+        text: str = "",
+        content_type: str = "text/plain",
+        body: Optional[bytes] = None,
+    ):
+        self.status = status
+        self.text = text
+        self.content_type = content_type
+        self.body = body if body is not None else text.encode("utf-8")
+
+
+def _make_response(web, *, status: int = 200, text: str = "OK"):
+    if web is not None:
+        return web.Response(status=status, text=text)
+    return _CompatResponse(status=status, text=text)
+
+
+def _make_json_response(web, data: dict, *, status: int = 200):
+    body = json.dumps(data, ensure_ascii=False).encode("utf-8")
+    if web is not None:
+        return web.json_response(data, status=status)
+    return _CompatResponse(
+        status=status,
+        text=body.decode("utf-8"),
+        content_type="application/json",
+        body=body,
+    )
+
+
 class KakaoWebhookServer:
     """
     KakaoTalk payload adapter.
@@ -112,14 +147,15 @@ class KakaoWebhookServer:
     async def handle_webhook(self, request):
         """POST handler for Kakao Skill payloads."""
         _, web = _import_aiohttp_web()
-        if web is None:
-            raise RuntimeError("aiohttp not available")
+        # IMPORTANT:
+        # CI unit tests call this handler directly without aiohttp installed.
+        # Keep this path runnable; do not replace with a hard RuntimeError.
 
         try:
             body_bytes = await request.read()
             payload = json.loads(body_bytes)
         except json.JSONDecodeError:
-            return web.Response(status=400, text="Bad JSON")
+            return _make_response(web, status=400, text="Bad JSON")
 
         # S32: Replay Guard (Content Hash Dedup)
         # We use a hash of the body bytes as the "nonce" for deduplication.
@@ -127,9 +163,8 @@ class KakaoWebhookServer:
         content_hash = hashlib.sha256(body_bytes).hexdigest()
         if not self._replay_guard.check_and_record(content_hash):
             logger.warning(f"Replay rejected for Kakao hash: {content_hash}")
-            return web.Response(
-                status=200, text="OK"
-            )  # Return 200 to stop Kakao retries
+            # Return 200 to stop Kakao retries
+            return _make_response(web, status=200, text="OK")
 
         # Normalization
         # userRequest.user.id is the opaque user ID (botUserKey)
@@ -169,8 +204,15 @@ class KakaoWebhookServer:
 
         try:
             resp = await self.router.handle(req)
-            if resp.text:
-                return self._build_text_response(resp.text)
+            # IMPORTANT:
+            # Router mocks in unit tests may return non-string `.text` values.
+            # Normalize defensively to avoid turning a valid routing flow into
+            # a JSON serialization error path.
+            resp_text = getattr(resp, "text", "")
+            if not isinstance(resp_text, str):
+                resp_text = str(resp_text) if resp_text is not None else ""
+            if resp_text:
+                return self._build_text_response(resp_text)
             else:
                 # No response content (e.g. valid command but no output intended?)
                 # Kakao requires *some* response payload or it treats as error.
@@ -192,7 +234,7 @@ class KakaoWebhookServer:
             "version": "2.0",
             "template": {"outputs": [{"simpleText": {"text": text}}]},
         }
-        return web.json_response(resp_data)
+        return _make_json_response(web, resp_data)
 
     def _build_error_response(self, error_msg: str):
         """Build simple error text response."""

connector/platforms/wechat_webhook.py

@@ -47,6 +47,35 @@ def _import_aiohttp_web():
     return aiohttp, web
 
 
+class _CompatResponse:
+    """Minimal response shim for unit tests when aiohttp is unavailable."""
+
+    def __init__(
+        self,
+        *,
+        status: int = 200,
+        text: str = "",
+        content_type: str = "text/plain",
+        body: Optional[bytes] = None,
+    ):
+        self.status = status
+        self.text = text
+        self.content_type = content_type
+        self.body = body if body is not None else text.encode("utf-8")
+
+
+def _make_response(
+    web,
+    *,
+    status: int = 200,
+    text: str = "",
+    content_type: str = "text/plain",
+):
+    if web is not None:
+        return web.Response(status=status, text=text, content_type=content_type)
+    return _CompatResponse(status=status, text=text, content_type=content_type)
+
+
 # ---------------------------------------------------------------------------
 # R74 — Protocol constants
 # ---------------------------------------------------------------------------
@@ -343,8 +372,9 @@ class WeChatWebhookServer:
         Must return echostr as plain text if signature is valid.
         """
         _, web = _import_aiohttp_web()
-        if web is None:
-            raise RuntimeError("aiohttp not available")
+        # IMPORTANT:
+        # CI unit tests invoke handler logic directly without aiohttp installed.
+        # Do not hard-raise here; return compat responses so security logic remains testable.
 
         signature = request.query.get("signature", "")
         timestamp = request.query.get("timestamp", "")
@@ -355,10 +385,10 @@ class WeChatWebhookServer:
 
         if verify_wechat_signature(token, timestamp, nonce, signature):
             logger.info("WeChat webhook verification succeeded")
-            return web.Response(text=echostr, content_type="text/plain")
+            return _make_response(web, text=echostr, content_type="text/plain")
 
         logger.warning("WeChat webhook verification failed")
-        return web.Response(status=403, text="Verification failed")
+        return _make_response(web, status=403, text="Verification failed")
 
     # ------------------------------------------------------------------
     # POST — Inbound Messages (R74 + S31)
@@ -367,8 +397,9 @@ class WeChatWebhookServer:
     async def handle_webhook(self, request):
         """POST handler for WeChat XML messages/events."""
         _, web = _import_aiohttp_web()
-        if web is None:
-            raise RuntimeError("aiohttp not available")
+        # IMPORTANT:
+        # Keep handler behavior testable in environments without aiohttp.
+        # Server startup still requires aiohttp, but direct handler unit tests should not crash.
 
         # S31: Signature verification
         signature = request.query.get("signature", "")
@@ -378,12 +409,12 @@ class WeChatWebhookServer:
 
         if not verify_wechat_signature(token, timestamp, nonce, signature):
             logger.warning("Invalid WeChat POST signature")
-            return web.Response(status=401, text="Invalid Signature")
+            return _make_response(web, status=401, text="Invalid Signature")
 
         # S31: Replay protection — nonce dedup
         if nonce and not self._replay_guard.check_and_record(nonce):
             logger.warning(f"Replay rejected for WeChat nonce: {nonce}")
-            return web.Response(status=403, text="Replay Rejected")
+            return _make_response(web, status=403, text="Replay Rejected")
 
         # S31: Timestamp freshness
         try:
@@ -394,7 +425,7 @@ class WeChatWebhookServer:
         age_sec = now - ts_val
         if age_sec > self.REPLAY_WINDOW_SEC or age_sec < -60:
             logger.warning(f"Stale WeChat request: age={age_sec}s")
-            return web.Response(status=403, text="Stale Request")
+            return _make_response(web, status=403, text="Stale Request")
 
         # Read and parse XML with S31 budgets
         body_bytes = await request.read()
@@ -403,13 +434,13 @@ class WeChatWebhookServer:
             fields = parse_wechat_xml(body_bytes)
         except XMLBudgetExceeded as e:
             logger.warning(f"WeChat XML budget exceeded: {e}")
-            return web.Response(status=400, text="Bad Request")
+            return _make_response(web, status=400, text="Bad Request")
 
         # R74: Normalize to canonical event
         event = normalize_wechat_event(fields)
         if event is None:
             # Unsupported message type — return empty success to WeChat
-            return web.Response(text="success", content_type="text/plain")
+            return _make_response(web, text="success", content_type="text/plain")
 
         # S31: Allowlist check (soft-deny)
         sender_id = event["sender_id"]
@@ -429,7 +460,7 @@ class WeChatWebhookServer:
         # Per-message dedup (MsgId-based, distinct from nonce-based replay)
         if message_id and not self._replay_guard.check_and_record(f"msg:{message_id}"):
             logger.debug(f"Duplicate WeChat MsgId: {message_id}")
-            return web.Response(text="success", content_type="text/plain")
+            return _make_response(web, text="success", content_type="text/plain")
 
         req = CommandRequest(
             platform="wechat",
@@ -448,14 +479,15 @@ class WeChatWebhookServer:
                 to_user = event["sender_id"]
                 from_user = event["to_user"]
                 reply_xml = build_text_reply_xml(to_user, from_user, resp.text)
-                return web.Response(
+                return _make_response(
+                    web,
                     text=reply_xml,
                     content_type="application/xml",
                 )
         except Exception as e:
             logger.exception(f"Error handling WeChat command: {e}")
 
-        return web.Response(text="success", content_type="text/plain")
+        return _make_response(web, text="success", content_type="text/plain")
 
     # ------------------------------------------------------------------
     # Outbound: Text (Customer Service Message API)

debug_s35_import.py

@@ -0,0 +1,32 @@
+"""
+Debug script for S35 Transform Isolation.
+Verifies that the correct executor (TransformProcessRunner) is allowed/loaded.
+"""
+
+import os
+import sys
+
+# Ensure project root is in path
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from services.constrained_transforms import get_transform_executor
+from services.transform_runner import TransformProcessRunner
+
+
+def main():
+    print("Checking S35 Transform Executor...")
+    executor = get_transform_executor()
+    print(f"Executor Type: {type(executor)}")
+
+    if isinstance(executor, TransformProcessRunner):
+        print("SUCCESS: TransformProcessRunner is active (Isolation Enabled).")
+    else:
+        print("WARNING: TransformProcessRunner is NOT active.")
+        # Check if feature flag is enabled
+        from services.transform_common import is_transforms_enabled
+
+        print(f"Feature Flag Enabled: {is_transforms_enabled()}")
+
+
+if __name__ == "__main__":
+    main()

scripts/pre_push_checks.sh

@@ -40,13 +40,50 @@ require_cmd() {
   fi
 }
 
+pip_install_or_fail() {
+  local why="$1"
+  shift
+  if "$VENV_PY" -m pip install "$@"; then
+    return 0
+  fi
+  echo "[pre-push] ERROR: failed to install dependency ($why): $*" >&2
+  exit 1
+}
+
+is_wsl() {
+  grep -qiE "(microsoft|wsl)" /proc/version 2>/dev/null
+}
+
+select_venv_dir() {
+  # Explicit override for advanced/local setups.
+  if [ -n "${OPENCLAW_TEST_VENV:-}" ]; then
+    echo "$OPENCLAW_TEST_VENV"
+    return 0
+  fi
+
+  case "$UNAME_S" in
+    MINGW*|MSYS*|CYGWIN*)
+      echo "$ROOT_DIR/.venv"
+      ;;
+    *)
+      # IMPORTANT:
+      # In WSL, prefer dedicated Linux venv to avoid mixing with Windows .venv.
+      if is_wsl; then
+        echo "$ROOT_DIR/.venv-wsl"
+      else
+        echo "$ROOT_DIR/.venv"
+      fi
+      ;;
+  esac
+}
+
 resolve_venv_python() {
   case "$UNAME_S" in
     MINGW*|MSYS*|CYGWIN*)
-      echo "$ROOT_DIR/.venv/Scripts/python.exe"
+      echo "$VENV_DIR/Scripts/python.exe"
       ;;
     *)
-      echo "$ROOT_DIR/.venv/bin/python"
+      echo "$VENV_DIR/bin/python"
       ;;
   esac
 }
@@ -70,24 +107,24 @@ bootstrap_venv() {
   fi
 
   if [ -e "$venv_py" ]; then
-    echo "[pre-push] WARN: existing .venv is invalid; recreating with a Windows-native Python." >&2
-    rm -rf "$ROOT_DIR/.venv"
+    echo "[pre-push] WARN: existing venv is invalid; recreating: $VENV_DIR" >&2
+    rm -rf "$VENV_DIR"
   fi
 
-  echo "[pre-push] INFO: creating project .venv ..." >&2
+  echo "[pre-push] INFO: creating project venv at $VENV_DIR ..." >&2
   case "$UNAME_S" in
     MINGW*|MSYS*|CYGWIN*)
       # CRITICAL: on Git Bash, `python3` may resolve to MSYS `/usr/bin/python`,
       # which creates a broken Windows venv (`No Python at "/usr/bin\python.exe"`).
       # Always prefer Windows-native launchers/interpreters.
       if command -v py.exe >/dev/null 2>&1; then
-        py.exe -3 -m venv "$ROOT_DIR/.venv"
+        py.exe -3 -m venv "$VENV_DIR"
       elif [ -x "/c/Windows/py.exe" ]; then
-        /c/Windows/py.exe -3 -m venv "$ROOT_DIR/.venv"
+        /c/Windows/py.exe -3 -m venv "$VENV_DIR"
       elif command -v python.exe >/dev/null 2>&1; then
-        python.exe -m venv "$ROOT_DIR/.venv"
+        python.exe -m venv "$VENV_DIR"
       elif command -v py >/dev/null 2>&1; then
-        py -3 -m venv "$ROOT_DIR/.venv"
+        py -3 -m venv "$VENV_DIR"
       else
         echo "[pre-push] ERROR: no Windows Python launcher found (py.exe/python.exe)." >&2
         exit 1
@@ -95,9 +132,9 @@ bootstrap_venv() {
       ;;
     *)
       if command -v python3 >/dev/null 2>&1; then
-        python3 -m venv "$ROOT_DIR/.venv"
+        python3 -m venv "$VENV_DIR"
       elif command -v python >/dev/null 2>&1; then
-        python -m venv "$ROOT_DIR/.venv"
+        python -m venv "$VENV_DIR"
       else
         echo "[pre-push] ERROR: no bootstrap Python found (python3/python)." >&2
         exit 1
@@ -106,7 +143,7 @@ bootstrap_venv() {
   esac
 
   if ! is_venv_python_healthy "$venv_py"; then
-    echo "[pre-push] ERROR: failed to initialize project .venv." >&2
+    echo "[pre-push] ERROR: failed to initialize project venv: $VENV_DIR" >&2
     exit 1
   fi
   echo "$venv_py"
@@ -116,19 +153,33 @@ pre_commit_cmd() {
   "$VENV_PY" -m pre_commit "$@"
 }
 
-# CRITICAL: pre-push must always run pre-commit from project .venv.
+# CRITICAL: pre-push must always run pre-commit from project venv.
 # Do not switch this back to global `pre-commit` command lookup.
 # This prevents mixed global/user installs from hijacking hook execution.
+VENV_DIR="$(select_venv_dir)"
 VENV_PY="$(bootstrap_venv)"
 if ! "$VENV_PY" -m pre_commit --version >/dev/null 2>&1; then
-  echo "[pre-push] INFO: installing pre-commit into project .venv ..." >&2
-  "$VENV_PY" -m pip install -U pip pre-commit
+  echo "[pre-push] INFO: installing pre-commit into project venv ($VENV_DIR) ..." >&2
+  pip_install_or_fail "required for pre-commit hooks" -U pip pre-commit
 fi
 if ! "$VENV_PY" -c "import black" >/dev/null 2>&1; then
   # Keep black in the same interpreter used by local black-single hook.
-  echo "[pre-push] INFO: installing black into project .venv ..." >&2
-  "$VENV_PY" -m pip install black==24.1.1
+  echo "[pre-push] INFO: installing black into project venv ($VENV_DIR) ..." >&2
+  pip_install_or_fail "required by black-single hook" black==24.1.1
 fi
+
+# IMPORTANT:
+# Pre-push now runs backend unit tests. Keep minimal runtime deps aligned with
+# CI unit-test job to avoid "passes locally, fails on GitHub" drift.
+if ! "$VENV_PY" -c "import numpy, PIL" >/dev/null 2>&1; then
+  echo "[pre-push] INFO: installing numpy/pillow into project venv ($VENV_DIR) ..." >&2
+  pip_install_or_fail "required by unit tests" numpy pillow
+fi
+if ! "$VENV_PY" -c "import aiohttp" >/dev/null 2>&1; then
+  echo "[pre-push] INFO: installing aiohttp into project venv ($VENV_DIR) ..." >&2
+  pip_install_or_fail "required by unit tests/import paths" aiohttp
+fi
+
 require_cmd npm
 
 run_pre_commit_safe() {
@@ -225,13 +276,17 @@ if [ "$NODE_MAJOR" -lt 18 ]; then
 fi
 
 echo "[pre-push] Node version: $(node -v)"
-echo "[pre-push] 1/3 detect-secrets"
+echo "[pre-push] 1/4 detect-secrets"
 run_pre_commit_safe run detect-secrets --all-files
 
-echo "[pre-push] 2/3 pre-commit all hooks"
+echo "[pre-push] 2/4 pre-commit all hooks"
 run_pre_commit_safe run --all-files
 
-echo "[pre-push] 3/3 npm test (Playwright)"
+echo "[pre-push] 3/4 backend unit tests"
+MOLTBOT_STATE_DIR="$ROOT_DIR/moltbot_state/_pre_push_unit" \
+  "$VENV_PY" scripts/run_unittests.py --start-dir tests --pattern "test_*.py"
+
+echo "[pre-push] 4/4 npm test (Playwright)"
 npm test
 
 echo "[pre-push] PASS"

scripts/precommit_black_single.sh

@@ -6,30 +6,64 @@ ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
 export BLACK_CACHE_DIR="${BLACK_CACHE_DIR:-$ROOT_DIR/.tmp/black-cache}"
 mkdir -p "$BLACK_CACHE_DIR"
 
-# CRITICAL: Always prefer project-local .venv interpreter for Black.
+# CRITICAL: Always prefer project-local venv interpreter for Black.
 # Without this, Windows can accidentally pick global Python (e.g. C:\Program Files\Python312)
 # where `black` is not installed, causing flaky pre-commit failures.
 # DO NOT change this back to "python/python3 from PATH first" unless you also
 # guarantee black is installed in every global interpreter used by contributors.
+
+is_wsl() {
+  grep -qiE "(microsoft|wsl)" /proc/version 2>/dev/null
+}
+
 can_use_python() {
   local candidate="$1"
   [ -f "$candidate" ] || return 1
   "$candidate" -c "import sys; print(sys.executable)" >/dev/null 2>&1
 }
 
-has_project_venv=false
-[ -d "$ROOT_DIR/.venv" ] && has_project_venv=true
+select_venv_dirs() {
+  local dirs=()
+  if [ -n "${OPENCLAW_TEST_VENV:-}" ]; then
+    dirs+=("$OPENCLAW_TEST_VENV")
+  fi
+  if is_wsl; then
+    dirs+=("$ROOT_DIR/.venv-wsl")
+  fi
+  dirs+=("$ROOT_DIR/.venv")
+  printf '%s\n' "${dirs[@]}"
+}
 
-if can_use_python "$ROOT_DIR/.venv/Scripts/python.exe"; then
-  PY_CMD="$ROOT_DIR/.venv/Scripts/python.exe"
-elif can_use_python "$ROOT_DIR/.venv/bin/python"; then
-  PY_CMD="$ROOT_DIR/.venv/bin/python"
-elif [ "$has_project_venv" = true ]; then
-  # CRITICAL: if .venv exists but is broken, fail fast instead of silently
-  # falling back to a random global Python (which reintroduces flakiness).
-  # DO NOT relax this guard; it is intentional to prevent silent environment drift.
-  echo "ERROR: project .venv exists but Python is unusable. Recreate .venv and retry." >&2
-  exit 1
+check_venv_dir() {
+  local dir="$1"
+  if can_use_python "$dir/Scripts/python.exe"; then
+    PY_CMD="$dir/Scripts/python.exe"
+    return 0
+  fi
+  if can_use_python "$dir/bin/python"; then
+    PY_CMD="$dir/bin/python"
+    return 0
+  fi
+  return 1
+}
+
+while IFS= read -r vdir; do
+  [ -n "$vdir" ] || continue
+  if [ -d "$vdir" ]; then
+    if check_venv_dir "$vdir"; then
+      break
+    fi
+    # IMPORTANT:
+    # If a preferred project venv exists but is unusable, fail fast instead of
+    # silently falling back to random global Python (prevents environment drift).
+    echo "ERROR: project venv exists but Python is unusable: $vdir" >&2
+    echo "Recreate this venv and retry." >&2
+    exit 1
+  fi
+done < <(select_venv_dirs)
+
+if [ -n "${PY_CMD:-}" ]; then
+  :
 elif command -v python >/dev/null 2>&1; then
   # Fallback chain only for environments that intentionally do not use .venv.
   PY_CMD="$(command -v python)"

scripts/run_full_tests_linux.sh

@@ -22,6 +22,25 @@ require_cmd() {
   fi
 }
 
+is_wsl() {
+  grep -qiE "(microsoft|wsl)" /proc/version 2>/dev/null
+}
+
+select_venv_dir() {
+  # Explicit override for advanced/local setups.
+  if [ -n "${OPENCLAW_TEST_VENV:-}" ]; then
+    echo "$OPENCLAW_TEST_VENV"
+    return 0
+  fi
+  # IMPORTANT:
+  # In WSL, prefer dedicated Linux venv to avoid clashing with Windows .venv.
+  if is_wsl; then
+    echo "$ROOT_DIR/.venv-wsl"
+  else
+    echo "$ROOT_DIR/.venv"
+  fi
+}
+
 pip_install_or_fail() {
   local why="$1"
   shift
@@ -30,7 +49,7 @@ pip_install_or_fail() {
   fi
   echo "[tests] ERROR: failed to install dependency ($why): $*" >&2
   echo "[tests] HINT: check internet/proxy, then retry the script." >&2
-  echo "[tests] HINT: if offline, pre-install into .venv manually: $VENV_PY -m pip install $*" >&2
+  echo "[tests] HINT: if offline, pre-install into venv manually: $VENV_PY -m pip install $*" >&2
   exit 1
 }
 
@@ -38,13 +57,14 @@ require_cmd node
 require_cmd npm
 
 # Always use project-local venv to avoid global interpreter / tool drift.
-VENV_PY="$ROOT_DIR/.venv/bin/python"
+VENV_DIR="$(select_venv_dir)"
+VENV_PY="$VENV_DIR/bin/python"
 if [ ! -x "$VENV_PY" ]; then
-  echo "[tests] Creating project venv at $ROOT_DIR/.venv ..."
+  echo "[tests] Creating project venv at $VENV_DIR ..."
   if command -v python3 >/dev/null 2>&1; then
-    python3 -m venv "$ROOT_DIR/.venv"
+    python3 -m venv "$VENV_DIR"
   elif command -v python >/dev/null 2>&1; then
-    python -m venv "$ROOT_DIR/.venv"
+    python -m venv "$VENV_DIR"
   else
     echo "[tests] ERROR: no bootstrap Python found (need python3 or python)" >&2
     exit 1
@@ -52,12 +72,12 @@ if [ ! -x "$VENV_PY" ]; then
 fi
 
 if ! "$VENV_PY" -m pre_commit --version >/dev/null 2>&1; then
-  echo "[tests] Installing pre-commit into project venv ..."
+  echo "[tests] Installing pre-commit into project venv ($VENV_DIR) ..."
   pip_install_or_fail "required for detect-secrets and hook validation" -U pip pre-commit
 fi
 
 if ! "$VENV_PY" -c "import aiohttp" >/dev/null 2>&1; then
-  echo "[tests] Installing aiohttp into project venv ..."
+  echo "[tests] Installing aiohttp into project venv ($VENV_DIR) ..."
   pip_install_or_fail "required by import paths used in unit tests" aiohttp
 fi
 

services/approvals/storage.py

@@ -57,16 +57,19 @@ def _atomic_write(path: str, data: Dict) -> None:
         raise
 
 
+from ..integrity import IntegrityError, load_verified, save_verified
+
+
 def load_approvals() -> Dict[str, ApprovalRequest]:
-    """Load approvals from disk."""
+    """Load approvals from disk with integrity check."""
     path = _get_approvals_path()
 
     if not os.path.exists(path):
         return {}
 
     try:
-        with open(path, "r", encoding="utf-8") as f:
-            data = json.load(f)
+        # R77: Load verification
+        data = load_verified(path, expected_version=1, migrate=True)
 
         result = {}
         if isinstance(data, dict) and "approvals" in data:
@@ -79,13 +82,20 @@ def load_approvals() -> Dict[str, ApprovalRequest]:
 
         logger.info(f"Loaded {len(result)} approval records")
         return result
-    except (json.JSONDecodeError, OSError) as e:
+    except IntegrityError as e:
+        # R77: Fail-closed logic with escalation
+        logger.critical(
+            f"R77: Integrity violation detected in approvals file {path}: {e}"
+        )
+        # Return empty (deny all pending approvals) which is safe fail-state
+        return {}
+    except Exception as e:
         logger.error(f"Failed to load approvals: {e}")
         return {}
 
 
 def save_approvals(approvals: Dict[str, ApprovalRequest]) -> bool:
-    """Save approvals to disk."""
+    """Save approvals to disk with integrity envelope."""
     path = _get_approvals_path()
 
     try:
@@ -94,7 +104,8 @@ def save_approvals(approvals: Dict[str, ApprovalRequest]) -> bool:
             "saved_at": datetime.now(timezone.utc).isoformat(),
             "approvals": [a.to_dict() for a in approvals.values()],
         }
-        _atomic_write(path, data)
+        # R77: Atomic verified save
+        save_verified(path, data, version=1)
         return True
     except Exception as e:
         logger.error(f"Failed to save approvals: {e}")

services/checkpoints.py

@@ -18,6 +18,8 @@ except ImportError:
     # Fallback for tests or decoupled run
     DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data")
 
+from .integrity import IntegrityError, load_verified, save_verified
+
 logger = logging.getLogger("ComfyUI-OpenClaw.services.checkpoints")
 
 CHECKPOINTS_DIR = os.path.join(DATA_DIR, "checkpoints")
@@ -45,9 +47,14 @@ def list_checkpoints() -> List[Dict[str, Any]]:
             for entry in it:
                 if entry.name.endswith(".meta.json") and entry.is_file():
                     try:
-                        with open(entry.path, "r", encoding="utf-8") as f:
-                            meta = json.load(f)
-                            checkpoints.append(meta)
+
+                        meta = load_verified(entry.path, migrate=True)
+                        checkpoints.append(meta)
+                    except IntegrityError as e:
+                        logger.critical(
+                            f"R77: Integrity violation in checkpoint {entry.name}: {e}"
+                        )
+                        # Skip this file (fail-closed for this item)
                     except Exception:
                         logger.warning(f"Failed to read checkpoint meta: {entry.name}")
     except OSError:
@@ -66,12 +73,14 @@ def get_checkpoint(checkpoint_id: str) -> Optional[Dict[str, Any]]:
         return None
 
     try:
-        with open(meta_path, "r", encoding="utf-8") as f:
-            meta = json.load(f)
-        with open(payload_path, "r", encoding="utf-8") as f:
-            workflow = json.load(f)
+
+        meta = load_verified(meta_path, migrate=True)
+        workflow = load_verified(payload_path, migrate=True)
 
         return {"id": checkpoint_id, "meta": meta, "workflow": workflow}
+    except IntegrityError as e:
+        logger.critical(f"R77: Integrity violation in checkpoint {checkpoint_id}: {e}")
+        return None
     except Exception as e:
         logger.error(f"Error reading checkpoint {checkpoint_id}: {e}")
         return None
@@ -148,8 +157,8 @@ def create_checkpoint(
     meta_path, payload_path = _get_paths(cid)
 
     try:
-        _atomic_write(meta_path, json.dumps(meta, indent=2))
-        _atomic_write(payload_path, workflow_json)
+        save_verified(meta_path, meta)
+        save_verified(payload_path, workflow)
     except Exception as e:
         # Cleanup on fail (although atomic write minimizes this risk for individual files)
         if os.path.exists(meta_path):

services/constrained_transforms.py

@@ -16,311 +16,55 @@ Default posture: DISABLED. Requires OPENCLAW_ENABLE_TRANSFORMS=1.
 
 from __future__ import annotations
 
-import hashlib
 import importlib
 import importlib.util
 import json
 import logging
 import os
-import signal
-import sys
 import threading
 import time
-from dataclasses import asdict, dataclass, field
-from enum import Enum
-from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Set
+from typing import Any, Dict, List, Optional
+
+from .transform_common import (
+    _FEATURE_FLAG,
+    DEFAULT_MAX_OUTPUT_BYTES,
+    DEFAULT_MAX_TRANSFORMS_PER_REQUEST,
+    DEFAULT_TRANSFORM_TIMEOUT_SEC,
+    MAX_TRANSFORM_MODULE_SIZE_BYTES,
+    TransformLimits,
+    TransformRegistry,
+    TransformRegistryError,
+    TransformResult,
+    TransformStatus,
+    TrustedTransform,
+    get_transform_registry,
+    is_transforms_enabled,
+)
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.constrained_transforms")
 
-# ---------------------------------------------------------------------------
-# Feature gate
-# ---------------------------------------------------------------------------
-
-_FEATURE_FLAG = "OPENCLAW_ENABLE_TRANSFORMS"
-
-
-def is_transforms_enabled() -> bool:
-    """Check if constrained transforms are enabled (default: OFF)."""
-    val = os.environ.get(_FEATURE_FLAG, "").strip().lower()
-    return val in ("1", "true", "yes", "on")
-
-
-# ---------------------------------------------------------------------------
-# Runtime limits
-# ---------------------------------------------------------------------------
-
-DEFAULT_TRANSFORM_TIMEOUT_SEC = 5
-DEFAULT_MAX_OUTPUT_BYTES = 64 * 1024  # 64KB
-DEFAULT_MAX_TRANSFORMS_PER_REQUEST = 5
-MAX_TRANSFORM_MODULE_SIZE_BYTES = 50 * 1024  # 50KB — prevent loading huge scripts
-
-
-@dataclass
-class TransformLimits:
-    """Runtime limits for transform execution."""
-
-    timeout_sec: float = DEFAULT_TRANSFORM_TIMEOUT_SEC
-    max_output_bytes: int = DEFAULT_MAX_OUTPUT_BYTES
-    max_transforms_per_request: int = DEFAULT_MAX_TRANSFORMS_PER_REQUEST
-
-    @classmethod
-    def from_env(cls) -> "TransformLimits":
-        """Load limits from environment variables."""
-
-        def _env_int(key: str, default: int) -> int:
-            try:
-                return int(os.environ.get(key, str(default)))
-            except (ValueError, TypeError):
-                return default
-
-        def _env_float(key: str, default: float) -> float:
-            try:
-                return float(os.environ.get(key, str(default)))
-            except (ValueError, TypeError):
-                return default
-
-        return cls(
-            timeout_sec=_env_float(
-                "OPENCLAW_TRANSFORM_TIMEOUT", DEFAULT_TRANSFORM_TIMEOUT_SEC
-            ),
-            max_output_bytes=_env_int(
-                "OPENCLAW_TRANSFORM_MAX_OUTPUT", DEFAULT_MAX_OUTPUT_BYTES
-            ),
-            max_transforms_per_request=_env_int(
-                "OPENCLAW_TRANSFORM_MAX_PER_REQUEST", DEFAULT_MAX_TRANSFORMS_PER_REQUEST
-            ),
-        )
-
-
-# ---------------------------------------------------------------------------
-# Transform result
-# ---------------------------------------------------------------------------
-
-
-class TransformStatus(str, Enum):
-    SUCCESS = "success"
-    ERROR = "error"
-    TIMEOUT = "timeout"
-    DENIED = "denied"
-    SKIPPED = "skipped"
-
-
-@dataclass
-class TransformResult:
-    """Result of a single transform execution."""
-
-    transform_id: str
-    status: str  # TransformStatus.value
-    output: Optional[Dict[str, Any]] = None
-    error: str = ""
-    duration_ms: float = 0.0
-    output_bytes: int = 0
-    audit: Dict[str, Any] = field(default_factory=dict)
-
-    def to_dict(self) -> Dict[str, Any]:
-        d: Dict[str, Any] = {
-            "transform_id": self.transform_id,
-            "status": self.status,
-            "duration_ms": round(self.duration_ms, 2),
-            "output_bytes": self.output_bytes,
-        }
-        if self.output is not None:
-            d["output"] = self.output
-        if self.error:
-            d["error"] = self.error
-        if self.audit:
-            d["audit"] = self.audit
-        return d
-
-
-# ---------------------------------------------------------------------------
-# Transform registry (trusted modules)
-# ---------------------------------------------------------------------------
-
-
-@dataclass
-class TrustedTransform:
-    """A registered, integrity-pinned transform module."""
-
-    id: str
-    label: str
-    module_path: str  # Absolute path to .py module
-    sha256: str  # Integrity hash of the module file
-    description: str = ""
-    trusted_source: str = ""  # Who published this transform
-    registered_at: float = 0.0
-
-    def to_dict(self) -> Dict[str, Any]:
-        return asdict(self)
-
-
-class TransformRegistryError(Exception):
-    """Error in transform registry operations."""
-
-    pass
-
-
-class TransformRegistry:
-    """
-    Manages trusted transform modules with integrity pinning.
-
-    Transforms can only be loaded from explicitly trusted directories.
-    Each module is pinned by its SHA256 hash at registration time.
-    """
-
-    def __init__(self, state_dir: str, trusted_dirs: Optional[List[str]] = None):
-        self._state_dir = state_dir
-        self._registry_dir = os.path.join(state_dir, "transforms")
-        self._index_path = os.path.join(self._registry_dir, "registry.json")
-        self._transforms: Dict[str, TrustedTransform] = {}
-
-        # Trusted directories where transform modules can live
-        self._trusted_dirs: Set[str] = set()
-        if trusted_dirs:
-            for d in trusted_dirs:
-                resolved = str(Path(d).resolve())
-                self._trusted_dirs.add(resolved)
-
-        os.makedirs(self._registry_dir, exist_ok=True)
-        self._load()
-
-    def _load(self) -> None:
-        """Load transform registry from disk."""
-        if not os.path.exists(self._index_path):
-            self._transforms = {}
-            return
-        try:
-            with open(self._index_path, "r", encoding="utf-8") as f:
-                data = json.load(f)
-            self._transforms = {}
-            for tid, tdata in data.items():
-                self._transforms[tid] = TrustedTransform(**tdata)
-        except Exception as e:
-            logger.error(f"Failed to load transform registry: {e}")
-            self._transforms = {}
-
-    def _save(self) -> None:
-        """Persist transform registry to disk."""
-        try:
-            data = {k: v.to_dict() for k, v in self._transforms.items()}
-            tmp_path = self._index_path + ".tmp"
-            with open(tmp_path, "w", encoding="utf-8") as f:
-                json.dump(data, f, indent=2)
-                f.write("\n")
-            os.replace(tmp_path, self._index_path)
-        except Exception as e:
-            logger.error(f"Failed to save transform registry: {e}")
-
-    @staticmethod
-    def _compute_sha256(file_path: str) -> str:
-        """Compute SHA256 hash of a file."""
-        h = hashlib.sha256()
-        with open(file_path, "rb") as f:
-            for chunk in iter(lambda: f.read(4096), b""):
-                h.update(chunk)
-        return h.hexdigest()
-
-    def _is_in_trusted_dir(self, module_path: str) -> bool:
-        """Check if a module path is inside a trusted directory."""
-        resolved = str(Path(module_path).resolve())
-        for trusted in self._trusted_dirs:
-            if resolved.startswith(trusted + os.sep) or resolved == trusted:
-                return True
-        return False
-
-    def register_transform(
-        self,
-        transform_id: str,
-        module_path: str,
-        *,
-        label: str = "",
-        description: str = "",
-        trusted_source: str = "",
-    ) -> TrustedTransform:
-        """
-        Register a transform module with integrity pinning.
-
-        The module must be in a trusted directory and within size limits.
-        """
-        if not is_transforms_enabled():
-            raise TransformRegistryError(
-                f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable."
-            )
-
-        abs_path = str(Path(module_path).resolve())
-
-        # Security: must be in trusted directory
-        if not self._is_in_trusted_dir(abs_path):
-            raise TransformRegistryError(
-                f"Module path is not in a trusted directory: {abs_path}"
-            )
-
-        if not os.path.isfile(abs_path):
-            raise TransformRegistryError(f"Module file not found: {abs_path}")
-
-        # Size check
-        file_size = os.path.getsize(abs_path)
-        if file_size > MAX_TRANSFORM_MODULE_SIZE_BYTES:
-            raise TransformRegistryError(
-                f"Module exceeds size limit ({file_size} > {MAX_TRANSFORM_MODULE_SIZE_BYTES})"
-            )
-
-        # Must be a .py file
-        if not abs_path.endswith(".py"):
-            raise TransformRegistryError("Only .py modules are allowed as transforms")
-
-        sha256 = self._compute_sha256(abs_path)
-
-        transform = TrustedTransform(
-            id=transform_id,
-            label=label or transform_id,
-            module_path=abs_path,
-            sha256=sha256,
-            description=description,
-            trusted_source=trusted_source,
-            registered_at=time.time(),
-        )
-
-        self._transforms[transform_id] = transform
-        self._save()
-        logger.info(f"F42: Registered transform '{transform_id}' from {abs_path}")
-        return transform
-
-    def unregister_transform(self, transform_id: str) -> bool:
-        """Remove a transform from the registry."""
-        if not is_transforms_enabled():
-            raise TransformRegistryError(
-                f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable."
-            )
-
-        if transform_id not in self._transforms:
-            return False
-
-        del self._transforms[transform_id]
-        self._save()
-        logger.info(f"F42: Unregistered transform '{transform_id}'")
-        return True
-
-    def get_transform(self, transform_id: str) -> Optional[TrustedTransform]:
-        """Get a registered transform by ID."""
-        return self._transforms.get(transform_id)
-
-    def list_transforms(self) -> List[TrustedTransform]:
-        """List all registered transforms."""
-        return list(self._transforms.values())
-
-    def verify_integrity(self, transform_id: str) -> bool:
-        """Verify that a registered transform's file hasn't been modified."""
-        transform = self._transforms.get(transform_id)
-        if not transform:
-            return False
-
-        if not os.path.isfile(transform.module_path):
-            return False
-
-        actual_hash = self._compute_sha256(transform.module_path)
-        return actual_hash == transform.sha256
+# IMPORTANT:
+# Keep compatibility exports in this module even after refactor to
+# `services.transform_common`; tests and downstream imports still reference
+# `services.constrained_transforms` directly.
+__all__ = [
+    "_FEATURE_FLAG",
+    "DEFAULT_MAX_OUTPUT_BYTES",
+    "DEFAULT_MAX_TRANSFORMS_PER_REQUEST",
+    "DEFAULT_TRANSFORM_TIMEOUT_SEC",
+    "MAX_TRANSFORM_MODULE_SIZE_BYTES",
+    "TransformLimits",
+    "TransformRegistry",
+    "TransformRegistryError",
+    "TransformResult",
+    "TransformStatus",
+    "TrustedTransform",
+    "TransformExecutor",
+    "TransformTimeoutError",
+    "get_transform_executor",
+    "get_transform_registry",
+    "is_transforms_enabled",
+]
 
 
 # ---------------------------------------------------------------------------
@@ -536,47 +280,34 @@ class TransformExecutor:
 # Module-level convenience
 # ---------------------------------------------------------------------------
 
-_registry: Optional[TransformRegistry] = None
+
 _executor: Optional[TransformExecutor] = None
 
 
-def get_transform_registry() -> TransformRegistry:
-    """Get or create the global transform registry."""
-    global _registry
-    if _registry is None:
-        try:
-            from .state_dir import get_state_dir
-
-            state_dir = get_state_dir()
-        except ImportError:
-            try:
-                from services.state_dir import get_state_dir
-
-                state_dir = get_state_dir()
-            except ImportError:
-                state_dir = os.path.join(
-                    os.path.dirname(os.path.dirname(__file__)), "data"
-                )
-
-        # Default trusted directory: pack-local transforms dir
-        pack_root = Path(__file__).resolve().parent.parent
-        trusted_dirs = [str(pack_root / "data" / "transforms")]
-
-        # Allow additional trusted dirs from env
-        extra = os.environ.get("OPENCLAW_TRANSFORM_TRUSTED_DIRS", "")
-        if extra:
-            for d in extra.split(os.pathsep):
-                d = d.strip()
-                if d:
-                    trusted_dirs.append(d)
-
-        _registry = TransformRegistry(state_dir, trusted_dirs=trusted_dirs)
-    return _registry
-
-
 def get_transform_executor() -> TransformExecutor:
-    """Get or create the global transform executor."""
+    """
+    Get or create the global transform executor.
+
+    If S35 isolation is enabled (default: True in this hardening wave),
+    returns a TransformProcessRunner instance.
+    """
     global _executor
     if _executor is None:
-        _executor = TransformExecutor(get_transform_registry())
+        # Check for process isolation flag (defaulting to on for S35)
+        # We can use the same enable flag, or a specific isolation one.
+        # Let's assume strict isolation is part of the enabling.
+
+        # Local import to avoid circular dependency
+        try:
+            from .transform_runner import TransformProcessRunner
+
+            registry = get_transform_registry()
+            # We treat TransformProcessRunner as compatible with TransformExecutor interface
+            _executor = TransformProcessRunner(registry)  # type: ignore
+        except ImportError as e:
+            logger.warning(
+                f"S35: Could not import transform_runner ({e}), falling back to thread executor."
+            )
+            _executor = TransformExecutor(get_transform_registry())
+
     return _executor

services/csrf_protection.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 """
 S26+: Same-Origin CSRF Protection for Localhost Convenience Mode
 
@@ -7,16 +9,38 @@ when no admin token is configured (convenience mode).
 Purpose: Prevent cross-origin requests from abusing localhost-only admin endpoints.
 """
 
+import json
 import logging
 import os
 from typing import Optional
 
-from aiohttp import web
+try:
+    from aiohttp import web  # type: ignore
+except ModuleNotFoundError:  # pragma: no cover - CI/minimal env path
+    web = None  # type: ignore
+
+# IMPORTANT:
+# Keep this module importable even when aiohttp is unavailable (CI/minimal env).
+# Tests import `is_same_origin_request` directly; a hard import error here causes
+# unrelated auth-hardening suites to fail before skip guards can apply.
+
+
+class _CompatResponse:
+    """Minimal fallback response when aiohttp is unavailable."""
+
+    def __init__(self, *, status: int, body: bytes, content_type: str):
+        self.status = status
+        self.body = body
+        self.text = body.decode("utf-8", errors="replace")
+        self.content_type = content_type
+
+
+def _json_response(payload: dict, *, status: int):
+    if web is not None:
+        return web.json_response(payload, status=status)
+    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+    return _CompatResponse(status=status, body=body, content_type="application/json")
 
-# NOTE FOR CALLERS:
-# This module has an import-time aiohttp dependency. If a caller must remain
-# importable in minimal test/CI environments without aiohttp, guard the import
-# at the caller boundary (see `api/config.py` for the required pattern).
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.csrf_protection")
 
@@ -117,7 +141,7 @@ def require_same_origin_if_no_token(
         logger.warning(
             f"S26+: CSRF protection denied cross-origin request to {request.path}"
         )
-        return web.json_response(
+        return _json_response(
             {
                 "ok": False,
                 "error": "csrf_protection",

services/integrity.py

@@ -0,0 +1,136 @@
+"""
+R77 Integrity Envelopes.
+
+Provides canonical serialization and integrity verification for persisted state.
+"""
+
+import hashlib
+import json
+import logging
+import os
+import shutil
+import tempfile
+from dataclasses import asdict, dataclass
+from typing import Any, Dict, Optional, Union
+
+logger = logging.getLogger("ComfyUI-OpenClaw.services.integrity")
+
+
+@dataclass
+class IntegrityEnvelope:
+    """
+    Wrapper for persisted data with integrity metadata.
+    """
+
+    version: int
+    data: Dict[str, Any]
+    hash: str  # SHA256 of canonical(data)
+    algo: str = "sha256"
+    meta: Optional[Dict[str, Any]] = None
+
+
+class IntegrityError(Exception):
+    """Raised when integrity verification fails."""
+
+    pass
+
+
+def canonical_dumps(data: Any) -> bytes:
+    """
+    Serialize data to canonical JSON (sorted keys, no whitespace).
+    """
+    return json.dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def calculate_hash(data: Any, algo: str = "sha256") -> str:
+    """
+    Calculate hash of canonicalized data.
+    """
+    if algo != "sha256":
+        raise ValueError(f"Unsupported hash algorithm: {algo}")
+
+    payload = canonical_dumps(data)
+    return hashlib.sha256(payload).hexdigest()
+
+
+def save_verified(path: str, data: Dict[str, Any], version: int = 1) -> None:
+    """
+    Save data wrapped in an integrity envelope.
+    Atomic write.
+    """
+    data_hash = calculate_hash(data)
+    envelope = IntegrityEnvelope(
+        version=version, data=data, hash=data_hash, algo="sha256"
+    )
+
+    # Write to temp string first to Ensure serialization works
+    try:
+        content = json.dumps(asdict(envelope), indent=2)
+    except Exception as e:
+        logger.error(f"Failed to serialize integrity envelope for {path}: {e}")
+        raise
+
+    # Atomic write
+    dir_name = os.path.dirname(os.path.abspath(path))
+    os.makedirs(dir_name, exist_ok=True)
+
+    fd, tmp_path = tempfile.mkstemp(dir=dir_name, text=True)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            f.write(content)
+            f.flush()
+            os.fsync(fd)
+
+        # Renaissance-style atomic rename
+        os.replace(tmp_path, path)
+    except Exception as e:
+        logger.error(f"Failed to save verified file {path}: {e}")
+        if os.path.exists(tmp_path):
+            os.remove(tmp_path)
+        raise
+
+
+def load_verified(
+    path: str, expected_version: int = 1, migrate: bool = True
+) -> Dict[str, Any]:
+    """
+    Load data from an integrity envelope.
+
+    If `migrate` is True and the file is valid legacy JSON (no envelope),
+    it returns the data as-is (caller should save back to upgrade).
+
+    Raises IntegrityError if hash mismatch or malformed.
+    """
+    if not os.path.exists(path):
+        raise FileNotFoundError(f"File not found: {path}")
+
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            raw = json.load(f)
+    except json.JSONDecodeError as e:
+        raise IntegrityError(f"Corrupt JSON file {path}: {e}")
+
+    # Check if it's an envelope
+    if isinstance(raw, dict) and "hash" in raw and "data" in raw and "version" in raw:
+        # Verify integrity
+        stored_hash = raw["hash"]
+        stored_data = raw["data"]
+
+        computed_hash = calculate_hash(stored_data)
+        if computed_hash != stored_hash:
+            raise IntegrityError(f"Integrity check failed for {path} (hash mismatch)")
+
+        # Verify version if needed
+        # We can implement version migration logic here if multiple envelope versions exist
+
+        return stored_data
+
+    # Legacy Fallback
+    if migrate:
+        logger.info(
+            f"R77: Loaded legacy file {path}, integrity check skipped (pending migration)."
+        )
+        # For legacy files, we assume the whole content is the data.
+        return raw
+
+    raise IntegrityError(f"File {path} is not a valid integrity envelope")

services/runtime_config.py

@@ -145,6 +145,12 @@ SCHEDULER_ENV_MAPPINGS = {
     "skip_missed_intervals": ("OPENCLAW_SCHEDULER_SKIP_MISSED", ""),
 }
 
+# IMPORTANT:
+# Keep effective-config merge order deterministic.
+# Using a set iteration here makes legacy warning assertions flaky because the
+# first env key read can vary per process/hash seed.
+LLM_KEY_ORDER = tuple(ENV_MAPPINGS.keys())
+
 
 def _clamp(value: int, min_val: int, max_val: int) -> int:
     """Clamp an integer to a range."""
@@ -271,7 +277,11 @@ def get_effective_config() -> Tuple[Dict[str, Any], Dict[str, str]]:
     effective = {}
     sources = {}
 
-    for key in ALLOWED_LLM_KEYS:
+    ordered_keys = list(LLM_KEY_ORDER) + [
+        k for k in sorted(ALLOWED_LLM_KEYS) if k not in ENV_MAPPINGS
+    ]
+
+    for key in ordered_keys:
         # 1. Check ENV override
         env_val = _get_env_value(key)
         if env_val is not None:

services/transform_common.py

@@ -0,0 +1,348 @@
+"""
+Common types and registry for constrained transforms (S35/F42).
+Refactored to avoid circular imports.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import time
+from dataclasses import asdict, dataclass, field
+from enum import Enum
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set
+
+logger = logging.getLogger("ComfyUI-OpenClaw.services.transform_common")
+
+# ---------------------------------------------------------------------------
+# Feature gate
+# ---------------------------------------------------------------------------
+
+_FEATURE_FLAG = "OPENCLAW_ENABLE_TRANSFORMS"
+
+
+def is_transforms_enabled() -> bool:
+    """Check if constrained transforms are enabled (default: OFF)."""
+    val = os.environ.get(_FEATURE_FLAG, "").strip().lower()
+    return val in ("1", "true", "yes", "on")
+
+
+# ---------------------------------------------------------------------------
+# Runtime limits
+# ---------------------------------------------------------------------------
+
+DEFAULT_TRANSFORM_TIMEOUT_SEC = 5
+DEFAULT_MAX_OUTPUT_BYTES = 64 * 1024  # 64KB
+DEFAULT_MAX_TRANSFORMS_PER_REQUEST = 5
+MAX_TRANSFORM_MODULE_SIZE_BYTES = 50 * 1024  # 50KB — prevent loading huge scripts
+
+
+@dataclass
+class TransformLimits:
+    """Runtime limits for transform execution."""
+
+    timeout_sec: float = DEFAULT_TRANSFORM_TIMEOUT_SEC
+    max_output_bytes: int = DEFAULT_MAX_OUTPUT_BYTES
+    max_transforms_per_request: int = DEFAULT_MAX_TRANSFORMS_PER_REQUEST
+
+    @classmethod
+    def from_env(cls) -> "TransformLimits":
+        """Load limits from environment variables."""
+
+        def _env_int(key: str, default: int) -> int:
+            try:
+                return int(os.environ.get(key, str(default)))
+            except (ValueError, TypeError):
+                return default
+
+        def _env_float(key: str, default: float) -> float:
+            try:
+                return float(os.environ.get(key, str(default)))
+            except (ValueError, TypeError):
+                return default
+
+        return cls(
+            timeout_sec=_env_float(
+                "OPENCLAW_TRANSFORM_TIMEOUT", DEFAULT_TRANSFORM_TIMEOUT_SEC
+            ),
+            max_output_bytes=_env_int(
+                "OPENCLAW_TRANSFORM_MAX_OUTPUT", DEFAULT_MAX_OUTPUT_BYTES
+            ),
+            max_transforms_per_request=_env_int(
+                "OPENCLAW_TRANSFORM_MAX_PER_REQUEST", DEFAULT_MAX_TRANSFORMS_PER_REQUEST
+            ),
+        )
+
+
+# ---------------------------------------------------------------------------
+# Transform result
+# ---------------------------------------------------------------------------
+
+
+class TransformStatus(str, Enum):
+    SUCCESS = "success"
+    ERROR = "error"
+    TIMEOUT = "timeout"
+    DENIED = "denied"
+    SKIPPED = "skipped"
+
+
+@dataclass
+class TransformResult:
+    """Result of a single transform execution."""
+
+    transform_id: str
+    status: str  # TransformStatus.value
+    output: Optional[Dict[str, Any]] = None
+    error: str = ""
+    duration_ms: float = 0.0
+    output_bytes: int = 0
+    audit: Dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> Dict[str, Any]:
+        d: Dict[str, Any] = {
+            "transform_id": self.transform_id,
+            "status": self.status,
+            "duration_ms": round(self.duration_ms, 2),
+            "output_bytes": self.output_bytes,
+        }
+        if self.output is not None:
+            d["output"] = self.output
+        if self.error:
+            d["error"] = self.error
+        if self.audit:
+            d["audit"] = self.audit
+        return d
+
+
+# ---------------------------------------------------------------------------
+# Transform registry (trusted modules)
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class TrustedTransform:
+    """A registered, integrity-pinned transform module."""
+
+    id: str
+    label: str
+    module_path: str  # Absolute path to .py module
+    sha256: str  # Integrity hash of the module file
+    description: str = ""
+    trusted_source: str = ""  # Who published this transform
+    registered_at: float = 0.0
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+
+class TransformRegistryError(Exception):
+    """Error in transform registry operations."""
+
+    pass
+
+
+class TransformRegistry:
+    """
+    Manages trusted transform modules with integrity pinning.
+
+    Transforms can only be loaded from explicitly trusted directories.
+    Each module is pinned by its SHA256 hash at registration time.
+    """
+
+    def __init__(self, state_dir: str, trusted_dirs: Optional[List[str]] = None):
+        self._state_dir = state_dir
+        self._registry_dir = os.path.join(state_dir, "transforms")
+        self._index_path = os.path.join(self._registry_dir, "registry.json")
+        self._transforms: Dict[str, TrustedTransform] = {}
+
+        # Trusted directories where transform modules can live
+        self._trusted_dirs: Set[str] = set()
+        if trusted_dirs:
+            for d in trusted_dirs:
+                resolved = str(Path(d).resolve())
+                self._trusted_dirs.add(resolved)
+
+        os.makedirs(self._registry_dir, exist_ok=True)
+        self._load()
+
+    def _load(self) -> None:
+        """Load transform registry from disk."""
+        if not os.path.exists(self._index_path):
+            self._transforms = {}
+            return
+        try:
+            with open(self._index_path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+            self._transforms = {}
+            for tid, tdata in data.items():
+                self._transforms[tid] = TrustedTransform(**tdata)
+        except Exception as e:
+            logger.error(f"Failed to load transform registry: {e}")
+            self._transforms = {}
+
+    def _save(self) -> None:
+        """Persist transform registry to disk."""
+        try:
+            data = {k: v.to_dict() for k, v in self._transforms.items()}
+            tmp_path = self._index_path + ".tmp"
+            with open(tmp_path, "w", encoding="utf-8") as f:
+                json.dump(data, f, indent=2)
+                f.write("\n")
+            os.replace(tmp_path, self._index_path)
+        except Exception as e:
+            logger.error(f"Failed to save transform registry: {e}")
+
+    @staticmethod
+    def _compute_sha256(file_path: str) -> str:
+        """Compute SHA256 hash of a file."""
+        h = hashlib.sha256()
+        with open(file_path, "rb") as f:
+            for chunk in iter(lambda: f.read(4096), b""):
+                h.update(chunk)
+        return h.hexdigest()
+
+    def _is_in_trusted_dir(self, module_path: str) -> bool:
+        """Check if a module path is inside a trusted directory."""
+        resolved = str(Path(module_path).resolve())
+        for trusted in self._trusted_dirs:
+            if resolved.startswith(trusted + os.sep) or resolved == trusted:
+                return True
+        return False
+
+    def register_transform(
+        self,
+        transform_id: str,
+        module_path: str,
+        *,
+        label: str = "",
+        description: str = "",
+        trusted_source: str = "",
+    ) -> TrustedTransform:
+        """
+        Register a transform module with integrity pinning.
+
+        The module must be in a trusted directory and within size limits.
+        """
+        if not is_transforms_enabled():
+            raise TransformRegistryError(
+                f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable."
+            )
+
+        abs_path = str(Path(module_path).resolve())
+
+        # Security: must be in trusted directory
+        if not self._is_in_trusted_dir(abs_path):
+            raise TransformRegistryError(
+                f"Module path is not in a trusted directory: {abs_path}"
+            )
+
+        if not os.path.isfile(abs_path):
+            raise TransformRegistryError(f"Module file not found: {abs_path}")
+
+        # Size check
+        file_size = os.path.getsize(abs_path)
+        if file_size > MAX_TRANSFORM_MODULE_SIZE_BYTES:
+            raise TransformRegistryError(
+                f"Module exceeds size limit ({file_size} > {MAX_TRANSFORM_MODULE_SIZE_BYTES})"
+            )
+
+        # Must be a .py file
+        if not abs_path.endswith(".py"):
+            raise TransformRegistryError("Only .py modules are allowed as transforms")
+
+        sha256 = self._compute_sha256(abs_path)
+
+        transform = TrustedTransform(
+            id=transform_id,
+            label=label or transform_id,
+            module_path=abs_path,
+            sha256=sha256,
+            description=description,
+            trusted_source=trusted_source,
+            registered_at=time.time(),
+        )
+
+        self._transforms[transform_id] = transform
+        self._save()
+        logger.info(f"F42: Registered transform '{transform_id}' from {abs_path}")
+        return transform
+
+    def unregister_transform(self, transform_id: str) -> bool:
+        """Remove a transform from the registry."""
+        if not is_transforms_enabled():
+            raise TransformRegistryError(
+                f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable."
+            )
+
+        if transform_id not in self._transforms:
+            return False
+
+        del self._transforms[transform_id]
+        self._save()
+        logger.info(f"F42: Unregistered transform '{transform_id}'")
+        return True
+
+    def get_transform(self, transform_id: str) -> Optional[TrustedTransform]:
+        """Get a registered transform by ID."""
+        return self._transforms.get(transform_id)
+
+    def list_transforms(self) -> List[TrustedTransform]:
+        """List all registered transforms."""
+        return list(self._transforms.values())
+
+    def verify_integrity(self, transform_id: str) -> bool:
+        """Verify that a registered transform's file hasn't been modified."""
+        transform = self._transforms.get(transform_id)
+        if not transform:
+            return False
+
+        if not os.path.isfile(transform.module_path):
+            return False
+
+        actual_hash = self._compute_sha256(transform.module_path)
+        return actual_hash == transform.sha256
+
+
+# ---------------------------------------------------------------------------
+# Module-level convenience
+# ---------------------------------------------------------------------------
+
+_registry: Optional[TransformRegistry] = None
+
+
+def get_transform_registry() -> TransformRegistry:
+    """Get or create the global transform registry."""
+    global _registry
+    if _registry is None:
+        try:
+            from .state_dir import get_state_dir
+
+            state_dir = get_state_dir()
+        except ImportError:
+            try:
+                from services.state_dir import get_state_dir
+
+                state_dir = get_state_dir()
+            except ImportError:
+                state_dir = os.path.join(
+                    os.path.dirname(os.path.dirname(__file__)), "data"
+                )
+
+        # Default trusted directory: pack-local transforms dir
+        pack_root = Path(__file__).resolve().parent.parent
+        trusted_dirs = [str(pack_root / "data" / "transforms")]
+
+        # Allow additional trusted dirs from env
+        extra = os.environ.get("OPENCLAW_TRANSFORM_TRUSTED_DIRS", "")
+        if extra:
+            for d in extra.split(os.pathsep):
+                d = d.strip()
+                if d:
+                    trusted_dirs.append(d)
+
+        _registry = TransformRegistry(state_dir, trusted_dirs=trusted_dirs)
+    return _registry

services/transform_runner.py

@@ -0,0 +1,183 @@
+"""
+S35 Transform Isolation Runner.
+
+Executes transforms in a separate process via `services.transform_worker`.
+"""
+
+import json
+import logging
+import os
+import subprocess
+import sys
+import time
+from typing import Any, Dict, Optional
+
+from .transform_common import (
+    TransformLimits,
+    TransformRegistry,
+    TransformResult,
+    TransformStatus,
+)
+
+logger = logging.getLogger("ComfyUI-OpenClaw.services.transform_runner")
+
+
+class TransformProcessRunner:
+    """
+    Executes transforms in an isolated subprocess.
+    """
+
+    def __init__(
+        self,
+        registry: TransformRegistry,
+        limits: Optional[TransformLimits] = None,
+    ):
+        self._registry = registry
+        self._limits = limits or TransformLimits.from_env()
+
+    def execute_transform(
+        self,
+        transform_id: str,
+        input_data: Dict[str, Any],
+        *,
+        trace_id: str = "",
+    ) -> TransformResult:
+        """
+        Execute a single registered transform in a subprocess.
+        """
+        transform = self._registry.get_transform(transform_id)
+        if not transform:
+            return TransformResult(
+                transform_id=transform_id,
+                status=TransformStatus.ERROR.value,
+                error=f"Transform '{transform_id}' not found in registry",
+            )
+
+        # Integrity Check (R77 pre-check)
+        if not self._registry.verify_integrity(transform_id):
+            return TransformResult(
+                transform_id=transform_id,
+                status=TransformStatus.DENIED.value,
+                error="Integrity verification failed — module modified",
+                audit={"reason": "integrity_check_failed", "trace_id": trace_id},
+            )
+
+        start_time = time.monotonic()
+
+        # Prepare Worker Command
+        worker_script = os.path.join(os.path.dirname(__file__), "transform_worker.py")
+        cmd = [sys.executable, worker_script, transform.module_path]
+
+        # Prepare Input
+        payload = {"input": input_data, "context": {"trace_id": trace_id}}
+        input_json = json.dumps(payload)
+
+        try:
+            # capability-deny: no environment variable inheritance by default?
+            # Or minimal env.
+            env = os.environ.copy()
+            # Remove sensitive vars if needed?
+            # S34 Obs: redact env in logs, but here process sees env.
+            # Best practice: clear sensitive vars.
+            for key in list(env.keys()):
+                if "TOKEN" in key or "SECRET" in key or "KEY" in key:
+                    del env[key]
+
+            # Subprocess Run
+            proc = subprocess.run(
+                cmd,
+                input=input_json,
+                capture_output=True,
+                text=True,
+                timeout=self._limits.timeout_sec,
+                env=env,
+                check=False,  # We handle return codes
+            )
+
+            # Execution Time
+            elapsed_ms = (time.monotonic() - start_time) * 1000
+
+            # Handle Return Code
+            if proc.returncode != 0:
+                # Script crashed or printed error to stdout/stderr
+                # Try to parse stdout error first
+                error_msg = proc.stderr.strip() or "Process crashed with unknown error"
+                try:
+                    out_json = json.loads(proc.stdout)
+                    if out_json.get("status") == "error":
+                        error_msg = out_json.get("error", error_msg)
+                except Exception:
+                    pass
+
+                return TransformResult(
+                    transform_id=transform_id,
+                    status=TransformStatus.ERROR.value,
+                    error=f"Worker process failed (exit {proc.returncode}): {error_msg}",
+                    duration_ms=elapsed_ms,
+                    audit={"exit_code": proc.returncode, "trace_id": trace_id},
+                )
+
+            # Parse Output
+            try:
+                result_json = json.loads(proc.stdout)
+            except json.JSONDecodeError:
+                return TransformResult(
+                    transform_id=transform_id,
+                    status=TransformStatus.ERROR.value,
+                    error="Worker returned invalid JSON output",
+                    duration_ms=elapsed_ms,
+                    audit={"raw_stdout": proc.stdout[:1000], "trace_id": trace_id},
+                )
+
+            if result_json.get("status") == "error":
+                return TransformResult(
+                    transform_id=transform_id,
+                    status=TransformStatus.ERROR.value,
+                    error=result_json.get("error", "Unknown worker error"),
+                    duration_ms=elapsed_ms,
+                    audit={
+                        "trace_id": trace_id,
+                        "traceback": result_json.get("traceback"),
+                    },
+                )
+
+            output_data = result_json.get("output", {})
+            output_bytes = len(json.dumps(output_data).encode("utf-8"))
+
+            if output_bytes > self._limits.max_output_bytes:
+                return TransformResult(
+                    transform_id=transform_id,
+                    status=TransformStatus.ERROR.value,
+                    error=f"Output size limit exceeded ({output_bytes} > {self._limits.max_output_bytes})",
+                    duration_ms=elapsed_ms,
+                    output_bytes=output_bytes,
+                    audit={"trace_id": trace_id},
+                )
+
+            return TransformResult(
+                transform_id=transform_id,
+                status=TransformStatus.SUCCESS.value,
+                output=output_data,
+                duration_ms=elapsed_ms,
+                output_bytes=output_bytes,
+                audit={"trace_id": trace_id, "isolation": "process"},
+            )
+
+        except subprocess.TimeoutExpired:
+            elapsed_ms = (time.monotonic() - start_time) * 1000
+            return TransformResult(
+                transform_id=transform_id,
+                status=TransformStatus.TIMEOUT.value,
+                error=f"Transform timeout exceeded ({self._limits.timeout_sec}s)",
+                duration_ms=elapsed_ms,
+                audit={"trace_id": trace_id, "timeout": True},
+            )
+        except Exception as e:
+            elapsed_ms = (time.monotonic() - start_time) * 1000
+            return TransformResult(
+                transform_id=transform_id,
+                status=TransformStatus.ERROR.value,
+                error=f"Runner exception: {str(e)}",
+                duration_ms=elapsed_ms,
+                audit={"trace_id": trace_id},
+            )

services/transform_worker.py

@@ -0,0 +1,102 @@
+"""
+S35 Transform Isolation Worker.
+
+This script runs in a separate process to execute a transform module.
+It reads input JSON from stdin and writes output JSON to stdout.
+
+Usage:
+    python -m services.transform_worker <module_path>
+
+Protocol:
+    Input (stdin): JSON object {"input": {...}, "context": {...}}
+    Output (stdout): JSON object {"status": "success", "output": {...}} or {"status": "error", "error": "..."}
+    Exit Code: 0 on success/handled error, non-zero on crash.
+"""
+
+import argparse
+import importlib.util
+import json
+import os
+import socket
+import sys
+import traceback
+from typing import Any, Dict
+
+# S35: Capability Deny-by-Default
+# We can't easily drop OS privileges in a cross-platform way without deps,
+# but being in a separate process isolates memory and crashes.
+# We could monkeypatch network libs here to enforce "no network".
+
+
+def _deny_network(*args, **kwargs):
+    raise RuntimeError("Network access denied by S35 transform isolation policy")
+
+
+# Monkeypatch socket to deny network access
+socket.socket = _deny_network
+socket.create_connection = _deny_network
+# TODO: Monkeypatch http.client, urllib, requests if present?
+# Standard library socket blocks most.
+
+
+def load_module(module_path: str):
+    """Load the transform module from path."""
+    if not os.path.exists(module_path):
+        raise FileNotFoundError(f"Module not found: {module_path}")
+
+    spec = importlib.util.spec_from_file_location("transform_module", module_path)
+    if not spec or not spec.loader:
+        raise ImportError(f"Could not load spec for {module_path}")
+
+    module = importlib.util.module_from_spec(spec)
+    sys.modules["transform_module"] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("module_path", help="Absolute path to the transform module")
+    args = parser.parse_args()
+
+    # Read input payload
+    try:
+        input_raw = sys.stdin.read()
+        if not input_raw:
+            raise ValueError("Empty input on stdin")
+        payload = json.loads(input_raw)
+        data = payload.get("input", {})
+    except Exception as e:
+        response = {"status": "error", "error": f"Failed to read input: {e}"}
+        print(json.dumps(response))
+        sys.exit(1)
+
+    # Execute transform
+    try:
+        module = load_module(args.module_path)
+
+        if not hasattr(module, "transform"):
+            raise AttributeError("Module missing 'transform' function")
+
+        func = getattr(module, "transform")
+        if not callable(func):
+            raise TypeError("'transform' is not callable")
+
+        # Run
+        result = func(data)
+
+        # Validate result (JSON serializable?)
+        # We try to dump it. If it fails, that's an error.
+        output_payload = {"status": "success", "output": result}
+        print(json.dumps(output_payload, default=str))
+
+    except Exception as e:
+        # Capture traceback for diagnostics
+        tb = traceback.format_exc()
+        response = {"status": "error", "error": str(e), "traceback": tb}
+        print(json.dumps(response))
+        sys.exit(0)  # Exit 0 because we handled the error gracefully
+
+
+if __name__ == "__main__":
+    main()

tests/TEST_SOP.md

@@ -11,6 +11,7 @@ Every implementation plan must include the **full test validation procedure** in
 - Python 3.10+ (CI uses 3.10/3.11)
 - Node.js 18+ (CI uses 20)
 - `pre-commit` installed: `python -m pip install pre-commit`
+- Backend test deps available in the same interpreter (`numpy`, `pillow`, `aiohttp`)
 - Frontend deps installed: `npm install`
 
 ## Environment Sanity (Required Guardrails)
@@ -18,11 +19,13 @@ Every implementation plan must include the **full test validation procedure** in
 - **Python interpreter must be consistent** for all test commands.
   - Verify: `python -c "import sys; print(sys.executable)"`
   - If you use conda or venv, ensure the same interpreter runs unit tests and connector tests.
-- **Project venv recommended**: use `.venv` when possible to avoid mixed dependencies.
-  - Create: `python -m venv .venv`
-  - Activate (bash): `source .venv/bin/activate`
+- **Project venv recommended**: use an OS-specific local venv to avoid mixed dependencies.
+  - Linux/WSL recommended path: `.venv-wsl` (especially when Windows also uses `.venv` in the same repo)
+  - Other environments: `.venv`
+  - Create: `python -m venv .venv-wsl` (WSL) or `python -m venv .venv`
+  - Activate (bash): `source .venv-wsl/bin/activate` (or `.venv/bin/activate`)
   - Activate (pwsh): `.\.venv\Scripts\Activate.ps1`
-  - If tests fail due to missing deps in CI parity, **rerun in `.venv` and record that in the implementation record**.
+  - If tests fail due to missing deps in CI parity, rerun in the project venv used by scripts and record that in the implementation record.
 - **Node version must be 18+** before E2E:
   - Verify: `node -v`
   - If mismatch in WSL, use the Node 18 path specified below.
@@ -121,8 +124,9 @@ Use these checks before assuming the hook runner is broken:
 ### Optional: One-Command Full Test Scripts (Fastest)
 
 Use these if you want a single command that runs **all required steps** (detect-secrets, pre-commit, unit tests, E2E). These scripts also handle the most common environment issues (Windows cache locks, Black cache, Node 18).
-Both scripts enforce a project-local `.venv` and will bootstrap missing test tooling (`pre-commit`, and `aiohttp` where needed for imports).
-If `.venv` exists but is invalid for the current OS (for example created in WSL then reused in Windows), rerun via the script so it can recreate the environment.
+Scripts enforce a project-local venv and will bootstrap missing test tooling (`pre-commit`, and `aiohttp` where needed for imports).
+On WSL, scripts prefer `.venv-wsl`; on Windows they use `.venv`.
+If the selected venv exists but is invalid for the current OS/interpreter, rerun via the script so it can recreate that venv.
 Linux script includes an explicit offline fail-fast guard: if dependency bootstrap fails (for example `aiohttp` / `pre-commit` install), it stops with remediation hints instead of continuing with partial state.
 
 - Linux/WSL:
@@ -144,6 +148,16 @@ Then every `git push` will run:
 bash scripts/pre_push_checks.sh
 ```
 
+`scripts/pre_push_checks.sh` is the CI-parity guard and must include all 4 stages:
+1) `detect-secrets`
+2) all `pre-commit` hooks
+3) backend unit tests (`scripts/run_unittests.py --pattern "test_*.py"`)
+4) frontend E2E (`npm test`)
+
+IMPORTANT:
+- Do not remove stage (3). If pre-push skips backend unit tests, local pushes can pass while GitHub CI fails later.
+- Keep dependency bootstrap in this script aligned with `.github/workflows/ci.yml` unit-test dependencies.
+
 1) Detect Secrets (baseline-based)
 
 ```bash

tests/test_r77_integrity.py

@@ -0,0 +1,107 @@
+"""
+R77 Integrity Tests.
+"""
+
+import json
+import os
+import shutil
+import tempfile
+import unittest
+
+from services.integrity import (
+    IntegrityEnvelope,
+    IntegrityError,
+    calculate_hash,
+    canonical_dumps,
+    load_verified,
+    save_verified,
+)
+
+
+class TestR77Integrity(unittest.TestCase):
+    def setUp(self):
+        self.test_dir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        shutil.rmtree(self.test_dir)
+
+    def test_canonical_serialization(self):
+        """Test that JSON is canonicalized (sorted keys, no whitespace)."""
+        data1 = {"b": 2, "a": 1}
+        data2 = {"a": 1, "b": 2}
+
+        c1 = canonical_dumps(data1)
+        c2 = canonical_dumps(data2)
+
+        self.assertEqual(c1, c2)
+        # Check format: {"a":1,"b":2}
+        self.assertEqual(c1, b'{"a":1,"b":2}')
+
+    def test_hashing(self):
+        """Test hash consistency."""
+        data = {"foo": "bar"}
+        h1 = calculate_hash(data)
+        h2 = calculate_hash(data)
+        self.assertEqual(h1, h2)
+        self.assertTrue(len(h1) == 64)  # SHA256 hex digest
+
+    def test_save_and_load_verified(self):
+        """Test saving and loading with integrity envelope."""
+        path = os.path.join(self.test_dir, "test.json")
+        data = {"key": "value", "list": [1, 2, 3]}
+
+        save_verified(path, data)
+
+        # Verify file structure on disk
+        with open(path, "r") as f:
+            envelope = json.load(f)
+
+        self.assertIn("version", envelope)
+        self.assertIn("hash", envelope)
+        self.assertIn("data", envelope)
+        self.assertEqual(envelope["data"], data)
+
+        # Load back
+        loaded = load_verified(path)
+        self.assertEqual(loaded, data)
+
+    def test_tamper_detection(self):
+        """Test that modification of data voids the integrity check."""
+        path = os.path.join(self.test_dir, "tampered.json")
+        data = {"secret": "123"}
+        save_verified(path, data)
+
+        # Tamper with the file
+        with open(path, "r") as f:
+            envelope = json.load(f)
+
+        envelope["data"]["secret"] = "666"  # Evil modification
+        # Hash is NOT updated
+
+        with open(path, "w") as f:
+            json.dump(envelope, f)
+
+        # Load should fail
+        with self.assertRaises(IntegrityError):
+            load_verified(path)
+
+    def test_legacy_migration(self):
+        """Test verifying legacy (non-envelope) files."""
+        path = os.path.join(self.test_dir, "legacy.json")
+        data = {"old": "data"}
+
+        # Save as raw JSON
+        with open(path, "w") as f:
+            json.dump(data, f)
+
+        # load_verified with migrate=True (default) should succeed
+        loaded = load_verified(path, migrate=True)
+        self.assertEqual(loaded, data)
+
+        # load_verified with migrate=False should fail
+        with self.assertRaises(IntegrityError):
+            load_verified(path, migrate=False)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_runtime_config.py

@@ -43,6 +43,20 @@ class TestRuntimeConfig(unittest.TestCase):
         patcher.start()
         self.addCleanup(patcher.stop)
 
+        # IMPORTANT:
+        # runtime_config._get_env_value caches legacy-warning emission in a function
+        # attribute (`_warned_legacy`). Without resetting it here, test order can
+        # suppress expected warning logs and produce false negatives in CI.
+        try:
+            for mod_name, mod in list(sys.modules.items()):
+                if not mod_name.endswith("runtime_config"):
+                    continue
+                fn = getattr(mod, "_get_env_value", None)
+                if callable(fn):
+                    setattr(fn, "_warned_legacy", set())
+        except Exception:
+            pass
+
         for key in [
             "MOLTBOT_LLM_PROVIDER",
             "MOLTBOT_LLM_MODEL",

tests/test_s33r76r78_auth_hardening.py

@@ -25,9 +25,16 @@ from services.csrf_protection import is_same_origin_request
 # classified here, otherwise tests fail.
 AUTH_CLASS_BY_ROUTE = {
     ("GET", "/health"): "public-safe",
-    ("GET", "/logs/tail"): "observability",
+    # IMPORTANT:
+    # `/logs/tail` was hardened to admin-only because log payload can expose
+    # high-sensitivity prompt/runtime context (S34). Keep this auth class in
+    # sync with api/logs_tail.py to avoid accidental privilege regression.
+    ("GET", "/logs/tail"): "admin",
     ("GET", "/jobs"): "public-safe",
-    ("GET", "/trace/{prompt_id}"): "observability",
+    # IMPORTANT:
+    # Trace endpoint now returns high-sensitivity execution context and is
+    # intentionally admin-only (S34). Keep as admin to prevent data leakage.
+    ("GET", "/trace/{prompt_id}"): "admin",
     ("POST", "/webhook"): "webhook-auth",
     ("POST", "/webhook/submit"): "webhook-auth",
     ("POST", "/webhook/validate"): "webhook-auth",

tests/test_s34_observability_tiers.py

@@ -0,0 +1,79 @@
+"""
+S34 Observability Tiers Tests.
+"""
+
+import os
+import unittest
+from unittest.mock import MagicMock, patch
+
+from services.access_control import require_admin_token, require_observability_access
+
+
+# Mock helpers
+class MockRequest:
+    def __init__(self, headers=None, remote="1.2.3.4"):
+        self.headers = headers or {}
+        self.remote = remote
+        self.match_info = {}
+        self.query = {}
+
+
+class TestS34ObservabilityTiers(unittest.TestCase):
+    def setUp(self):
+        self.obs_token = "obs-secret"
+        self.admin_token = "admin-secret"
+        os.environ["OPENCLAW_OBSERVABILITY_TOKEN"] = self.obs_token
+        os.environ["OPENCLAW_ADMIN_TOKEN"] = self.admin_token
+
+    def tearDown(self):
+        if "OPENCLAW_OBSERVABILITY_TOKEN" in os.environ:
+            del os.environ["OPENCLAW_OBSERVABILITY_TOKEN"]
+        if "OPENCLAW_ADMIN_TOKEN" in os.environ:
+            del os.environ["OPENCLAW_ADMIN_TOKEN"]
+
+    def test_low_sensitivity_access(self):
+        """Health/Config should allow Obs Token."""
+        req = MockRequest(headers={"X-OpenClaw-Obs-Token": self.obs_token})
+        allowed, _ = require_observability_access(req)
+        self.assertTrue(allowed, "Obs token should allow low sensitivity access")
+
+    def test_high_sensitivity_denial(self):
+        """Trace/Log should DENY Obs Token (require Admin)."""
+        # Admin check with ONLY obs token should fail
+        req = MockRequest(headers={"X-OpenClaw-Obs-Token": self.obs_token})
+        allowed, _ = require_admin_token(req)
+        self.assertFalse(allowed, "Obs token should NOT pass Admin check")
+
+    def test_high_sensitivity_allow(self):
+        """Trace/Log should ALLOW Admin Token."""
+        req = MockRequest(headers={"X-Moltbot-Admin-Token": self.admin_token})
+        allowed, _ = require_admin_token(req)
+        self.assertTrue(allowed, "Admin token should pass Admin check")
+
+    @patch("api.routes.trace_store")
+    @patch("api.routes.require_admin_token")
+    @patch("api.routes.web")
+    def test_trace_handler_tier_enforcement(
+        self, mock_web, mock_require_admin, mock_trace_store
+    ):
+        """Verify API handler calls the right check."""
+        import asyncio
+
+        from api.routes import trace_handler
+
+        # Setup
+        mock_require_admin.return_value = (False, "Denied")
+        req = MockRequest()
+
+        async def run_test():
+            await trace_handler(req)
+
+        # Execute
+        asyncio.run(run_test())
+
+        # Verify
+        mock_require_admin.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_s35_transform_isolation.py

@@ -0,0 +1,135 @@
+"""
+S35 Transform Isolation Tests.
+"""
+
+import json
+import os
+import sys
+import unittest
+from unittest.mock import MagicMock, patch
+
+# Add project root to path
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from services.transform_common import (
+    TransformLimits,
+    TransformRegistry,
+    TransformStatus,
+    TrustedTransform,
+)
+from services.transform_runner import TransformProcessRunner
+
+
+class TestS35TransformIsolation(unittest.TestCase):
+
+    def setUp(self):
+        self.registry = MagicMock(spec=TransformRegistry)
+        self.limits = TransformLimits(
+            timeout_sec=2.0, max_output_bytes=1024, max_transforms_per_request=1
+        )
+        self.runner = TransformProcessRunner(self.registry, self.limits)
+
+    def test_process_execution_success(self):
+        """Test successful execution in a subprocess."""
+        # Create a dummy transform module on disk
+        import tempfile
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
+            f.write("def transform(data):\n    return {'echo': data['input']}")
+            module_path = f.name
+
+        try:
+            # Mock registry to return this module
+            self.registry.get_transform.return_value = TrustedTransform(
+                id="test_echo",
+                label="Echo",
+                module_path=module_path,
+                sha256="dummy_hash",
+            )
+            self.registry.verify_integrity.return_value = True
+
+            result = self.runner.execute_transform(
+                "test_echo", {"input": "hello"}, trace_id="test_s35"
+            )
+
+            self.assertEqual(result.status, TransformStatus.SUCCESS.value)
+            self.assertEqual(result.output, {"echo": "hello"})
+            # Ensure it ran in a process? Hard to prove from here without spying on subprocess.
+            # But the runner uses subprocess.run.
+        finally:
+            if os.path.exists(module_path):
+                os.remove(module_path)
+
+    def test_timeout_enforcement(self):
+        """Test that slow transforms are killed."""
+        import tempfile
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
+            f.write(
+                "import time\ndef transform(data):\n    time.sleep(5)\n    return {}"
+            )
+            module_path = f.name
+
+        try:
+            self.registry.get_transform.return_value = TrustedTransform(
+                id="test_slow",
+                label="Slow",
+                module_path=module_path,
+                sha256="dummy_hash",
+            )
+            self.registry.verify_integrity.return_value = True
+
+            result = self.runner.execute_transform(
+                "test_slow", {}, trace_id="test_timeout"
+            )
+
+            self.assertEqual(result.status, TransformStatus.TIMEOUT.value)
+            self.assertIn("timeout exceeded", result.error)
+        finally:
+            if os.path.exists(module_path):
+                os.remove(module_path)
+
+    def test_capability_denial_network(self):
+        """Test that network access is denied (by monkeypatch in worker)."""
+        import tempfile
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
+            f.write(
+                """
+import socket
+def transform(data):
+    try:
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.connect(('example.com', 80))
+        return {'status': 'connected'}
+    except Exception as e:
+        return {'error': str(e)}
+"""
+            )
+            module_path = f.name
+
+        try:
+            self.registry.get_transform.return_value = TrustedTransform(
+                id="test_net", label="Net", module_path=module_path, sha256="dummy_hash"
+            )
+            self.registry.verify_integrity.return_value = True
+
+            result = self.runner.execute_transform(
+                "test_net", {}, trace_id="test_net_deny"
+            )
+
+            # The worker monkeypatches socket to raise RuntimeError
+            # So the transform returns {'error': ...}
+            # Or crashes if it didn't catch it.
+            # The script above catches it and returns it.
+
+            output = result.output or {}
+            self.assertEqual(result.status, TransformStatus.SUCCESS.value)
+            self.assertIn("Network access denied", output.get("error", ""))
+        finally:
+            if os.path.exists(module_path):
+                os.remove(module_path)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_s36s37r79_egress_hardening.py

@@ -19,8 +19,27 @@ from unittest.mock import ANY, MagicMock, patch
 import services.safe_io as safe_io
 import services.webhook_auth as webhook_auth
 
-# Disable logging during tests to avoid noise
-logging.disable(logging.CRITICAL)
+# IMPORTANT:
+# Do NOT disable logging at import time. unittest discovery imports all modules
+# before test execution, so import-time logging.disable() leaks globally and
+# breaks unrelated assertLogs tests.
+_PREV_LOG_DISABLE_LEVEL = None
+
+
+def setUpModule():
+    """Disable noisy logs only while this module's tests are running."""
+    global _PREV_LOG_DISABLE_LEVEL
+    _PREV_LOG_DISABLE_LEVEL = logging.root.manager.disable
+    logging.disable(logging.CRITICAL)
+
+
+def tearDownModule():
+    """Restore global logging state for downstream modules."""
+    global _PREV_LOG_DISABLE_LEVEL
+    if _PREV_LOG_DISABLE_LEVEL is None:
+        logging.disable(logging.NOTSET)
+    else:
+        logging.disable(_PREV_LOG_DISABLE_LEVEL)
 
 
 class TestS36WebhookReplay(unittest.TestCase):