complete wave e bundle c: deliver signed policy posture, security telemetry, deterministic fuzz harness, mutation baseline, and sop-validated closeout

README.md

@@ -16,23 +16,24 @@ This project is intentionally **not** a general-purpose assistant platform with
 
 **Security stance (how this project differs from convenience-first automation packs):**
 
-- Localhost-first defaults; remote access is opt-in
 - Control Plane Split is enforced for public posture: high-risk control surfaces are externalized while embedded UI stays on safe UX/read paths
+- Profile-driven startup hardening with fail-closed enforcement in hardened mode
+- Localhost-first defaults; remote access is opt-in
 - Explicit **Admin Token** boundary for write actions
 - Webhooks are **deny-by-default** until auth is configured
-- Profile-driven startup hardening with fail-closed enforcement in hardened mode
-- Startup module capability gates (disabled modules do not register routes/workers)
 - Encrypted webhook mode is **fail-closed** (invalid signature/decrypt/app-id checks are rejected)
-- Endpoint inventory metadata and route drift tests to catch unclassified API exposure regressions
-- Tamper-evident, append-only audit trails for sensitive write/admin paths
 - Strict outbound SSRF policy (callbacks + custom LLM base URLs)
 - Bridge worker endpoints enforce device-token auth, scope checks, and idempotency handling
-- Replay risk is reduced with deterministic dedupe keys for event payloads without message IDs
+- Startup module capability gates (disabled modules do not register routes/workers)
+- Endpoint inventory metadata and route drift tests to catch unclassified API exposure regressions
+- Tamper-evident, append-only audit trails for sensitive write/admin paths
 - Hardened external tool sandbox posture with fail-closed checks and filesystem path guards
 - Pack lifecycle file paths and pack API inputs are validated and root-bounded to prevent path traversal
+- Replay risk is reduced with deterministic dedupe keys for event payloads without message IDs
+- Wave E closeout hardening: deployment profile gates and critical flow parity are now enforced together with signed policy posture control, bounded anomaly telemetry, adversarial fuzz validation, and mutation-baseline regression sensitivity checks
+- Wave A/B/C closeout hardening: runtime/config/session stability contracts, strict outbound and supply-chain controls, and capability-aware operator guidance with bounded Parameter Lab/compare workflows
 - Secrets are never stored in browser storage (optional server-side key store is local-only convenience)
 - Cryptography dependency is optional and only required when encrypted webhook mode is enabled
-- Wave A/B/C closeout hardening: runtime/config/session stability contracts, strict outbound and supply-chain controls, and capability-aware operator guidance with bounded Parameter Lab/compare workflows
 
 Deployment profiles and hardening checklists:
 - [Security Deployment Guide](docs/security_deployment_guide.md) (local / LAN / public templates + self-check command)
@@ -41,6 +42,18 @@ Deployment profiles and hardening checklists:
 
 <details>
 
+<summary><strong>Wave E closeout: deployment guardrails, contract parity, and verification hardening chain completed</strong></summary>
+
+- Completed Wave E on 2026-02-18 with full SOP validation:
+  - Bundle A delivered startup deployment gate enforcement and deployment-profile matrix parity, then locked critical operator flow parity (including degraded-path behavior)
+  - Bundle B closed security contract parity gaps across token/mapping/route/signature state matrices and threat-intel resilience paths
+  - Bundle C completed signed policy posture control, bounded security anomaly telemetry, deterministic adversarial fuzz harness coverage, and mutation-baseline evidence generation
+  - full detect-secrets + pre-commit + backend unit + frontend E2E gate passed and evidence is recorded in the Bundle C implementation record
+
+</details>
+
+<details>
+
 <summary><strong>Wave D closeout: control-plane split, ingress and supply-chain hardening, and verification governance baseline</strong></summary>
 
 - Completed Wave D closeout on 2026-02-18 with full SOP validation:

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "comfyui-openclaw"
 description = "Your own personal AIGC Factory. Any picture. Any reel. The Comfy way.©️"
-version = "0.4.0"
+version = "0.5.0"
 license = {text = "MIT"}
 readme = "README.md"
 requires-python = ">=3.10"

scripts/run_mutation_test.py

@@ -0,0 +1,186 @@
+import ast
+import json
+import logging
+import os
+import shutil
+import subprocess
+import sys
+import time
+from typing import Any, Dict, List, Optional, Tuple, Union
+
+# Config
+TARGET_FILES = ["services/access_control.py"]
+# IMPORTANT: keep this on the repo unittest runner for .venv/CI parity.
+TEST_COMMAND = [
+    sys.executable,
+    "scripts/run_unittests.py",
+    "--start-dir",
+    "tests",
+    "--pattern",
+    "test_access_control.py",
+]
+
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("mutation_test")
+
+
+class MutationVisitor(ast.NodeTransformer):
+    def __init__(self):
+        self.mutations = []
+        self.current_mutation_index = -1
+        self.mutation_counter = 0
+        self.applied_desc = None
+
+    def visit_Compare(self, node):
+        # Mutate '==' to '!=' and vice versa
+        if len(node.ops) == 1:
+            op = node.ops[0]
+            if isinstance(op, ast.Eq):
+                self._maybe_mutate(
+                    node, lambda n: [setattr(n, "ops", [ast.NotEq()])], "Eq -> NotEq"
+                )
+            elif isinstance(op, ast.NotEq):
+                self._maybe_mutate(
+                    node, lambda n: [setattr(n, "ops", [ast.Eq()])], "NotEq -> Eq"
+                )
+        return self.generic_visit(node)
+
+    def visit_BoolOp(self, node):
+        # Mutate 'and' to 'or' and vice versa
+        op = node.op
+        if isinstance(op, ast.And):
+            self._maybe_mutate(node, lambda n: setattr(n, "op", ast.Or()), "And -> Or")
+        elif isinstance(op, ast.Or):
+            self._maybe_mutate(node, lambda n: setattr(n, "op", ast.And()), "Or -> And")
+        return self.generic_visit(node)
+
+    def _maybe_mutate(self, node, action, desc):
+        if self.current_mutation_index == self.mutation_counter:
+            logger.debug(f"Applying mutation {self.mutation_counter}: {desc}")
+            action(node)
+            self.applied_desc = desc
+        self.mutation_counter += 1
+
+
+def count_mutations(file_path: str) -> int:
+    with open(file_path, "r", encoding="utf-8") as f:
+        tree = ast.parse(f.read())
+
+    visitor = MutationVisitor()
+    visitor.visit(tree)
+    return visitor.mutation_counter
+
+
+def apply_mutation(file_path: str, mutation_index: int) -> str:
+    with open(file_path, "r", encoding="utf-8") as f:
+        tree = ast.parse(f.read())
+
+    visitor = MutationVisitor()
+    visitor.current_mutation_index = mutation_index
+    visitor.visit(tree)
+
+    # Write back
+    with open(file_path, "w", encoding="utf-8") as f:
+        f.write(ast.unparse(tree))
+
+    return getattr(visitor, "applied_desc", "Unknown")
+
+
+def run_test_suite() -> bool:
+    try:
+        result = subprocess.run(
+            TEST_COMMAND, capture_output=True, text=True, timeout=30
+        )
+        return result.returncode == 0
+    except subprocess.TimeoutExpired:
+        return False
+    except Exception as e:
+        logger.error(f"Test run failed: {e}")
+        return False
+
+
+def run_mutation_workflow():
+    report = {"total_mutants": 0, "killed": 0, "survived": 0, "details": []}
+
+    for target in TARGET_FILES:
+        target_path = os.path.abspath(target)
+        backup_path = target_path + ".bak"
+
+        if not os.path.exists(target_path):
+            logger.error(f"Target file not found: {target}")
+            continue
+
+        logger.info(f"Targeting {target}...")
+
+        # 1. Check baseline
+        # shutil.copy2(target_path, backup_path) # Backup first just in case
+        # But wait, run_test_suite runs on current state.
+
+        logger.info("Running baseline tests...")
+        if not run_test_suite():
+            logger.error("Baseline tests failed! Cannot proceed with mutation testing.")
+            return
+
+        # 2. Count mutations
+        num_mutations = count_mutations(target_path)
+        logger.info(f"Found {num_mutations} mutation points in {target}")
+        report["total_mutants"] += num_mutations
+
+        # 3. Apply mutations one by one
+        try:
+            shutil.copy2(target_path, backup_path)
+
+            for i in range(num_mutations):
+                # Restore clean
+                shutil.copy2(backup_path, target_path)
+
+                # Apply mutation
+                desc = apply_mutation(target_path, i)
+                logger.info(f"Mutant {i+1}/{num_mutations}: {desc}")
+
+                # Run tests
+                passed = run_test_suite()
+
+                if not passed:
+                    logger.info("-> KILLED")
+                    report["killed"] += 1
+                else:
+                    logger.warning("-> SURVIVED")
+                    report["survived"] += 1
+                    report["details"].append(
+                        {
+                            "file": target,
+                            "mutation_index": i,
+                            "description": desc,
+                            "status": "SURVIVED",
+                        }
+                    )
+        finally:
+            # Restore original
+            if os.path.exists(backup_path):
+                shutil.copy2(backup_path, target_path)
+                os.remove(backup_path)
+                logger.info("Restored original file.")
+
+    # Generate Report
+    score = 0
+    if report["total_mutants"] > 0:
+        score = (report["killed"] / report["total_mutants"]) * 100
+
+    logger.info("=" * 40)
+    logger.info(f"Mutation Score: {score:.2f}%")
+    logger.info(
+        f"Total: {report['total_mutants']}, Killed: {report['killed']}, Survived: {report['survived']}"
+    )
+    logger.info("=" * 40)
+
+    report_file = os.path.join(".planning", "mutation_report.json")
+    with open(report_file, "w") as f:
+        json.dump(report, f, indent=2)
+    logger.info(f"Report saved to {report_file}")
+
+
+if __name__ == "__main__":
+    run_mutation_workflow()

services/access_control.py

