test: add adaptive adversarial gate with hotspot strictness

scripts/pre_push_checks.sh

@@ -358,9 +358,9 @@ echo "[pre-push] 5/7 R121 retry partition contract"
 MOLTBOT_STATE_DIR="$ROOT_DIR/moltbot_state/_pre_push_retry_partition" \
   "$VENV_PY" scripts/run_unittests.py --module tests.test_r121_retry_partition_contract --enforce-skip-policy tests/skip_policy.json --max-skipped 0
 
-echo "[pre-push] 6/7 R118 adversarial gate (smoke)"
+echo "[pre-push] 6/7 R118 adversarial gate (adaptive: smoke/extended)"
 MOLTBOT_STATE_DIR="$ROOT_DIR/moltbot_state/_pre_push_adversarial" \
-  "$VENV_PY" scripts/run_adversarial_gate.py --profile smoke --seed 42 --artifact-dir .tmp/adversarial
+  "$VENV_PY" scripts/run_adversarial_gate.py --profile auto --seed 42 --artifact-dir .tmp/adversarial
 
 echo "[pre-push] 7/7 npm test (Playwright)"
 npm test

scripts/run_adversarial_gate.py

@@ -8,11 +8,12 @@ Unified entry point for adversarial verification suites:
 Supports two profiles:
 - ``smoke``: Fast, bounded, deterministic -- required on PR/push CI.
 - ``extended``: Deeper coverage -- nightly/manual dispatch.
+- ``auto``: Diff-aware selector; escalates to ``extended`` on high-risk path changes.
 
 Usage:
     python scripts/run_adversarial_gate.py --profile smoke
     python scripts/run_adversarial_gate.py --profile extended --seed 42
-    python scripts/run_adversarial_gate.py --profile smoke --artifact-dir .tmp/adversarial
+    python scripts/run_adversarial_gate.py --profile auto --artifact-dir .tmp/adversarial
 
 CRITICAL: keep fuzz seed/runner deterministic and bounded.
 IMPORTANT: do not downgrade mutation threshold to report-only unless explicitly