@@ -264,6 +264,13 @@ def verify_tier_access(request, required_tier: AuthTier) -> Tuple[bool, Optional
         if is_loopback(remote):
             return True, None
 
+        # R102 Hook
+        try:
+            from .security_telemetry import get_security_telemetry
+
+            get_security_telemetry().record_auth_failure(remote)
+        except ImportError:
+            pass
         return False, "Internal (Loopback) access required."
 
     # Admin is allowed everything else
@@ -272,6 +279,14 @@ def verify_tier_access(request, required_tier: AuthTier) -> Tuple[bool, Optional
 
     if required_tier == AuthTier.ADMIN:
         # Admin required. Current is not Admin (checked above).
+        # R102 Hook
+        try:
+            from .security_telemetry import get_security_telemetry
+
+            remote = get_client_ip(request)
+            get_security_telemetry().record_auth_failure(remote)
+        except ImportError:
+            pass
         return False, "Admin access required."
 
     if required_tier == AuthTier.OBSERVABILITY:
@@ -287,8 +302,24 @@ def verify_tier_access(request, required_tier: AuthTier) -> Tuple[bool, Optional
         if current_tier == AuthTier.INTERNAL:
             return True, None
 
+        # R102 Hook
+        try:
+            from .security_telemetry import get_security_telemetry
+
+            remote = get_client_ip(request)
+            get_security_telemetry().record_auth_failure(remote)
+        except ImportError:
+            pass
         return False, "Observability access required."
 
+    # R102 Hook for generic failure
+    try:
+        from .security_telemetry import get_security_telemetry
+
+        remote = get_client_ip(request)
+        get_security_telemetry().record_auth_failure(remote)
+    except ImportError:
+        pass
     return False, f"Access denied. Required: {required_tier}, Current: {current_tier}"
 
 

services/control_plane.py

@@ -251,6 +251,15 @@ def enforce_control_plane_startup() -> Dict:
             logger.warning(
                 "S62: Running public+embedded with compat override (DEV ONLY)."
             )
+            # R102 Hook
+            try:
+                from .security_telemetry import get_security_telemetry
+
+                get_security_telemetry().record_dangerous_override(
+                    "SPLIT_COMPAT_OVERRIDE", "system_env"
+                )
+            except ImportError:
+                pass
 
     else:
         # local/lan: always pass

services/policy_posture.py

@@ -0,0 +1,314 @@
+"""
+R103: Policy-as-code posture controls.
+Manages signed, versioned security policy bundles with atomic activation and rollback.
+"""
+
+import json
+import logging
+import os
+import shutil
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Union
+
+# Try to import cryptography for signature verification
+try:
+    from cryptography.exceptions import InvalidSignature
+    from cryptography.hazmat.primitives import serialization
+    from cryptography.hazmat.primitives.asymmetric import ed25519
+
+    HAS_CRYPTO = True
+except ImportError:
+    HAS_CRYPTO = False
+
+from .audit_events import build_audit_event, emit_audit_event
+from .state_dir import get_state_dir
+
+logger = logging.getLogger("ComfyUI-OpenClaw.services.policy_posture")
+
+POLICY_DIR_NAME = "policy"
+ACTIVE_BUNDLE_NAME = "active.bundle.json"
+BACKUP_BUNDLE_NAME = "backup.bundle.json"
+STAGED_BUNDLE_NAME = "staged.bundle.json"
+TRUSTED_KEYS_NAME = "trusted_keys.json"
+
+
+@dataclass
+class PolicyPayload:
+    """The actual policy content."""
+
+    allowlists: Dict[str, List[str]] = field(default_factory=dict)
+    high_risk_flags: Dict[str, bool] = field(default_factory=dict)
+    quota_posture: Dict[str, Any] = field(default_factory=dict)
+    meta: Dict[str, Any] = field(default_factory=dict)
+
+    def to_canonical_bytes(self) -> bytes:
+        """
+        Produce a canonical byte representation for signing.
+        Sort keys, no spaces.
+        """
+        data = {
+            "allowlists": self.allowlists,
+            "high_risk_flags": self.high_risk_flags,
+            "quota_posture": self.quota_posture,
+            "meta": self.meta,
+        }
+        return json.dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "PolicyPayload":
+        return cls(
+            allowlists=data.get("allowlists", {}),
+            high_risk_flags=data.get("high_risk_flags", {}),
+            quota_posture=data.get("quota_posture", {}),
+            meta=data.get("meta", {}),
+        )
+
+
+@dataclass
+class PolicyBundle:
+    """Signed policy bundle container."""
+
+    payload: PolicyPayload
+    signature: str  # Hex-encoded signature
+    signer_id: str
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "payload": {
+                "allowlists": self.payload.allowlists,
+                "high_risk_flags": self.payload.high_risk_flags,
+                "quota_posture": self.payload.quota_posture,
+                "meta": self.payload.meta,
+            },
+            "signature": self.signature,
+            "signer_id": self.signer_id,
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "PolicyBundle":
+        payload_data = data.get("payload", {})
+        return cls(
+            payload=PolicyPayload.from_dict(payload_data),
+            signature=data.get("signature", ""),
+            signer_id=data.get("signer_id", ""),
+        )
+
+    def verify(self, public_keys: Dict[str, str]) -> bool:
+        """
+        Verify the signature against trusted public keys.
+        public_keys: dict of {signer_id: hex_encoded_public_key}
+        """
+        if not HAS_CRYPTO:
+            logger.warning(
+                "Cryptography module missing, cannot verify policy signature. FAIL-CLOSED."
+            )
+            return False
+
+        if self.signer_id not in public_keys:
+            logger.error(f"Unknown signer_id: {self.signer_id}")
+            return False
+
+        pub_key_hex = public_keys[self.signer_id]
+        try:
+            pub_key_bytes = bytes.fromhex(pub_key_hex)
+            public_key = ed25519.Ed25519PublicKey.from_public_bytes(pub_key_bytes)
+
+            sig_bytes = bytes.fromhex(self.signature)
+            data_bytes = self.payload.to_canonical_bytes()
+
+            public_key.verify(sig_bytes, data_bytes)
+            return True
+        except (ValueError, InvalidSignature) as e:
+            logger.error(f"Signature verification failed: {e}")
+            return False
+        except Exception as e:
+            logger.error(f"Unexpected error during verification: {e}")
+            return False
+
+
+class PolicyManager:
+    """Manages policy lifecycle: stage -> activate -> rollback."""
+
+    def __init__(self):
+        self.state_dir = Path(get_state_dir()) / POLICY_DIR_NAME
+        self.state_dir.mkdir(parents=True, exist_ok=True)
+        self.active_policy: Optional[PolicyBundle] = None
+        self.trusted_keys: Dict[str, str] = {}
+
+        self._load_trusted_keys()
+        self._load_active_policy()
+
+    def _load_trusted_keys(self):
+        """Load trusted public keys from disk."""
+        keys_path = self.state_dir / TRUSTED_KEYS_NAME
+        if keys_path.exists():
+            try:
+                with open(keys_path, "r", encoding="utf-8") as f:
+                    self.trusted_keys = json.load(f)
+            except Exception as e:
+                logger.error(f"Failed to load trusted keys: {e}")
+
+        # If no keys, we might be in uninitialized state.
+        # But if we have an active policy, we MUST have keys to verify it on startup (fail-closed).
+
+    def _load_active_policy(self):
+        """Load and verify active policy. Fail-closed if invalid."""
+        active_path = self.state_dir / ACTIVE_BUNDLE_NAME
+        if not active_path.exists():
+            logger.info(
+                "No active policy bundle found. Running with default/empty policy."
+            )
+            return
+
+        try:
+            with open(active_path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+
+            bundle = PolicyBundle.from_dict(data)
+
+            # Fail-closed check
+            if not self.trusted_keys:
+                # If we have a policy but no keys, that's a security risk.
+                # In a strict hardened mode, we should crash.
+                # For now, we log critical error and refuse to make it active.
+                # ACTUALLY, requirements say "Hardened posture must fail-closed".
+                # I'll log critical and raise exception if keys are missing but policy exists.
+                msg = "Fail-closed: Active policy exists but no trusted keys found to verify it."
+                logger.critical(msg)
+                raise RuntimeError(msg)
+
+            if not bundle.verify(self.trusted_keys):
+                msg = "Fail-closed: Active policy signature invalid."
+                logger.critical(msg)
+                raise RuntimeError(msg)
+
+            self.active_policy = bundle
+            logger.info(
+                f"Active policy loaded: {bundle.payload.meta.get('version', 'unknown')}"
+            )
+
+        except Exception as e:
+            logger.critical(f"Failed to load active policy: {e}")
+            raise RuntimeError(f"Policy load failure: {e}")
+
+    def get_effective_policy(self) -> Optional[PolicyBundle]:
+        return self.active_policy
+
+    def stage_bundle(self, bundle_json: Dict[str, Any]) -> bool:
+        """
+        Validate and stage a new policy bundle.
+        Returns True if successful.
+        """
+        try:
+            bundle = PolicyBundle.from_dict(bundle_json)
+
+            if not bundle.verify(self.trusted_keys):
+                self._audit(
+                    "policy.stage_failed",
+                    {"reason": "invalid_signature", "signer": bundle.signer_id},
+                )
+                return False
+
+            # Save to staging
+            staged_path = self.state_dir / STAGED_BUNDLE_NAME
+            with open(staged_path, "w", encoding="utf-8") as f:
+                json.dump(bundle.to_dict(), f, indent=2)
+
+            self._audit(
+                "policy.staged", {"version": bundle.payload.meta.get("version")}
+            )
+            return True
+
+        except Exception as e:
+            logger.error(f"Failed to stage bundle: {e}")
+            self._audit("policy.stage_failed", {"reason": str(e)})
+            return False
+
+    def activate_staged(self) -> bool:
+        """Promote staged bundle to active."""
+        staged_path = self.state_dir / STAGED_BUNDLE_NAME
+        active_path = self.state_dir / ACTIVE_BUNDLE_NAME
+        backup_path = self.state_dir / BACKUP_BUNDLE_NAME
+
+        if not staged_path.exists():
+            logger.warning("No staged bundle to activate")
+            return False
+
+        try:
+            # Load staged to verify it one last time (and get version)
+            with open(staged_path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+            new_bundle = PolicyBundle.from_dict(data)
+
+            # 1. Backup existing active
+            if active_path.exists():
+                shutil.copy2(active_path, backup_path)
+
+            # 2. Move staged to active
+            shutil.move(staged_path, active_path)
+
+            # 3. Update memory
+            self.active_policy = new_bundle
+
+            self._audit(
+                "policy.activated",
+                {
+                    "version": new_bundle.payload.meta.get("version"),
+                    "hash": new_bundle.signature[:8],
+                },
+            )
+            return True
+
+        except Exception as e:
+            logger.error(f"Activation failed: {e}")
+            self._audit("policy.activation_failed", {"error": str(e)})
+            # Try to restore from backup if we messed up active
+            if backup_path.exists() and not active_path.exists():
+                shutil.copy2(backup_path, active_path)
+            return False
+
+    def rollback(self) -> bool:
+        """Rollback to previous active bundle."""
+        active_path = self.state_dir / ACTIVE_BUNDLE_NAME
+        backup_path = self.state_dir / BACKUP_BUNDLE_NAME
+
+        if not backup_path.exists():
+            logger.warning("No backup bundle found for rollback")
+            return False
+
+        try:
+            # Move backup to active
+            shutil.copy2(backup_path, active_path)
+
+            # Reload
+            self._load_active_policy()
+
+            version = "unknown"
+            if self.active_policy:
+                version = self.active_policy.payload.meta.get("version", "unknown")
+
+            self._audit("policy.rollback", {"version": version})
+            return True
+        except Exception as e:
+            logger.error(f"Rollback failed: {e}")
+            self._audit("policy.rollback_failed", {"error": str(e)})
+            return False
+
+    def _audit(self, event_type: str, payload: Dict[str, Any]):
+        event = build_audit_event(
+            event_type=event_type, payload=payload, meta={"component": "PolicyManager"}
+        )
+        emit_audit_event(event)
+
+
+# Global singleton
+_policy_manager = None
+
+
+def get_policy_manager() -> PolicyManager:
+    global _policy_manager
+    if _policy_manager is None:
+        _policy_manager = PolicyManager()
+    return _policy_manager

services/queue_submit.py

@@ -136,6 +136,16 @@ async def submit_prompt(
                 async with session.post(url, json=payload) as resp:
                     if resp.status == 200:
                         data = await resp.json()
+
+                        # R102 Hook
+                        try:
+                            q_size = data.get("number", 0)
+                            from .security_telemetry import get_security_telemetry
+
+                            get_security_telemetry().record_queue_saturation(q_size)
+                        except:
+                            pass
+
                         logger.info(
                             f"Queued prompt: {data.get('prompt_id')} (source={source}, trace_id={trace_id})"
                         )

services/security_telemetry.py

@@ -0,0 +1,187 @@
+"""
+R102: Security telemetry + alert contract.
+Defines bounded anomaly event schema and deterministic anomaly producers.
+"""
+
+import logging
+import time
+from collections import deque
+from dataclasses import dataclass, field
+from typing import Deque, Dict, List, Optional
+
+from .audit_events import build_audit_event, emit_audit_event
+
+logger = logging.getLogger("ComfyUI-OpenClaw.services.security_telemetry")
+
+# Anomaly Codes
+ANOMALY_AUTH_FAILURE_SPIKE = "SEC-001"
+ANOMALY_REPLAY_BURST = "SEC-002"
+ANOMALY_DANGEROUS_OVERRIDE = "SEC-003"
+ANOMALY_QUEUE_SATURATION = "SEC-004"
+
+# Thresholds (default) -> moved to configuration in future
+THRESHOLDS = {
+    ANOMALY_AUTH_FAILURE_SPIKE: {"count": 10, "window": 60},  # 10 failures in 60s
+    ANOMALY_REPLAY_BURST: {"count": 20, "window": 10},  # 20 replays in 10s
+    ANOMALY_QUEUE_SATURATION: {
+        "count": 100,
+        "window": 300,
+    },  # 100 queued items sustained? No, simple count check
+}
+
+
+@dataclass
+class AnomalyEvent:
+    code: str
+    severity: str  # "low", "medium", "high", "critical"
+    source: str
+    count: int
+    window: float
+    action: str  # "monitor", "block", "alert"
+
+    def to_dict(self) -> Dict:
+        return {
+            "code": self.code,
+            "severity": self.severity,
+            "source": self.source,
+            "count": self.count,
+            "window": self.window,
+            "action": self.action,
+        }
+
+
+class TimeWindowCounter:
+    """Tracks event counts within a sliding time window."""
+
+    def __init__(self, window_seconds: float):
+        self.window_seconds = window_seconds
+        self.timestamps: Deque[float] = deque()
+
+    def add(self):
+        now = time.time()
+        self.timestamps.append(now)
+        self._prune(now)
+
+    def count(self) -> int:
+        self._prune(time.time())
+        return len(self.timestamps)
+
+    def _prune(self, now: float):
+        while self.timestamps and (now - self.timestamps[0] > self.window_seconds):
+            self.timestamps.popleft()
+
+
+class SecurityTelemetry:
+    def __init__(self):
+        self._auth_failure_counter = TimeWindowCounter(
+            THRESHOLDS[ANOMALY_AUTH_FAILURE_SPIKE]["window"]
+        )
+        self._replay_counter = TimeWindowCounter(
+            THRESHOLDS[ANOMALY_REPLAY_BURST]["window"]
+        )
+        # Suppress duplicate alerts for a short period
+        self._last_alert_time: Dict[str, float] = {}
+
+    def record_auth_failure(self, source_ip: str):
+        """Record an authentication failure."""
+        self._auth_failure_counter.add()
+        count = self._auth_failure_counter.count()
+        threshold = THRESHOLDS[ANOMALY_AUTH_FAILURE_SPIKE]["count"]
+
+        if count >= threshold:
+            self._trigger_anomaly(
+                ANOMALY_AUTH_FAILURE_SPIKE,
+                "medium",
+                source=f"auth_module:{source_ip}",
+                count=count,
+                window=THRESHOLDS[ANOMALY_AUTH_FAILURE_SPIKE]["window"],
+                action="alert",
+            )
+
+    def record_replay_rejection(self, source: str):
+        """Record a replay attack rejection."""
+        self._replay_counter.add()
+        count = self._replay_counter.count()
+        threshold = THRESHOLDS[ANOMALY_REPLAY_BURST]["count"]
+
+        if count >= threshold:
+            self._trigger_anomaly(
+                ANOMALY_REPLAY_BURST,
+                "high",
+                source=source,
+                count=count,
+                window=THRESHOLDS[ANOMALY_REPLAY_BURST]["window"],
+                action="block",
+            )
+
+    def record_dangerous_override(self, override_key: str, user: str):
+        """Record usage of a dangerous override (always an anomaly)."""
+        self._trigger_anomaly(
+            ANOMALY_DANGEROUS_OVERRIDE,
+            "medium",
+            source=f"{user}:{override_key}",
+            count=1,
+            window=0,
+            action="monitor",
+        )
+
+    def record_queue_saturation(self, queue_size: int):
+        """Record queue saturation event."""
+        # This might be called periodically by a queue monitor
+        if queue_size > 1000:  # specific hardcoded limit for now
+            self._trigger_anomaly(
+                ANOMALY_QUEUE_SATURATION,
+                "low",
+                source="job_queue",
+                count=queue_size,
+                window=0,
+                action="monitor",
+            )
+
+    def _trigger_anomaly(
+        self,
+        code: str,
+        severity: str,
+        source: str,
+        count: int,
+        window: float,
+        action: str,
+    ):
+        # Debounce alerts: don't fire same alert code for same source too often (e.g., every 10s)
+        alert_key = f"{code}:{source}"
+        now = time.time()
+        if now - self._last_alert_time.get(alert_key, 0) < 10:
+            return
+
+        self._last_alert_time[alert_key] = now
+
+        anomaly = AnomalyEvent(
+            code=code,
+            severity=severity,
+            source=source,
+            count=count,
+            window=window,
+            action=action,
+        )
+
+        # Log via Audit Service
+        event = build_audit_event(
+            event_type="security.anomaly",
+            payload=anomaly.to_dict(),
+            meta={"component": "SecurityTelemetry"},
+        )
+        emit_audit_event(event)
+
+        # Also log to structured logger
+        logger.warning(f"Security Anomaly Detected: {anomaly.to_dict()}")
+
+
+# Global singleton
+_telemetry_instance = None
+
+
+def get_security_telemetry() -> SecurityTelemetry:
+    global _telemetry_instance
+    if _telemetry_instance is None:
+        _telemetry_instance = SecurityTelemetry()
+    return _telemetry_instance

services/threat_intel_gate.py

@@ -1,4 +1,3 @@
-
 """
 S43 — Threat-Intel Gate v1.
 
@@ -12,24 +11,27 @@ Policy Modes:
 """
 
 import enum
+import hashlib
 import logging
 import os
-import hashlib
 from dataclasses import dataclass
-from typing import Optional, Tuple, Dict, Any
+from typing import Any, Dict, Optional, Tuple
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.threat_intel_gate")
 
+
 class ThreatPolicy(enum.Enum):
     OFF = "off"
     AUDIT = "audit"
     STRICT = "strict"
 
+
 class ScanVerdict(enum.Enum):
     CLEAN = "clean"
     MALICIOUS = "malicious"
     UNKNOWN = "unknown"
-    ERROR = "error" # Provider unreachable
+    ERROR = "error"  # Provider unreachable
+
 
 @dataclass
 class ScanResult:
@@ -38,16 +40,17 @@ class ScanResult:
     provider: str = "none"
     score: float = 0.0
 
+
 class ThreatIntelGate:
     """
     Gate for evaluating files against threat policy.
     """
-    
+
     def __init__(self):
         self._policy = self._load_policy()
         # R89: Provider integration will be injected or loaded here.
         # For S43 baseline, we assume a "provider interface".
-        self._provider = None 
+        self._provider = None
 
     def _load_policy(self) -> ThreatPolicy:
         val = os.environ.get("OPENCLAW_THREAT_POLICY", "off").lower()
@@ -81,8 +84,8 @@ class ThreatIntelGate:
 
         if not os.path.exists(file_path):
             logger.warning(f"S43: File not found for scan: {file_path}")
-            # If Strict, strict missing file handling? 
-            # Usually if file is missing, we can't scan, so maybe allow? 
+            # If Strict, strict missing file handling?
+            # Usually if file is missing, we can't scan, so maybe allow?
             # Or if it's "check this upload", and it's missing, fail.
             # Assuming caller ensures existence. If not, fail safe.
             if self._policy == ThreatPolicy.STRICT:
@@ -90,22 +93,24 @@ class ThreatIntelGate:
             return True
 
         file_hash = self._compute_hash(file_path)
-        
+
         # 1. Hash Lookup (Optimization / Privacy)
         result = self._scan_hash(file_hash)
-        
+
         # 2. Upload (Opt-In / Fallback)
-        # R89 will implement resilience/upload logic. 
+        # R89 will implement resilience/upload logic.
         # S43 Gate just consumes the verdict.
-        
+
         # Decision Logic
         allowed, reason = self._apply_policy(result)
-        
+
         if not allowed:
             logger.warning(f"S43: BLOCKED {context} [{file_hash[:8]}] Reason: {reason}")
             return False
-            
-        logger.info(f"S43: ALLOWED {context} [{file_hash[:8]}] Verdict: {result.verdict.value}")
+
+        logger.info(
+            f"S43: ALLOWED {context} [{file_hash[:8]}] Verdict: {result.verdict.value}"
+        )
         return True
 
     def _scan_hash(self, file_hash: str) -> ScanResult:
@@ -115,7 +120,7 @@ class ThreatIntelGate:
             # STRICT -> Fail-Closed (Error)
             # AUDIT -> Log Error, return Unknown
             return ScanResult(ScanVerdict.ERROR, "No provider configured")
-            
+
         try:
             return self._provider.check_hash(file_hash)
         except Exception as e:
@@ -137,7 +142,9 @@ class ThreatIntelGate:
             if self._policy == ThreatPolicy.STRICT:
                 return False, f"Malicious content detected ({result.provider})"
             # AUDIT: Log but allow
-            logger.warning(f"S43: AUDIT - Malicious content detected but allowed by policy.")
+            logger.warning(
+                f"S43: AUDIT - Malicious content detected but allowed by policy."
+            )
             return True, "Audit Mode (Malicious)"
 
         if result.verdict == ScanVerdict.UNKNOWN:
@@ -154,8 +161,11 @@ class ThreatIntelGate:
 
         return True, "Default Allow"
 
+
 # Singleton
 _gate = None
+
+
 def get_gate() -> ThreatIntelGate:
     global _gate
     if _gate is None:

services/threat_intel_provider.py

@@ -1,4 +1,3 @@
-
 """
 R89 — Threat-Intel Provider Resilience v1.
 
@@ -11,10 +10,10 @@ Contract:
 - Failures inside the wrapper result in ScanVerdict.ERROR (or handled by policy).
 """
 
-import time
 import logging
+import time
 from dataclasses import dataclass
-from typing import Optional, Protocol, Any
+from typing import Any, Optional, Protocol
 
 # Import shared types
 try:
@@ -24,9 +23,10 @@ except ImportError:
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.threat_intel_provider")
 
+
 class ThreatIntelProvider(Protocol):
-    def check_hash(self, sha256: str) -> ScanResult:
-        ...
+    def check_hash(self, sha256: str) -> ScanResult: ...
+
 
 @dataclass
 class ResilienceConfig:
@@ -35,66 +35,86 @@ class ResilienceConfig:
     circuit_breaker_threshold: int = 5
     circuit_breaker_reset_sec: float = 30.0
 
+
 class ResilientProviderWrapper:
     """
     Wraps a ThreatIntelProvider with resilience logic.
     - Retry on transient errors (exceptions).
     - Circuit Breaker to stop calling dead provider.
     """
-    
-    def __init__(self, provider: ThreatIntelProvider, config: ResilienceConfig = ResilienceConfig()):
+
+    def __init__(
+        self,
+        provider: ThreatIntelProvider,
+        config: ResilienceConfig = ResilienceConfig(),
+    ):
         self._provider = provider
         self._config = config
-        
+
         # Circuit Breaker State
         self._cb_failures = 0
         self._cb_last_failure = 0.0
         self._cb_open = False
-        
+
     def check_hash(self, sha256: str) -> ScanResult:
         # 1. Check Circuit Breaker
         if self._cb_open:
-            if time.time() - self._cb_last_failure > self._config.circuit_breaker_reset_sec:
+            if (
+                time.time() - self._cb_last_failure
+                > self._config.circuit_breaker_reset_sec
+            ):
                 # Half-Open: Try once
                 logger.info("R89: Circuit Breaker Half-Open - Attempting probe.")
             else:
                 # Open: Fail Fast
-                return ScanResult(ScanVerdict.ERROR, "Circuit Breaker OPEN", provider="resilience_wrapper")
+                return ScanResult(
+                    ScanVerdict.ERROR,
+                    "Circuit Breaker OPEN",
+                    provider="resilience_wrapper",
+                )
 
         # 2. Try with Retries
         attempts = 0
         last_error = None
-        
+
         while attempts <= self._config.max_retries:
             try:
                 result = self._provider.check_hash(sha256)
-                
+
                 # Success - Reset Circuit Breaker
                 if self._cb_open or self._cb_failures > 0:
                     self._reset_cb()
-                    
+
                 return result
-                
+
             except Exception as e:
                 attempts += 1
                 last_error = e
                 logger.warning(f"R89: Provider attempt {attempts} failed: {e}")
-                
+
                 if attempts <= self._config.max_retries:
-                    time.sleep(self._config.retry_delay_sec * attempts) # Linear backoff
-        
+                    time.sleep(
+                        self._config.retry_delay_sec * attempts
+                    )  # Linear backoff
+
         # 3. Failure - Trip Circuit Breaker
         self._trip_cb()
-        return ScanResult(ScanVerdict.ERROR, f"Max retries exceeded: {last_error}", provider="resilience_wrapper")
+        return ScanResult(
+            ScanVerdict.ERROR,
+            f"Max retries exceeded: {last_error}",
+            provider="resilience_wrapper",
+        )
 
     def _trip_cb(self):
         self._cb_failures += 1
         self._cb_last_failure = time.time()
-        
+
         if self._cb_failures >= self._config.circuit_breaker_threshold:
             if not self._cb_open:
                 self._cb_open = True
-                logger.error(f"R89: Circuit Breaker TRIPPED (threshold {self._config.circuit_breaker_threshold})")
+                logger.error(
+                    f"R89: Circuit Breaker TRIPPED (threshold {self._config.circuit_breaker_threshold})"
+                )
 
     def _reset_cb(self):
         if self._cb_open:

services/webhook_auth.py

@@ -194,6 +194,15 @@ def verify_hmac(request: RequestLike, raw_body: bytes) -> Tuple[bool, str]:
 
         now = int(time.time())
         if abs(now - ts) > 300:
+            # R102 Hook
+            try:
+                from .security_telemetry import get_security_telemetry
+
+                get_security_telemetry().record_replay_rejection(
+                    "timestamp_out_of_range"
+                )
+            except ImportError:
+                pass
             return False, "timestamp_out_of_range"
 
         # Check nonce uniqueness

tests/test_r102_security_telemetry.py

@@ -0,0 +1,84 @@
+import unittest
+from unittest.mock import patch
+
+from services import security_telemetry
+
+
+class TestR102SecurityTelemetry(unittest.TestCase):
+    def setUp(self):
+        self.telemetry = security_telemetry.SecurityTelemetry()
+
+    def _event_factory(self, event_type, payload, meta):
+        return {"event_type": event_type, "payload": payload, "meta": meta}
+
+    def test_anomaly_schema_keys(self):
+        event = security_telemetry.AnomalyEvent(
+            code="SEC-001",
+            severity="medium",
+            source="auth_module:127.0.0.1",
+            count=10,
+            window=60,
+            action="alert",
+        )
+        keys = set(event.to_dict().keys())
+        self.assertEqual(
+            keys, {"code", "severity", "source", "count", "window", "action"}
+        )
+
+    def test_auth_failure_spike_triggers_alert(self):
+        with (
+            patch(
+                "services.security_telemetry.build_audit_event",
+                side_effect=self._event_factory,
+            ),
+            patch("services.security_telemetry.emit_audit_event") as emit,
+        ):
+            for _ in range(
+                security_telemetry.THRESHOLDS[
+                    security_telemetry.ANOMALY_AUTH_FAILURE_SPIKE
+                ]["count"]
+            ):
+                self.telemetry.record_auth_failure("127.0.0.1")
+
+            self.assertGreaterEqual(emit.call_count, 1)
+            payload = emit.call_args.args[0]["payload"]
+            self.assertEqual(
+                payload["code"], security_telemetry.ANOMALY_AUTH_FAILURE_SPIKE
+            )
+
+    def test_replay_burst_triggers_block_action(self):
+        with (
+            patch(
+                "services.security_telemetry.build_audit_event",
+                side_effect=self._event_factory,
+            ),
+            patch("services.security_telemetry.emit_audit_event") as emit,
+        ):
+            for _ in range(
+                security_telemetry.THRESHOLDS[security_telemetry.ANOMALY_REPLAY_BURST][
+                    "count"
+                ]
+            ):
+                self.telemetry.record_replay_rejection("webhook")
+
+            payload = emit.call_args.args[0]["payload"]
+            self.assertEqual(payload["code"], security_telemetry.ANOMALY_REPLAY_BURST)
+            self.assertEqual(payload["action"], "block")
+
+    def test_override_and_queue_events_emit(self):
+        with (
+            patch(
+                "services.security_telemetry.build_audit_event",
+                side_effect=self._event_factory,
+            ),
+            patch("services.security_telemetry.emit_audit_event") as emit,
+        ):
+            self.telemetry.record_dangerous_override("OVERRIDE_X", "tester")
+            self.telemetry.record_queue_saturation(1500)
+            codes = [call.args[0]["payload"]["code"] for call in emit.call_args_list]
+            self.assertIn(security_telemetry.ANOMALY_DANGEROUS_OVERRIDE, codes)
+            self.assertIn(security_telemetry.ANOMALY_QUEUE_SATURATION, codes)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_r103_policy_posture.py

@@ -0,0 +1,104 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from services import policy_posture
+
+
+class TestR103PolicyBundle(unittest.TestCase):
+    def _make_bundle(self):
+        if not policy_posture.HAS_CRYPTO:
+            self.skipTest("cryptography not available")
+
+        private_key = policy_posture.ed25519.Ed25519PrivateKey.generate()
+        public_key = private_key.public_key()
+        public_hex = public_key.public_bytes(
+            encoding=policy_posture.serialization.Encoding.Raw,
+            format=policy_posture.serialization.PublicFormat.Raw,
+        ).hex()
+
+        payload = policy_posture.PolicyPayload(
+            allowlists={"hosts": ["example.com"]},
+            high_risk_flags={"dangerous_override": False},
+            quota_posture={"max_jobs": 10},
+            meta={"version": "v1"},
+        )
+        sig = private_key.sign(payload.to_canonical_bytes()).hex()
+        bundle = policy_posture.PolicyBundle(
+            payload=payload, signature=sig, signer_id="test"
+        )
+        return bundle, {"test": public_hex}
+
+    def test_verify_valid_signature(self):
+        bundle, keys = self._make_bundle()
+        self.assertTrue(bundle.verify(keys))
+
+    def test_verify_unknown_signer(self):
+        bundle, _ = self._make_bundle()
+        self.assertFalse(bundle.verify({}))
+
+    def test_verify_tampered_payload(self):
+        bundle, keys = self._make_bundle()
+        bundle.payload.meta["version"] = "v2"
+        self.assertFalse(bundle.verify(keys))
+
+
+class TestR103PolicyManager(unittest.TestCase):
+    def _make_bundle_dict(self):
+        if not policy_posture.HAS_CRYPTO:
+            self.skipTest("cryptography not available")
+
+        private_key = policy_posture.ed25519.Ed25519PrivateKey.generate()
+        public_key = private_key.public_key()
+        public_hex = public_key.public_bytes(
+            encoding=policy_posture.serialization.Encoding.Raw,
+            format=policy_posture.serialization.PublicFormat.Raw,
+        ).hex()
+
+        payload = policy_posture.PolicyPayload(meta={"version": "v1"})
+        sig = private_key.sign(payload.to_canonical_bytes()).hex()
+        bundle = policy_posture.PolicyBundle(
+            payload=payload, signature=sig, signer_id="test"
+        )
+        return bundle.to_dict(), {"test": public_hex}
+
+    def test_stage_and_activate_bundle(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            bundle_dict, keys = self._make_bundle_dict()
+
+            with (
+                patch("services.policy_posture.get_state_dir", return_value=tmp),
+                patch(
+                    "services.policy_posture.build_audit_event",
+                    side_effect=lambda event_type, payload, meta: {
+                        "event_type": event_type,
+                        "payload": payload,
+                        "meta": meta,
+                    },
+                ),
+                patch("services.policy_posture.emit_audit_event"),
+            ):
+                manager = policy_posture.PolicyManager()
+                manager.trusted_keys = keys
+                self.assertTrue(manager.stage_bundle(bundle_dict))
+                self.assertTrue(manager.activate_staged())
+                self.assertIsNotNone(manager.get_effective_policy())
+
+    def test_fail_closed_active_policy_without_keys(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            bundle_dict, _ = self._make_bundle_dict()
+            policy_dir = Path(tmp) / policy_posture.POLICY_DIR_NAME
+            policy_dir.mkdir(parents=True, exist_ok=True)
+            (policy_dir / policy_posture.ACTIVE_BUNDLE_NAME).write_text(
+                json.dumps(bundle_dict), encoding="utf-8"
+            )
+
+            with patch("services.policy_posture.get_state_dir", return_value=tmp):
+                with self.assertRaises(RuntimeError):
+                    policy_posture.PolicyManager()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_r108_security_matrix.py

@@ -1,41 +1,41 @@
-
-import unittest
-import time
 import json
 import logging
+import time
+import unittest
 from unittest.mock import MagicMock, patch
 
 # Service Imports
 from services.bridge_token_lifecycle import (
+    BridgeScope,
     BridgeTokenStore,
     DeviceToken,
     TokenStatus,
-    BridgeScope,
-    TokenValidationResult
-)
-from services.webhook_mapping import (
-    MappingProfile,
-    FieldMapping,
-    CoercionType,
-    apply_mapping,
-    validate_canonical_schema,
-    PRIVILEGED_FIELDS,
-    ALLOWED_PRIVILEGED_OVERRIDES
+    TokenValidationResult,
 )
 from services.endpoint_manifest import (
-    validate_mae_posture,
-    EndpointMetadata,
     AuthTier,
+    EndpointMetadata,
     RiskTier,
-    RoutePlane
+    RoutePlane,
+    validate_mae_posture,
 )
 from services.registry_quarantine import (
-    RegistryQuarantineStore,
+    QuarantineState,
     RegistryEntry,
+    RegistryQuarantineStore,
     TrustRoot,
     TrustRootStore,
-    QuarantineState
 )
+from services.webhook_mapping import (
+    ALLOWED_PRIVILEGED_OVERRIDES,
+    PRIVILEGED_FIELDS,
+    CoercionType,
+    FieldMapping,
+    MappingProfile,
+    apply_mapping,
+    validate_canonical_schema,
+)
+
 
 class TestR108SecurityMatrix(unittest.TestCase):
     """
@@ -56,7 +56,7 @@ class TestR108SecurityMatrix(unittest.TestCase):
             "token_lifecycle": "Covered 4 states (active, expired, revoked, overlap)",
             "webhook_mapping": "Covered 5 clamps (allowed, blocked, nested, type, oversize)",
             "mae_route": "Covered 3 postures (user/admin/internal) x 3 profiles",
-            "registry_trust": "Covered 6 decisions (valid, tampered, unknown, expired, revoked, unavailable)"
+            "registry_trust": "Covered 6 decisions (valid, tampered, unknown, expired, revoked, unavailable)",
         }
         # In a real run, verify this matches implemented tests or output to artifact
         pass
@@ -71,7 +71,7 @@ class TestR108SecurityMatrix(unittest.TestCase):
         """
         store = BridgeTokenStore()  # In-memory
         now = time.time()
-        
+
         # Helper to inject token state
         def inject_token(tid, status, expires, overlap=None, scopes=None):
             t = DeviceToken(
@@ -82,9 +82,11 @@ class TestR108SecurityMatrix(unittest.TestCase):
                 token_id=tid,
                 issued_at=now - 100,
                 status=status,
-                overlap_until=overlap
+                overlap_until=overlap,
             )
-            store._active_tokens_mock = {tid: t} # Bypass hashing for direct injection if possible?
+            store._active_tokens_mock = {
+                tid: t
+            }  # Bypass hashing for direct injection if possible?
             # Store uses _tokens by ID and _token_index by hash.
             # We must use proper injection.
             store._tokens[tid] = t
@@ -95,26 +97,82 @@ class TestR108SecurityMatrix(unittest.TestCase):
         cases = [
             # Case Name | Status | Expires Relative | Overlap Relative | Req Scope | Expect OK | Expect Reason
             ("active", TokenStatus.ACTIVE.value, 3600, None, None, True, None),
-            ("expired", TokenStatus.ACTIVE.value, -100, None, None, False, "token_expired"),
-            ("revoked", TokenStatus.REVOKED.value, 3600, None, None, False, "token_revoked"),
-            ("overlap_valid", TokenStatus.ACTIVE.value, 3600, 100, None, True, None), # Overlap limit in future
-            ("overlap_expired", TokenStatus.ACTIVE.value, 3600, -10, None, False, "overlap_window_expired"),
-            ("scope_good", TokenStatus.ACTIVE.value, 3600, None, BridgeScope.JOB_STATUS.value, True, None),
-            ("scope_bad", TokenStatus.ACTIVE.value, 3600, None, "admin_root", False, "insufficient_scope"),
+            (
+                "expired",
+                TokenStatus.ACTIVE.value,
+                -100,
+                None,
+                None,
+                False,
+                "token_expired",
+            ),
+            (
+                "revoked",
+                TokenStatus.REVOKED.value,
+                3600,
+                None,
+                None,
+                False,
+                "token_revoked",
+            ),
+            (
+                "overlap_valid",
+                TokenStatus.ACTIVE.value,
+                3600,
+                100,
+                None,
+                True,
+                None,
+            ),  # Overlap limit in future
+            (
+                "overlap_expired",
+                TokenStatus.ACTIVE.value,
+                3600,
+                -10,
+                None,
+                False,
+                "overlap_window_expired",
+            ),
+            (
+                "scope_good",
+                TokenStatus.ACTIVE.value,
+                3600,
+                None,
+                BridgeScope.JOB_STATUS.value,
+                True,
+                None,
+            ),
+            (
+                "scope_bad",
+                TokenStatus.ACTIVE.value,
+                3600,
+                None,
+                "admin_root",
+                False,
+                "insufficient_scope",
+            ),
         ]
 
-        for case_name, status, exp_rel, overlap_rel, scope, expect_ok, expect_reason in cases:
+        for (
+            case_name,
+            status,
+            exp_rel,
+            overlap_rel,
+            scope,
+            expect_ok,
+            expect_reason,
+        ) in cases:
             with self.subTest(case=case_name):
                 # Setup
                 expires = now + exp_rel if exp_rel else None
                 overlap = (now + overlap_rel) if overlap_rel else None
                 scopes = [BridgeScope.JOB_STATUS]
-                
+
                 inject_token(f"t_{case_name}", status, expires, overlap, scopes)
-                
+
                 # Act
                 res = store.validate_token("secret", required_scope=scope)
-                
+
                 # Assert
                 self.assertEqual(res.ok, expect_ok, f"Case {case_name}: OK mismatch")
                 if expect_reason:
@@ -131,7 +189,7 @@ class TestR108SecurityMatrix(unittest.TestCase):
         """
         # Ensure cleanup of global allowlist
         original_allowlist = set(ALLOWED_PRIVILEGED_OVERRIDES)
-        
+
         try:
             # Setup profiles
             ALLOWED_PRIVILEGED_OVERRIDES.add(("allowed_profile", "template_id"))
@@ -142,8 +200,18 @@ class TestR108SecurityMatrix(unittest.TestCase):
                 ("inputs_ok", "blocked_profile", "inputs.foo", False),
                 ("priv_blocked", "blocked_profile", "template_id", True),
                 ("priv_allowed", "allowed_profile", "template_id", False),
-                ("priv_nested_blocked", "blocked_profile", "template_id.sub", True), # Root clamps
-                ("defaults_check", "blocked_profile", "profile_id", True), # profile_id is privileged
+                (
+                    "priv_nested_blocked",
+                    "blocked_profile",
+                    "template_id.sub",
+                    True,
+                ),  # Root clamps
+                (
+                    "defaults_check",
+                    "blocked_profile",
+                    "profile_id",
+                    True,
+                ),  # profile_id is privileged
             ]
 
             for case_name, pid, target, expect_blocked in cases:
@@ -153,10 +221,10 @@ class TestR108SecurityMatrix(unittest.TestCase):
                         label="Test",
                         field_mappings=[
                             FieldMapping(source_path="src", target_path=target)
-                        ]
+                        ],
                     )
                     payload = {"src": "val"}
-                    
+
                     if expect_blocked:
                         with self.assertRaises(ValueError) as cm:
                             apply_mapping(profile, payload)
@@ -177,9 +245,13 @@ class TestR108SecurityMatrix(unittest.TestCase):
         """
         cases = [
             ("valid", {"template_id": "t1"}, True),
-            ("missing_req", {"profile_id": "p1"}, False), # Missing template_id
-            ("bad_type", {"template_id": 123}, False), # Expect str
-            ("oversize", {"template_id": "t1", "inputs": {"x": "x" * (256 * 1024 + 100)}}, False),
+            ("missing_req", {"profile_id": "p1"}, False),  # Missing template_id
+            ("bad_type", {"template_id": 123}, False),  # Expect str
+            (
+                "oversize",
+                {"template_id": "t1", "inputs": {"x": "x" * (256 * 1024 + 100)}},
+                False,
+            ),
         ]
 
         for case_name, payload, expect_valid in cases:
@@ -195,27 +267,49 @@ class TestR108SecurityMatrix(unittest.TestCase):
         Matrix: Route Plane vs Profile.
         Rule: Admin/Internal plane requires non-public auth in Public profile.
         """
-        
+
         # Define mock entries
         def mk_entry(plane, auth, method="GET", path="/"):
             return {
                 "method": method,
                 "path": path,
-                "metadata": {
-                    "plane": plane.value,
-                    "auth": auth.value
-                }
+                "metadata": {"plane": plane.value, "auth": auth.value},
             }
-        
+
         unclassified = {"method": "GET", "path": "/unc", "metadata": None}
 
         cases = [
             # Profile | Entry | Function | Expect Valid
-            ("local_admin_pub", "local", mk_entry(RoutePlane.ADMIN, AuthTier.PUBLIC), True), # Local allows all
-            ("pub_admin_pub", "public", mk_entry(RoutePlane.ADMIN, AuthTier.PUBLIC), False), # Violation
-            ("pub_admin_admin", "public", mk_entry(RoutePlane.ADMIN, AuthTier.ADMIN), True), # Protected OK
-            ("pub_user_pub", "public", mk_entry(RoutePlane.USER, AuthTier.PUBLIC), True), # User plane public OK
-            ("hard_internal_pub", "hardened", mk_entry(RoutePlane.INTERNAL, AuthTier.PUBLIC), False),
+            (
+                "local_admin_pub",
+                "local",
+                mk_entry(RoutePlane.ADMIN, AuthTier.PUBLIC),
+                True,
+            ),  # Local allows all
+            (
+                "pub_admin_pub",
+                "public",
+                mk_entry(RoutePlane.ADMIN, AuthTier.PUBLIC),
+                False,
+            ),  # Violation
+            (
+                "pub_admin_admin",
+                "public",
+                mk_entry(RoutePlane.ADMIN, AuthTier.ADMIN),
+                True,
+            ),  # Protected OK
+            (
+                "pub_user_pub",
+                "public",
+                mk_entry(RoutePlane.USER, AuthTier.PUBLIC),
+                True,
+            ),  # User plane public OK
+            (
+                "hard_internal_pub",
+                "hardened",
+                mk_entry(RoutePlane.INTERNAL, AuthTier.PUBLIC),
+                False,
+            ),
             ("hard_unclassified", "hardened", unclassified, False),
         ]
 
@@ -236,86 +330,94 @@ class TestR108SecurityMatrix(unittest.TestCase):
         # WP1 says "canonical matrix definitions".
         # We should test the logic in `verify_signature` of RegistryQuarantine (which delegates to TrustRootStore).
         # We will mock `TrustRootStore.verify_signature` or `RegistryQuarantineStore.trust_root_store`.
-        
-        # Actually, let's test `TrustRootStore.verify_signature` logic itself if possible, 
+
+        # Actually, let's test `TrustRootStore.verify_signature` logic itself if possible,
         # mocking the low-level crypto? Or using `registry_quarantine.py` logic which handles error mapping.
-        
+
         # Let's instantiate TrustRootStore with a temp dir and mock methods.
         store = TrustRootStore(state_dir=".")
-        
+
         # Mock _HAS_CRYPTO to True for logic testing
         with patch("services.registry_quarantine._HAS_CRYPTO", True):
             # We will patch `serialization` and `Ed25519PublicKey` to mock crypto validation results
             with patch("services.registry_quarantine.serialization") as mock_ser:
                 mock_key = MagicMock()
                 mock_ser.load_pem_public_key.return_value = mock_key
-                mock_key.__class__ = MagicMock() # Hack to pass isinstance check? 
+                mock_key.__class__ = MagicMock()  # Hack to pass isinstance check?
                 # The code checks `if not isinstance(public_key, Ed25519PublicKey): continue`
                 # We need to export Ed25519PublicKey or patch it.
                 # It is imported in the function scope? No, module level try-import.
-                
+
                 # Easier approach: Mock `verify_signature` of `TrustRootStore` when testing `RegistryQuarantineStore` flows,
                 # OR Mock `get_active_roots` and `public_key.verify`.
-                
+
                 # Let's test `TrustRootStore.verify_signature` matrix logic.
-                
+
                 # To pass `isinstance` check without real crypto lib (if missing), we might struggle.
                 # If crypto is present, we can use real keys?
                 # Assuming crypto IS present (dev environment). If not, tests skip or mock harder.
-                
+
                 # Let's simple-mock `get_active_roots` and the verification loop manually?
                 # The function is `verify_signature`.
-                
+
                 # Scenarios:
                 # 1. No Active Roots -> Fail
                 # 2. Key Found, Verify OK -> Pass
                 # 3. Key Found, Verify Fail -> Fail
                 # 4. Key Revoked -> Fail (Immediate)
-                
+
                 store.get_active_roots = MagicMock(return_value=[])
-                
+
                 # Case 1: No roots
-                ok, msg = store.verify_signature(b"data", "c2ln") # valid b64
+                ok, msg = store.verify_signature(b"data", "c2ln")  # valid b64
                 self.assertFalse(ok)
                 self.assertIn("No active trust roots", msg)
-                
+
                 # Case 2: Revoked
-                revoked_root = TrustRoot(key_id="k1", public_key_pem="pem", revoked=True, revocation_reason="stolen")
+                revoked_root = TrustRoot(
+                    key_id="k1",
+                    public_key_pem="pem",
+                    revoked=True,
+                    revocation_reason="stolen",
+                )
                 store._roots = {"k1": revoked_root}
                 ok, msg = store.verify_signature(b"data", "c2ln", key_id="k1")
                 self.assertFalse(ok)
                 self.assertIn("revoked", msg)
-                
+
                 # Case 3 Unknown Key
                 ok, msg = store.verify_signature(b"data", "c2ln", key_id="unknown")
                 self.assertFalse(ok)
                 self.assertIn("Unknown key", msg)
-                
+
                 # Case 4 Active Root (Mock crypto verify)
                 active_root = TrustRoot(key_id="k2", public_key_pem="pem")
                 store.get_active_roots = MagicMock(return_value=[active_root])
-                
-                # To allow the code to reach `.verify()`, we need `load_pem_public_key` to return a mock 
+
+                # To allow the code to reach `.verify()`, we need `load_pem_public_key` to return a mock
                 # that passes `isinstance(pk, Ed25519PublicKey)`.
                 # This requires patching `Ed25519PublicKey` in the service module.
-                with patch("services.registry_quarantine.Ed25519PublicKey", create=True) as MockAlgo:
+                with patch(
+                    "services.registry_quarantine.Ed25519PublicKey", create=True
+                ) as MockAlgo:
                     mock_ser.load_pem_public_key.return_value = MockAlgo()
                     # This mock instance will pass `isinstance(x, MockAlgo)`? No, `isinstance` checks class.
                     # We need `isinstance(obj, ServiceExpectedClass)`.
                     # We patched the class inside the service.
-                    
+
                     mock_pkey = mock_ser.load_pem_public_key.return_value
-                    
+
                     # Subcase: Verify succeeds
-                    mock_pkey.verify.return_value = None # returns None on success
-                    ok, msg = store.verify_signature(b"data", "c2ln") # valid b64
+                    mock_pkey.verify.return_value = None  # returns None on success
+                    ok, msg = store.verify_signature(b"data", "c2ln")  # valid b64
                     # This assumes we patched correct class.
                     # If this is too brittle, we verify `verify_signature` logic flow only.
-                    self.assertTrue(True) # Verified logic flow via reading code :) 
-                    
+                    self.assertTrue(True)  # Verified logic flow via reading code :)
+
                     # We can't easily mock the crypto class check without more setup.
                     # But the Matrix Logic (Revoked/Unknown/NoRoots) is covered above.
                     pass
 
-if __name__ == '__main__':
+
+if __name__ == "__main__":
     unittest.main()

tests/test_r109_contract_parity.py

@@ -1,16 +1,20 @@
-
-import unittest
-import tempfile
-import shutil
-import os
 import json
+import os
+import shutil
+import tempfile
 import time
+import unittest
 
 # Target Services (Real implementations)
-from services.bridge_token_lifecycle import BridgeTokenStore, TokenStatus, BridgeScope
-from services.registry_quarantine import RegistryQuarantineStore, RegistryQuarantineError, QuarantineState
+from services.bridge_token_lifecycle import BridgeScope, BridgeTokenStore, TokenStatus
+from services.registry_quarantine import (
+    QuarantineState,
+    RegistryQuarantineError,
+    RegistryQuarantineStore,
+)
 from services.webhook_mapping import BUILTIN_PROFILES, apply_mapping
 
+
 class TestR109ContractParity(unittest.TestCase):
     """
     R109: Mock-to-Contract Migration & Parity Checks.
@@ -36,11 +40,11 @@ class TestR109ContractParity(unittest.TestCase):
         Contract: BridgeTokenStore persists headers to disk and survives reload.
         """
         store = BridgeTokenStore(state_dir=self.test_dir)
-        
+
         # Act 1: Issue
         t1 = store.issue_token("dev-1", ttl_sec=3600)
         self.assertTrue(t1.token_id.startswith("bt_"))
-        
+
         # Act 2: Verify Persistence (Reload from disk)
         store2 = BridgeTokenStore(state_dir=self.test_dir)
         tokens = store2.list_tokens()
@@ -57,12 +61,12 @@ class TestR109ContractParity(unittest.TestCase):
         json_path = os.path.join(self.test_dir, "bridge_tokens.json")
         with open(json_path, "w") as f:
             f.write("{invalid-json...")
-            
+
         # Init store
         store = BridgeTokenStore(state_dir=self.test_dir)
         # Should be empty, logging error (which we don't capture here but verify state)
         self.assertEqual(len(store.list_tokens()), 0)
-        
+
         # Should be usable (able to issue new tokens)
         t_new = store.issue_token("dev-recovery")
         self.assertTrue(t_new)
@@ -78,13 +82,13 @@ class TestR109ContractParity(unittest.TestCase):
         # Ensure flag is OFF
         if "OPENCLAW_ENABLE_REGISTRY_SYNC" in os.environ:
             del os.environ["OPENCLAW_ENABLE_REGISTRY_SYNC"]
-            
+
         store = RegistryQuarantineStore(state_dir=self.test_dir)
-        
+
         # Verify read operations might work (list_entries) but write operations FAIL
         # Actually `list_entries` doesn't check `_require_enabled` in code I read?
         # Let's check `register_fetch` which does `self._require_enabled()`.
-        
+
         with self.assertRaises(RegistryQuarantineError) as cm:
             store.register_fetch("pkg", "1.0", "http://src", "sha")
         self.assertIn("disabled", str(cm.exception))
@@ -95,9 +99,9 @@ class TestR109ContractParity(unittest.TestCase):
         """
         os.environ["OPENCLAW_ENABLE_REGISTRY_SYNC"] = "1"
         store = RegistryQuarantineStore(state_dir=self.test_dir)
-        
+
         store.register_fetch("pkg-a", "1.0", "http://a", "sha256_hash")
-        
+
         # Reload
         store2 = RegistryQuarantineStore(state_dir=self.test_dir)
         entry = store2.get_entry("pkg-a", "1.0")
@@ -116,25 +120,27 @@ class TestR109ContractParity(unittest.TestCase):
         sample_github = {
             "repository": {"full_name": "user/repo"},
             "ref": "refs/heads/main",
-            "sender": {"login": "monalisa"}
+            "sender": {"login": "monalisa"},
         }
-        
+
         mapped, warnings = apply_mapping(profile, sample_github)
-        
+
         self.assertEqual(mapped["inputs"]["repo_name"], "user/repo")
         self.assertEqual(mapped["inputs"]["ref"], "refs/heads/main")
         self.assertEqual(mapped["inputs"]["actor"], "monalisa")
-        
+
         # Contract Verification:
         # The built-in github_push profile does NOT include a template_id.
         # This means raw mapping is successful, but it fails canonical schema validation.
         self.assertNotIn("template_id", mapped)
-        
+
         # Verify schema fails as expected
         from services.webhook_mapping import validate_canonical_schema
+
         valid, errors = validate_canonical_schema(mapped)
         self.assertFalse(valid)
         self.assertIn("Missing required field: 'template_id'", errors)
 
-if __name__ == '__main__':
+
+if __name__ == "__main__":
     unittest.main()

tests/test_r111_fuzz_harness.py

@@ -0,0 +1,246 @@
+"""
+R111: Fuzz/property adversarial harness.
+Simple property-based testing harness for security boundaries.
+"""
+
+import ipaddress
+import json
+import logging
+import os
+import random
+import socket
+import string
+import sys
+import time
+import traceback
+from typing import Any, Callable, Dict, List, Optional
+from unittest.mock import patch
+
+# Adjust path to import services
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from services.access_control import is_loopback
+from services.policy_posture import PolicyBundle
+from services.safe_io import SSRFError, validate_outbound_url
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("fuzz_harness")
+
+ARTIFACT_DIR = os.path.join(os.path.dirname(__file__), "fuzz_artifacts")
+os.makedirs(ARTIFACT_DIR, exist_ok=True)
+
+
+class FuzzStrategies:
+    @staticmethod
+    def random_string(min_len=0, max_len=100, chars=string.printable) -> str:
+        return "".join(random.choices(chars, k=random.randint(min_len, max_len)))
+
+    @staticmethod
+    def unsafe_strings() -> List[str]:
+        return [
+            "../../../etc/passwd",
+            "<script>alert(1)</script>",
+            "' OR '1'='1",
+            "\x00",
+            "\uffff",
+            "A" * 10000,  # Buffer overflow candidate
+            "http://169.254.169.254/latest/meta-data/",
+            "file:///etc/passwd",
+            "gopher://localhost:6379/_SLAVEOF...",
+        ]
+
+    @staticmethod
+    def random_json(depth=2) -> Any:
+        if depth == 0:
+            return random.choice(
+                [
+                    FuzzStrategies.random_string(),
+                    random.randint(-1000, 1000),
+                    random.random(),
+                    True,
+                    False,
+                    None,
+                ]
+            )
+
+        # 50% chance of complex structure
+        if random.random() < 0.5:
+            return random.choice(
+                [FuzzStrategies.random_string(), random.randint(-1000, 1000)]
+            )
+
+        is_list = random.random() < 0.5
+        if is_list:
+            return [
+                FuzzStrategies.random_json(depth - 1)
+                for _ in range(random.randint(0, 5))
+            ]
+        else:
+            return {
+                FuzzStrategies.random_string(
+                    1, 10, string.ascii_letters
+                ): FuzzStrategies.random_json(depth - 1)
+                for _ in range(random.randint(0, 5))
+            }
+
+
+class Fuzzer:
+    def __init__(self):
+        self.crashes = []
+        self.start_time = time.time()
+
+    def fuzz_target(
+        self, name: str, target_func: Callable, input_gen: Callable, max_runs=1000
+    ):
+        logger.info(f"Starting fuzzing for target: {name}")
+        for i in range(max_runs):
+            inp = input_gen()
+            try:
+                target_func(inp)
+            except (ValueError, TypeError, SSRFError, json.JSONDecodeError, KeyError):
+                # Expected errors
+                pass
+            except Exception as e:
+                # Unexpected crash
+                logger.error(f"CRASH in {name}: {e}")
+                traceback.print_exc()
+                self._save_crash(name, inp, e)
+
+        logger.info(f"Finished fuzzing {name}. Crashes: {len(self.crashes)}")
+
+    def _save_crash(self, name: str, inp: Any, exception: Exception):
+        filename = f"crash_{name}_{int(time.time()*1000)}.json"
+        path = os.path.join(ARTIFACT_DIR, filename)
+
+        crash_data = {
+            "target": name,
+            "input": str(inp),
+            "exception": str(exception),
+            "traceback": traceback.format_exc(),
+        }
+
+        with open(path, "w") as f:
+            json.dump(crash_data, f, indent=2)
+
+        self.crashes.append(path)
+
+
+# --- Fuzz Targets Wrappers ---
+
+
+def fuzz_url_validation(inp):
+    # CRITICAL: keep DNS resolver stubbed in fuzzing; live DNS makes this harness
+    # non-deterministic and can hang in CI/offline environments.
+    with patch(
+        "services.safe_io.socket.getaddrinfo", side_effect=_deterministic_getaddrinfo
+    ):
+        # Try validation. Should raise SSRFError or return tuple or raise ValueError
+        validate_outbound_url(inp, allow_any_public_host=True)
+
+
+def _deterministic_getaddrinfo(host, port, *_args, **_kwargs):
+    """
+    Fast, offline resolver stub for fuzzing.
+    - Preserve IP-host behavior (private IP inputs should still be blocked).
+    - Map hostname inputs to a fixed public test IP to avoid network dependency.
+    """
+    try:
+        ipaddress.ip_address(host)
+        resolved_ip = host
+    except ValueError:
+        resolved_ip = "93.184.216.34"
+
+    return [
+        (
+            socket.AF_INET,
+            socket.SOCK_STREAM,
+            socket.IPPROTO_TCP,
+            "",
+            (resolved_ip, int(port)),
+        )
+    ]
+
+
+def fuzz_policy_bundle(inp):
+    # Input should be a dict-like structure
+    if isinstance(inp, str):
+        try:
+            inp = json.loads(inp)
+        except:
+            return
+
+    if not isinstance(inp, dict):
+        return
+
+    # Try to parse
+    PolicyBundle.from_dict(inp)
+
+
+def fuzz_is_loopback(inp):
+    if not isinstance(inp, str):
+        return
+    is_loopback(inp)
+
+
+def run_fuzz_suite():
+    fuzzer = Fuzzer()
+
+    # 1. fuzz_url_validation
+    # Mixed random strings and known dangerous payloads
+    def url_gen():
+        if random.random() < 0.2:
+            return random.choice(FuzzStrategies.unsafe_strings())
+        return "http://" + FuzzStrategies.random_string(
+            1, 20, string.ascii_letters + ".:/"
+        )
+
+    fuzzer.fuzz_target(
+        "validate_outbound_url", fuzz_url_validation, url_gen, max_runs=500
+    )
+
+    # 2. fuzz_policy_bundle
+    def bundle_gen():
+        return FuzzStrategies.random_json(depth=3)
+
+    fuzzer.fuzz_target(
+        "PolicyBundle.from_dict", fuzz_policy_bundle, bundle_gen, max_runs=500
+    )
+
+    # 3. fuzz_is_loopback
+    def ip_gen():
+        if random.random() < 0.2:
+            return random.choice(FuzzStrategies.unsafe_strings())
+        # Generate random IP-like strings
+        return ".".join(str(random.randint(0, 300)) for _ in range(4))
+
+    fuzzer.fuzz_target("is_loopback", fuzz_is_loopback, ip_gen, max_runs=500)
+
+    # 4. fuzz_path_normalization
+    from services.safe_io import PathTraversalError, resolve_under_root
+
+    def path_gen():
+        # Mix of valid relative paths, traversals, and absolute paths
+        parts = ["foo", "..", "bar", "//", "\\", "C:", "/etc/passwd", "~", "."]
+        return os.path.join(
+            *[random.choice(parts) for _ in range(random.randint(1, 5))]
+        )
+
+    def fuzz_resolve(inp):
+        try:
+            # use a temp dir as root
+            resolve_under_root("/tmp/safe_root", inp)
+        except (PathTraversalError, ValueError):
+            pass
+
+    fuzzer.fuzz_target("resolve_under_root", fuzz_resolve, path_gen, max_runs=500)
+
+    if fuzzer.crashes:
+        print(f"FAILED: {len(fuzzer.crashes)} crashes detected. See {ARTIFACT_DIR}")
+        sys.exit(1)
+    else:
+        print("SUCCESS: No crashes detected.")
+
+
+if __name__ == "__main__":
+    run_fuzz_suite()

tests/test_r89_provider_resilience.py

@@ -1,45 +1,48 @@
-
-import unittest
 import time
+import unittest
 from unittest.mock import MagicMock, patch
 
 from services.threat_intel_provider import (
-    ResilientProviderWrapper,
     ResilienceConfig,
+    ResilientProviderWrapper,
     ScanResult,
     ScanVerdict,
-    ThreatIntelProvider
+    ThreatIntelProvider,
 )
 
+
 class TestR89ProviderResilience(unittest.TestCase):
-    
+
     def setUp(self):
         self.mock_provider = MagicMock()
         # Default config for testing
         self.config = ResilienceConfig(
             max_retries=1,
-            retry_delay_sec=0.01, # Fast tests
+            retry_delay_sec=0.01,  # Fast tests
             circuit_breaker_threshold=2,
-            circuit_breaker_reset_sec=0.1
+            circuit_breaker_reset_sec=0.1,
         )
         self.wrapper = ResilientProviderWrapper(self.mock_provider, self.config)
 
     def test_transient_failure_retry_success(self):
         """Mock provider fails once, then succeeds. Verify retry."""
         # Setup: Fail 1st call, Succeed 2nd
-        self.mock_provider.check_hash.side_effect = [Exception("Transient"), ScanResult(ScanVerdict.CLEAN)]
-        
+        self.mock_provider.check_hash.side_effect = [
+            Exception("Transient"),
+            ScanResult(ScanVerdict.CLEAN),
+        ]
+
         result = self.wrapper.check_hash("hash1")
-        
+
         self.assertEqual(result.verdict, ScanVerdict.CLEAN)
         self.assertEqual(self.mock_provider.check_hash.call_count, 2)
 
     def test_persistent_failure_max_retries(self):
         """Mock provider always fails. Verify max retries and error result."""
         self.mock_provider.check_hash.side_effect = Exception("Persistent")
-        
+
         result = self.wrapper.check_hash("hash2")
-        
+
         self.assertEqual(result.verdict, ScanVerdict.ERROR)
         # 1 initial + 1 retry (max_retries=1) = 2 calls
         self.assertEqual(self.mock_provider.check_hash.call_count, 2)
@@ -48,18 +51,20 @@ class TestR89ProviderResilience(unittest.TestCase):
     def test_circuit_breaker_trip_and_fail_fast(self):
         """Verify CB trips after threshold and fails fast."""
         self.mock_provider.check_hash.side_effect = Exception("Down")
-        
+
         # Call 1: Fails (retries exhausted) -> +1 failure count
         self.wrapper.check_hash("h1")
         # Call 2: Fails -> +1 failure count (Threshold=2 reached, Trip!)
         self.wrapper.check_hash("h2")
-        
-        self.assertTrue(self.wrapper._cb_open, "CB should be open after threshold failures")
-        
+
+        self.assertTrue(
+            self.wrapper._cb_open, "CB should be open after threshold failures"
+        )
+
         # Call 3: Should Fail Fast (0 calls to provider)
         self.mock_provider.check_hash.reset_mock()
         result = self.wrapper.check_hash("h3")
-        
+
         self.assertEqual(result.verdict, ScanVerdict.ERROR)
         self.assertIn("Circuit Breaker OPEN", result.details)
         self.mock_provider.check_hash.assert_not_called()
@@ -67,25 +72,26 @@ class TestR89ProviderResilience(unittest.TestCase):
     def test_circuit_breaker_recovery(self):
         """Verify CB recovers after reset timeout."""
         self.mock_provider.check_hash.side_effect = Exception("Down")
-        
+
         # Trip CB
         for _ in range(2):
             self.wrapper.check_hash("trip")
         self.assertTrue(self.wrapper._cb_open)
-        
+
         # Wait for reset timeout
-        time.sleep(0.15) 
-        
+        time.sleep(0.15)
+
         # Next call should probe (Half-Open)
         # Setup success
         self.mock_provider.check_hash.side_effect = None
         self.mock_provider.check_hash.return_value = ScanResult(ScanVerdict.CLEAN)
-        
+
         result = self.wrapper.check_hash("probe")
-        
+
         self.assertEqual(result.verdict, ScanVerdict.CLEAN)
         self.assertFalse(self.wrapper._cb_open, "CB should close on success")
         self.assertEqual(self.wrapper._cb_failures, 0)
 
-if __name__ == '__main__':
+
+if __name__ == "__main__":
     unittest.main()

tests/test_s43_threat_intel_gate.py

@@ -1,23 +1,32 @@
-
-import unittest
 import os
+import unittest
 from unittest.mock import MagicMock, patch
 
-from services.threat_intel_gate import ThreatIntelGate, ThreatPolicy, ScanVerdict, ScanResult, get_gate
+from services.threat_intel_gate import (
+    ScanResult,
+    ScanVerdict,
+    ThreatIntelGate,
+    ThreatPolicy,
+    get_gate,
+)
+
 
 class MockProvider:
     def __init__(self, result=ScanVerdict.CLEAN):
         self.result = result
         self.check_hash = MagicMock(return_value=ScanResult(result))
 
+
 class TestS43PolicyMatrix(unittest.TestCase):
-    
+
     def setUp(self):
         # Reset singleton if needed, or just new instance
         self.gate = ThreatIntelGate()
-        self.mock_provider = MockProvider()  # This might need adjustment to match interface
+        self.mock_provider = (
+            MockProvider()
+        )  # This might need adjustment to match interface
         self.gate.set_provider(self.mock_provider)
-        
+
         # Patch os.path.exists to always return True for "dummy"
         self.patcher = patch("os.path.exists", return_value=True)
         self.mock_exists = self.patcher.start()
@@ -28,7 +37,7 @@ class TestS43PolicyMatrix(unittest.TestCase):
     def test_policy_off_allows_all(self):
         """Policy OFF: Allows malicious and error states."""
         self.gate._policy = ThreatPolicy.OFF
-        
+
         # Malicious
         self.mock_provider.result = ScanVerdict.MALICIOUS
         self.mock_provider.check_hash.return_value = ScanResult(ScanVerdict.MALICIOUS)
@@ -43,17 +52,23 @@ class TestS43PolicyMatrix(unittest.TestCase):
     def test_policy_audit_logs_but_allows(self):
         """Policy AUDIT: Allows malicious/error but logs (verified by return/log mock)."""
         self.gate._policy = ThreatPolicy.AUDIT
-        
+
         # Malicious
         self.mock_provider.check_hash.return_value = ScanResult(ScanVerdict.MALICIOUS)
-        with self.assertLogs("ComfyUI-OpenClaw.services.threat_intel_gate", level="WARNING") as cm:
+        with self.assertLogs(
+            "ComfyUI-OpenClaw.services.threat_intel_gate", level="WARNING"
+        ) as cm:
             allowed = self.gate.scan_file("dummy", "test")
             self.assertTrue(allowed, "AUDIT should allow malicious")
-            self.assertTrue(any("AUDIT - Malicious content detected" in m for m in cm.output))
+            self.assertTrue(
+                any("AUDIT - Malicious content detected" in m for m in cm.output)
+            )
 
         # Error
         self.mock_provider.check_hash.return_value = ScanResult(ScanVerdict.ERROR)
-        with self.assertLogs("ComfyUI-OpenClaw.services.threat_intel_gate", level="WARNING") as cm:
+        with self.assertLogs(
+            "ComfyUI-OpenClaw.services.threat_intel_gate", level="WARNING"
+        ) as cm:
             allowed = self.gate.scan_file("dummy", "test")
             self.assertTrue(allowed, "AUDIT should fail-open on error")
             self.assertTrue(any("AUDIT - Provider error" in m for m in cm.output))
@@ -61,7 +76,7 @@ class TestS43PolicyMatrix(unittest.TestCase):
     def test_policy_strict_blocks_threats(self):
         """Policy STRICT: Blocks malicious and errors (Fail-Closed)."""
         self.gate._policy = ThreatPolicy.STRICT
-        
+
         # Malicious -> Block
         self.mock_provider.check_hash.return_value = ScanResult(ScanVerdict.MALICIOUS)
         allowed = self.gate.scan_file("dummy", "test")
@@ -79,5 +94,6 @@ class TestS43PolicyMatrix(unittest.TestCase):
             self.mock_provider.check_hash.return_value = ScanResult(ScanVerdict.CLEAN)
             self.assertTrue(self.gate.scan_file("dummy"), f"Clean failed in {policy}")
 
-if __name__ == '__main__':
+
+if __name__ == "__main__":
     unittest.main()