@@ -20,14 +21,139 @@ IMPORTANT: do not downgrade mutation threshold to report-only unless explicitly
 """
 
 import argparse
+import fnmatch
 import json
 import os
+import pathlib
 import random
 import re
+import shutil
 import subprocess
 import sys
 import time
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional, Set, Tuple
+
+DEFAULT_HIGH_RISK_PATTERNS = [
+    "services/access_control.py",
+    "services/tenant_context.py",
+    "api/routes.py",
+    "services/security_*.py",
+    "services/startup_profile_gate.py",
+    "services/control_plane.py",
+    "services/endpoint_manifest.py",
+    "services/webhook_auth.py",
+    "services/safe_io.py",
+]
+DEFAULT_MUTATION_ALLOWLIST_PATH = os.path.join(
+    os.path.dirname(__file__), "..", "tests", "mutation_survivor_allowlist.json"
+)
+
+
+def _normalize_rel_path(path: str) -> str:
+    return pathlib.PurePosixPath(path.replace("\\", "/")).as_posix().lstrip("./")
+
+
+def _run_git_diff(base: Optional[str], head: Optional[str]) -> List[str]:
+    if not shutil.which("git"):
+        return []
+    cmd: List[str]
+    if base and head:
+        # Prefer merge-base-aware comparison for branch-based refs.
+        cmd = ["git", "diff", "--name-only", f"{base}...{head}"]
+        result = subprocess.run(cmd, capture_output=True, text=True)
+        if result.returncode != 0:
+            cmd = ["git", "diff", "--name-only", base, head]
+            result = subprocess.run(cmd, capture_output=True, text=True)
+    elif base:
+        cmd = ["git", "diff", "--name-only", f"{base}...HEAD"]
+        result = subprocess.run(cmd, capture_output=True, text=True)
+        if result.returncode != 0:
+            cmd = ["git", "diff", "--name-only", base, "HEAD"]
+            result = subprocess.run(cmd, capture_output=True, text=True)
+    else:
+        # Include uncommitted changes first so local pre-push/full-test runs
+        # can escalate to extended before commit.
+        cmd = ["git", "diff", "--name-only", "HEAD"]
+        result = subprocess.run(cmd, capture_output=True, text=True)
+    if result.returncode != 0:
+        return []
+    files = [_normalize_rel_path(line) for line in result.stdout.splitlines() if line]
+    return sorted(set(files))
+
+
+def _collect_changed_files(
+    diff_base: Optional[str], diff_head: Optional[str]
+) -> Tuple[List[str], str]:
+    files = _run_git_diff(diff_base, diff_head)
+    if files:
+        if diff_base and diff_head:
+            return files, f"git diff {diff_base}...{diff_head}"
+        if diff_base:
+            return files, f"git diff {diff_base}...HEAD"
+        return files, "git diff HEAD (working tree)"
+
+    # Fallback for shallow/no-diff contexts.
+    if shutil.which("git"):
+        mb = subprocess.run(
+            ["git", "merge-base", "origin/main", "HEAD"],
+            capture_output=True,
+            text=True,
+        )
+        if mb.returncode == 0 and mb.stdout.strip():
+            files = _run_git_diff(mb.stdout.strip(), "HEAD")
+            if files:
+                return files, "git diff $(merge-base origin/main HEAD)...HEAD"
+
+        files = _run_git_diff("HEAD~1", "HEAD")
+        if files:
+            return files, "git diff HEAD~1...HEAD"
+
+    return [], "no git diff context"
+
+
+def _filter_high_risk_files(changed_files: List[str], patterns: List[str]) -> List[str]:
+    matched: Set[str] = set()
+    normalized_patterns = [_normalize_rel_path(p) for p in patterns if p.strip()]
+    for f in changed_files:
+        for pattern in normalized_patterns:
+            if fnmatch.fnmatch(f, pattern):
+                matched.add(f)
+                break
+    return sorted(matched)
+
+
+def _resolve_effective_profile(
+    requested_profile: str,
+    diff_base: Optional[str],
+    diff_head: Optional[str],
+    high_risk_patterns: List[str],
+) -> Tuple[str, List[str], List[str], str]:
+    if requested_profile != "auto":
+        return requested_profile, [], [], "explicit profile"
+
+    changed_files, diff_source = _collect_changed_files(diff_base, diff_head)
+    high_risk_changed = _filter_high_risk_files(changed_files, high_risk_patterns)
+    if high_risk_changed:
+        return "extended", changed_files, high_risk_changed, diff_source
+    return "smoke", changed_files, high_risk_changed, diff_source
+
+
+def _load_survivor_allowlist(path: str) -> Set[Tuple[str, int]]:
+    if not path or not os.path.isfile(path):
+        return set()
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        entries = data.get("entries", []) if isinstance(data, dict) else []
+        allowlist: Set[Tuple[str, int]] = set()
+        for entry in entries:
+            file_path = _normalize_rel_path(str(entry.get("file", "")))
+            mutation_index = entry.get("mutation_index")
+            if file_path and isinstance(mutation_index, int):
+                allowlist.add((file_path, mutation_index))
+        return allowlist
+    except Exception:
+        return set()
 
 
 def run_fuzz_suite(seed: int, max_runs: int, artifact_dir: str) -> Dict[str, Any]:
@@ -122,7 +248,13 @@ def run_fuzz_suite(seed: int, max_runs: int, artifact_dir: str) -> Dict[str, Any
     }
 
 
-def run_mutation_suite(threshold: float, artifact_dir: str) -> Dict[str, Any]:
+def run_mutation_suite(
+    threshold: float,
+    artifact_dir: str,
+    *,
+    strict_zero_survivor_files: Optional[List[str]] = None,
+    survivor_allowlist: Optional[Set[Tuple[str, int]]] = None,
+) -> Dict[str, Any]:
     """
     Run R113 mutation test with kill-rate threshold enforcement.
 
@@ -194,6 +326,47 @@ def run_mutation_suite(threshold: float, artifact_dir: str) -> Dict[str, Any]:
             # not emit report). Keep it visible in manifest for CI triage.
             error = "mutation report missing; used score fallback from process output"
 
+        strict_targets = sorted(
+            {
+                _normalize_rel_path(p)
+                for p in (strict_zero_survivor_files or [])
+                if str(p).strip()
+            }
+        )
+        allowlist = survivor_allowlist or set()
+        raw_details = (
+            report_data.get("details", []) if isinstance(report_data, dict) else []
+        )
+        surviving_details: List[Dict[str, Any]] = []
+        for detail in raw_details:
+            if isinstance(detail, dict) and detail.get("status") == "SURVIVED":
+                surviving_details.append(detail)
+
+        strict_violations: List[Dict[str, Any]] = []
+        allowlisted_survivors: List[Dict[str, Any]] = []
+        if strict_targets:
+            strict_set = set(strict_targets)
+            for detail in surviving_details:
+                file_path = _normalize_rel_path(str(detail.get("file", "")))
+                mutation_index = detail.get("mutation_index")
+                key = (
+                    file_path,
+                    mutation_index if isinstance(mutation_index, int) else -1,
+                )
+                if file_path not in strict_set:
+                    continue
+                if key in allowlist:
+                    allowlisted_survivors.append(detail)
+                else:
+                    strict_violations.append(detail)
+            if strict_violations:
+                passed = False
+                strict_err = (
+                    "strict zero-survivor violation on high-risk changed files: "
+                    f"{len(strict_violations)} non-allowlisted survivor(s)"
+                )
+                error = f"{error}; {strict_err}" if error else strict_err
+
         return {
             "suite": "r113_mutation",
             "score": round(score, 2),
@@ -207,6 +380,9 @@ def run_mutation_suite(threshold: float, artifact_dir: str) -> Dict[str, Any]:
             "stderr_tail": result.stderr[-500:] if result.stderr else "",
             "returncode": result.returncode,
             "error": error,
+            "strict_zero_survivor_files": strict_targets,
+            "strict_survivor_violations": strict_violations,
+            "allowlisted_survivors": allowlisted_survivors,
         }
 
     except subprocess.TimeoutExpired:
@@ -228,23 +404,33 @@ def run_mutation_suite(threshold: float, artifact_dir: str) -> Dict[str, Any]:
 
 
 def build_manifest(
-    profile: str,
+    requested_profile: str,
+    effective_profile: str,
     seed: int,
     fuzz_result: Dict[str, Any],
     mutation_result: Dict[str, Any],
     artifact_dir: str,
     elapsed_sec: float,
+    changed_files: Optional[List[str]] = None,
+    high_risk_changed_files: Optional[List[str]] = None,
+    diff_source: str = "",
 ) -> Dict[str, Any]:
     """Build machine-readable JSON manifest for CI artifact upload."""
     overall_passed = fuzz_result["passed"] and mutation_result["passed"]
 
     manifest = {
         "r118_version": "1.0",
-        "profile": profile,
+        "profile_requested": requested_profile,
+        "profile": effective_profile,
         "seed": seed,
         "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
         "elapsed_sec": round(elapsed_sec, 2),
         "decision": "PASS" if overall_passed else "FAIL",
+        "selection": {
+            "diff_source": diff_source,
+            "changed_files": changed_files or [],
+            "high_risk_changed_files": high_risk_changed_files or [],
+        },
         "suites": {
             "r111_fuzz": fuzz_result,
             "r113_mutation": mutation_result,
@@ -252,7 +438,7 @@ def build_manifest(
         "artifact_dir": os.path.abspath(artifact_dir),
         "replay_command": (
             f"python scripts/run_adversarial_gate.py "
-            f"--profile {profile} --seed {seed} "
+            f"--profile {effective_profile} --seed {seed} "
             f"--artifact-dir {artifact_dir}"
         ),
     }
@@ -264,9 +450,12 @@ def main() -> int:
     parser = argparse.ArgumentParser(description="R118 Adversarial Gate Runner")
     parser.add_argument(
         "--profile",
-        choices=["smoke", "extended"],
+        choices=["smoke", "extended", "auto"],
         default="smoke",
-        help="Execution profile (default: smoke)",
+        help=(
+            "Execution profile: smoke, extended, or auto. "
+            "auto escalates to extended when high-risk paths are changed."
+        ),
     )
     parser.add_argument(
         "--seed",
@@ -285,10 +474,58 @@ def main() -> int:
         default=None,
         help="Mutation kill-rate threshold %% (overrides profile default)",
     )
+    parser.add_argument(
+        "--diff-base",
+        default=os.environ.get("OPENCLAW_DIFF_BASE"),
+        help="Optional git diff base ref/sha for auto profile selection.",
+    )
+    parser.add_argument(
+        "--diff-head",
+        default=os.environ.get("OPENCLAW_DIFF_HEAD"),
+        help="Optional git diff head ref/sha for auto profile selection.",
+    )
+    parser.add_argument(
+        "--high-risk-pattern",
+        action="append",
+        default=None,
+        help=(
+            "Additional high-risk path pattern (glob). "
+            "Can be provided multiple times."
+        ),
+    )
+    parser.add_argument(
+        "--mutation-survivor-allowlist",
+        default=DEFAULT_MUTATION_ALLOWLIST_PATH,
+        help=(
+            "JSON allowlist for known equivalent mutation survivors used by "
+            "strict zero-survivor enforcement."
+        ),
+    )
+    parser.add_argument(
+        "--no-enforce-zero-survivor-hotspots",
+        action="store_true",
+        help=(
+            "Disable strict zero-survivor enforcement for changed high-risk files. "
+            "Use only for explicit emergency diagnostics."
+        ),
+    )
     args = parser.parse_args()
 
+    high_risk_patterns = list(DEFAULT_HIGH_RISK_PATTERNS)
+    if args.high_risk_pattern:
+        high_risk_patterns.extend(args.high_risk_pattern)
+
+    effective_profile, changed_files, high_risk_changed_files, diff_source = (
+        _resolve_effective_profile(
+            args.profile,
+            args.diff_base,
+            args.diff_head,
+            high_risk_patterns,
+        )
+    )
+
     # Profile defaults
-    if args.profile == "smoke":
+    if effective_profile == "smoke":
         fuzz_max_runs = 200
         mutation_threshold = args.mutation_threshold or 20.0
     else:  # extended
@@ -299,10 +536,18 @@ def main() -> int:
     artifact_dir = os.path.abspath(args.artifact_dir)
     os.makedirs(artifact_dir, exist_ok=True)
 
-    print(f"R118 Adversarial Gate -- profile={args.profile}, seed={seed}")
+    print(f"R118 Adversarial Gate -- profile={effective_profile}, seed={seed}")
     print(f"  Fuzz: {fuzz_max_runs} runs/target")
     print(f"  Mutation threshold: {mutation_threshold}%")
     print(f"  Artifacts: {artifact_dir}")
+    print(
+        f"  Profile selection: requested={args.profile}, "
+        f"diff_source={diff_source}, high_risk_changes={len(high_risk_changed_files)}"
+    )
+    if high_risk_changed_files:
+        print("  High-risk changed files:")
+        for p in high_risk_changed_files:
+            print(f"    - {p}")
     print("-" * 60)
 
     start = time.time()
@@ -315,7 +560,20 @@ def main() -> int:
 
     # Run mutation suite
     print("\n[R113] Running mutation suite...")
-    mutation_result = run_mutation_suite(mutation_threshold, artifact_dir)
+    strict_zero_survivor_files: List[str] = []
+    if (
+        not args.no_enforce_zero_survivor_hotspots
+        and high_risk_changed_files
+        and effective_profile == "extended"
+    ):
+        strict_zero_survivor_files = list(high_risk_changed_files)
+    survivor_allowlist = _load_survivor_allowlist(args.mutation_survivor_allowlist)
+    mutation_result = run_mutation_suite(
+        mutation_threshold,
+        artifact_dir,
+        strict_zero_survivor_files=strict_zero_survivor_files,
+        survivor_allowlist=survivor_allowlist,
+    )
     mutation_status = "PASS" if mutation_result["passed"] else "FAIL"
     print(
         f"[R113] {mutation_status} -- "
@@ -327,7 +585,16 @@ def main() -> int:
 
     # Build and write manifest
     manifest = build_manifest(
-        args.profile, seed, fuzz_result, mutation_result, artifact_dir, elapsed
+        args.profile,
+        effective_profile,
+        seed,
+        fuzz_result,
+        mutation_result,
+        artifact_dir,
+        elapsed,
+        changed_files=changed_files,
+        high_risk_changed_files=high_risk_changed_files,
+        diff_source=diff_source,
     )
 
     manifest_path = os.path.join(artifact_dir, "adversarial_manifest.json")

scripts/run_full_tests_linux.sh

@@ -186,9 +186,9 @@ echo "[tests] 6/8 Slack integration gates (R124/R125/R117/F57)"
 "$VENV_PY" scripts/run_unittests.py --module tests.test_f57_slack_transport_parity --enforce-skip-policy tests/skip_policy.json --max-skipped 0
 "$VENV_PY" scripts/run_unittests.py --module tests.test_f57_slack_socket_mode_startup --enforce-skip-policy tests/skip_policy.json --max-skipped 0
 
-echo "[tests] 7/8 R118 adversarial gate (smoke)"
+echo "[tests] 7/8 R118 adversarial gate (adaptive: smoke/extended)"
 MOLTBOT_STATE_DIR="$ROOT_DIR/moltbot_state/_local_adversarial" \
-  "$VENV_PY" scripts/run_adversarial_gate.py --profile smoke --seed 42 --artifact-dir .tmp/adversarial
+  "$VENV_PY" scripts/run_adversarial_gate.py --profile auto --seed 42 --artifact-dir .tmp/adversarial
 
 echo "[tests] 8/8 frontend E2E"
 npm test

scripts/run_full_tests_windows.ps1

@@ -254,10 +254,10 @@ Invoke-Checked "Slack integration gates" {
   & $venvPython scripts\run_unittests.py --module tests.test_f57_slack_socket_mode_startup --enforce-skip-policy tests\skip_policy.json --max-skipped 0
 }
 
-Write-Host "[tests] 7/8 R118 adversarial gate (smoke)"
+Write-Host "[tests] 7/8 R118 adversarial gate (adaptive: smoke/extended)"
 $env:MOLTBOT_STATE_DIR = "$root\moltbot_state\_local_adversarial"
-Invoke-Checked "R118 adversarial smoke" {
-  & $venvPython scripts\run_adversarial_gate.py --profile smoke --seed 42 --artifact-dir .tmp\adversarial
+Invoke-Checked "R118 adversarial adaptive" {
+  & $venvPython scripts\run_adversarial_gate.py --profile auto --seed 42 --artifact-dir .tmp\adversarial
 }
 
 Write-Host "[tests] 8/8 frontend E2E"