diff --git a/.github/workflows/distro_tests.yml b/.github/workflows/distro_tests.yml
index 66ca3f03a4..1c02631d70 100644
--- a/.github/workflows/distro_tests.yml
+++ b/.github/workflows/distro_tests.yml
@@ -51,7 +51,10 @@ jobs:
           uv python install 3.12
           uv python pin 3.12
           uv sync --group dev
-          uv run pytest --reruns 2 --exitfirst -o timeout_func_only=true --timeout 1200 --disable-warnings --log-cli-level=INFO .
+          uv run pytest -n $(uv run python bbot/test/worker_count.py) --dist loadgroup --reruns 2 --exitfirst -o timeout_func_only=true --timeout 1200 --disable-warnings --log-cli-level=INFO \
+            bbot/test/test_step_1/test_e2e.py \
+            bbot/test/test_step_1/test_depsinstaller.py \
+            bbot/test/test_step_1/test_command.py
       - name: Upload Debug Logs
         if: always()
         uses: actions/upload-artifact@v7
diff --git a/bbot/core/config/models.py b/bbot/core/config/models.py
index 3f3e85105d..1652f0c13b 100644
--- a/bbot/core/config/models.py
+++ b/bbot/core/config/models.py
@@ -294,6 +294,7 @@ class WebConfig(BaseModel):
     max_sleep_interval_429: Optional[int] = Field(default=None, alias="429_max_sleep_interval")
     debug: Optional[bool] = None
     http_max_redirects: Optional[int] = None
+    http_compare_max_differing_lines: Optional[int] = None
     ssl_verify_target: Optional[bool] = None
     ssl_verify_infrastructure: Optional[bool] = None
 
diff --git a/bbot/core/core.py b/bbot/core/core.py
index 983d489d0f..615e792047 100644
--- a/bbot/core/core.py
+++ b/bbot/core/core.py
@@ -64,21 +64,35 @@ def _prep_multiprocessing(self):
     def home(self):
         return Path(self.config["home"]).expanduser().resolve()
 
+    @property
+    def deps_home(self):
+        """
+        Install state, downloaded tools and libs are keyed on content, not on the scan,
+        so concurrent scans can share them. BBOT_SHARED_DEPS_DIR points them at one
+        location while everything scan-specific stays under home.
+        """
+        shared = os.environ.get("BBOT_SHARED_DEPS_DIR", "").strip()
+        if shared:
+            return Path(shared).expanduser().resolve()
+        return self.home
+
     @property
     def cache_dir(self):
-        return self.home / "cache"
+        return self.deps_home / "cache"
 
     @property
     def tools_dir(self):
-        return self.home / "tools"
+        return self.deps_home / "tools"
 
     @property
     def temp_dir(self):
-        return self.home / "temp"
+        # BBOT_TEMP is the scratch dir for dependency builds and is baked into the
+        # module hash, so it belongs with the other install-once dirs.
+        return self.deps_home / "temp"
 
     @property
     def lib_dir(self):
-        return self.home / "lib"
+        return self.deps_home / "lib"
 
     @property
     def scans_dir(self):
diff --git a/bbot/core/helpers/depsinstaller/installer.py b/bbot/core/helpers/depsinstaller/installer.py
index 82579b4396..7a89ecfb77 100644
--- a/bbot/core/helpers/depsinstaller/installer.py
+++ b/bbot/core/helpers/depsinstaller/installer.py
@@ -2,11 +2,14 @@
 import sys
 import stat
 import json
+import fcntl
 import mmh3
 import orjson
 import shutil
 import getpass
 import logging
+import uuid
+import asyncio
 from time import sleep
 from pathlib import Path
 from threading import Lock
@@ -25,6 +28,8 @@
 
 
 class DepsInstaller:
+    LOCK_POLL_INTERVAL = 0.25
+
     CORE_DEPS = {
         # core BBOT dependencies in the format of binary: package_name
         # each one will only be installed if the binary is not found
@@ -134,8 +139,13 @@ def __init__(self, parent_helper):
         self.data_dir = self.parent_helper.cache_dir / "depsinstaller"
         self.parent_helper.mkdir(self.data_dir)
         self.setup_status_cache = self.data_dir / "setup_status.json"
+        self._setup_status_stamp = None
         self.command_status = self.data_dir / "command_status"
         self.parent_helper.mkdir(self.command_status)
+        self.ansible_artifact_dir = self.data_dir / "ansible_artifacts"
+        self.parent_helper.mkdir(self.ansible_artifact_dir)
+        self.ansible_fact_cache = self.ansible_artifact_dir / "fact_cache"
+        self._discard_corrupt_fact_cache()
         self.setup_status = self.read_setup_status()
 
         # make sure we're using a minimal git config
@@ -152,13 +162,128 @@ def __init__(self, parent_helper):
 
         self.ensure_root_lock = Lock()
 
+        # pip specs covered by the current batch pass; install_module() skips these
+        self._batch_installed = set()
+
     async def install(self, *modules):
+        # Concurrent scans (notably xdist workers) share the deps dir, so serialize
+        # installs across processes: without this they race on the same files and
+        # each pays the full install anyway.
+        # Taking the lock unconditionally serializes every scan behind whichever one
+        # is actually installing, so check first whether there is any work to do.
+        # That check only reads, so it is safe outside the lock.
+        self.setup_status = self.read_setup_status()
+        nothing_to_do = self._all_deps_satisfied(modules)
+        if nothing_to_do is not None:
+            return nothing_to_do
+        # A holder installs every module it needs under a single lock hold, so blocking
+        # on the lock costs the holder's whole chain rather than just the part this
+        # caller needs. Poll instead, rechecking after each publish whether our own deps
+        # have arrived; a caller that needs nothing more then never takes the lock.
+        lock = open(self.data_dir / "install.lock", "w")
+        try:
+            while not self._try_lock(lock):
+                await asyncio.sleep(self.LOCK_POLL_INTERVAL)
+                if not self._setup_status_changed():
+                    continue
+                self.setup_status = self.read_setup_status()
+                satisfied = self._all_deps_satisfied(modules)
+                if satisfied is not None:
+                    return satisfied
+            # another process may have installed deps while we waited for the lock
+            self.setup_status = self.read_setup_status()
+            return await self._install(*modules)
+        finally:
+            lock.close()
+
+    def _all_deps_satisfied(self, modules):
+        """Return (succeeded, failed) if every module is already installed, else None.
+
+        Mirrors the accounting in _install() so the fast path and the locked path
+        agree on which modules count as succeeded.
+        """
+        # retry_failed only revisits modules recorded as failed; the loop below already
+        # declines on any status that is not True, so it is safe on this path.
+        # force_install reinstalls regardless of status, so it can never skip the lock.
+        if self.deps_behavior == "force_install":
+            return None
+        if not self._core_deps_cached():
+            return None
+        succeeded = []
+        for m in modules:
+            if self.deps_behavior == "disable":
+                succeeded.append(m)
+                continue
+            preloaded = self.all_modules_preloaded.get(m)
+            if preloaded is None:
+                return None
+            if not list(chain(*preloaded["deps"].values())):
+                succeeded.append(m)
+                continue
+            if self.setup_status.get(self._module_hash(preloaded), None) is not True:
+                return None
+            satisfied, _ = self._pip_deps_satisfied(preloaded["deps"]["pip"])
+            if not satisfied:
+                return None
+            succeeded.append(m)
+        succeeded.sort()
+        return succeeded, []
+
+    def _core_deps_cached(self):
+        core_deps_hash = str(mmh3.hash(orjson.dumps(self.CORE_DEPS, option=orjson.OPT_SORT_KEYS)))
+        return (self.parent_helper.cache_dir / core_deps_hash).exists()
+
+    def _module_hash(self, preloaded):
+        return self.parent_helper.sha1(
+            json.dumps(preloaded["deps"], sort_keys=True)
+            + self.venv
+            + str(self.parent_helper.deps_home)
+            + os.uname()[1]
+            + str(__version__)
+        ).hexdigest()
+
+    def _try_lock(self, f):
+        try:
+            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            return False
+        return True
+
+    def _setup_status_changed(self):
+        try:
+            stamp = self.setup_status_cache.stat().st_mtime_ns
+        except OSError:
+            stamp = None
+        changed = stamp != self._setup_status_stamp
+        self._setup_status_stamp = stamp
+        return changed
+
+    def _install_order(self, modules):
+        """Order modules cheapest-first so waiters unblock as early as possible.
+
+        The holder installs its whole set under one lock hold and publishes
+        setup_status per module. A pip-only module is already satisfied by
+        _batch_pip_install(), so processing it behind a module that compiles from
+        source strands every waiter on it for the length of that build.
+        """
+
+        def cost(m):
+            preloaded = self.all_modules_preloaded.get(m)
+            if preloaded is None:
+                return 0
+            deps = preloaded["deps"]
+            return 1 if (deps["apt"] or deps["shell"] or deps["ansible"] or deps["common"]) else 0
+
+        return sorted(modules, key=cost)
+
+    async def _install(self, *modules):
         await self.install_core_deps()
         succeeded = []
         failed = []
+        await self._batch_pip_install(modules)
         try:
             notified = False
-            for m in modules:
+            for m in self._install_order(modules):
                 # assume success if we're ignoring dependencies
                 if self.deps_behavior == "disable":
                     succeeded.append(m)
@@ -171,14 +296,8 @@ async def install(self, *modules):
                 preloaded = self.all_modules_preloaded[m]
                 log.debug(f"Installing {m} - Preloaded Deps {preloaded['deps']}")
                 # make a hash of the dependencies and check if it's already been handled
-                # take into consideration whether the venv or bbot home directory changes
-                module_hash = self.parent_helper.sha1(
-                    json.dumps(preloaded["deps"], sort_keys=True)
-                    + self.venv
-                    + str(self.parent_helper.bbot_home)
-                    + os.uname()[1]
-                    + str(__version__)
-                ).hexdigest()
+                # take into consideration whether the venv or the deps directory changes
+                module_hash = self._module_hash(preloaded)
                 success = self.setup_status.get(module_hash, None)
                 dependencies = list(chain(*preloaded["deps"].values()))
                 if len(dependencies) <= 0:
@@ -206,6 +325,9 @@ async def install(self, *modules):
                         self.ensure_root(f'Module "{m}" needs root privileges to install its dependencies.')
                     success = await self.install_module(m)
                     self.setup_status[module_hash] = success
+                    # waiters poll this file to learn their own deps have landed, so
+                    # publish per module instead of once the whole chain is done
+                    self.write_setup_status()
                     if success or self.deps_behavior == "ignore_failed":
                         log.debug(f'Setup succeeded for module "{m}"')
                         succeeded.append(m)
@@ -248,7 +370,13 @@ async def install_module(self, module):
         deps_pip = preloaded["deps"]["pip"]
         deps_pip_constraints = preloaded["deps"]["pip_constraints"]
         if deps_pip:
-            success &= await self.pip_install(deps_pip, constraints=deps_pip_constraints)
+            # _batch_pip_install() just installed these in one resolver pass, so repeating
+            # them per module spawns a no-op pip subprocess costing seconds. Only skip specs
+            # the batch actually covered, so --upgrade still runs for everything else.
+            if all(dep in self._batch_installed for dep in deps_pip):
+                log.debug(f'Pip dependencies for module "{module}" were installed in the batch pass')
+            else:
+                success &= await self.pip_install(deps_pip, constraints=deps_pip_constraints)
 
         # shared/common
         deps_common = preloaded["deps"]["common"]
@@ -358,6 +486,20 @@ def tasks(self, module, tasks):
             log.error(f"Failed to run Ansible tasks for {module}")
         return success
 
+    def _discard_corrupt_fact_cache(self):
+        # an unreadable entry makes every later playbook fail until it expires
+        if not self.ansible_fact_cache.is_dir():
+            return
+        for entry in self.ansible_fact_cache.iterdir():
+            if not entry.is_file():
+                continue
+            try:
+                json.loads(entry.read_text())
+            except Exception:
+                log.debug(f"Discarding corrupt ansible fact cache entry: {entry}")
+                with suppress(OSError):
+                    entry.unlink()
+
     def ansible_run(self, tasks=None, module=None, args=None, ansible_args=None):
         _ansible_args = {"ansible_connection": "local", "ansible_python_interpreter": sys.executable}
         if ansible_args is not None:
@@ -385,9 +527,17 @@ def ansible_run(self, tasks=None, module=None, args=None, ansible_args=None):
         shutil.rmtree(data_dir, ignore_errors=True)
         self.parent_helper.mkdir(data_dir)
 
+        # unique ident keeps each run's events isolated; fact_cache escapes it so
+        # facts are gathered once instead of on every playbook
+        ident = uuid.uuid4().hex
+
         res = run(
             playbook=playbook,
             private_data_dir=str(data_dir),
+            artifact_dir=str(self.ansible_artifact_dir),
+            ident=ident,
+            settings={"fact_cache": "../fact_cache", "fact_cache_type": "jsonfile"},
+            envvars={"ANSIBLE_GATHERING": "smart", "ANSIBLE_CACHE_PLUGIN_TIMEOUT": "86400"},
             host_pattern="localhost",
             inventory={
                 "all": {"hosts": {"localhost": _ansible_args}},
@@ -409,6 +559,8 @@ def ansible_run(self, tasks=None, module=None, args=None, ansible_args=None):
             if e["event"] == "runner_on_failed":
                 err = e["event_data"]["res"]["msg"]
                 break
+        # events are read lazily out of the artifact dir, so only discard it once they are consumed
+        shutil.rmtree(self.ansible_artifact_dir / ident, ignore_errors=True)
         return success, err
 
     def read_setup_status(self):
@@ -420,8 +572,11 @@ def read_setup_status(self):
         return setup_status
 
     def write_setup_status(self):
-        with open(self.setup_status_cache, "w") as f:
+        # readers take no lock, so swap the file in atomically to avoid a torn read
+        tmp = self.setup_status_cache.with_suffix(f".{os.getpid()}.tmp")
+        with open(tmp, "w") as f:
             json.dump(self.setup_status, f)
+        os.replace(tmp, self.setup_status_cache)
 
     def ensure_root(self, message=""):
         self._install_sudo_askpass()
@@ -472,6 +627,63 @@ def _core_dep_satisfied(self, command):
             )
         return bool(self.parent_helper.which(command))
 
+    async def _batch_pip_install(self, modules):
+        """Pre-install every module's pip deps in one resolver pass.
+
+        install_module() still runs per module afterward, but skips any spec this pass
+        covered. Already-satisfied specs are batched rather than dropped so that
+        --upgrade still runs for them, in one resolver pass instead of one subprocess
+        per module. Only modules using the default constraints are batched, since
+        custom constraints must be resolved against their own set.
+        """
+        # scoped to this pass: a later install() call may face a different environment,
+        # so stale coverage must never suppress its per-module installs
+        self._batch_installed = set()
+        if self.deps_behavior == "disable":
+            return
+
+        packages = []
+        seen = set()
+        constrained = set()
+        for m in modules:
+            preloaded = self.all_modules_preloaded.get(m)
+            if not preloaded:
+                continue
+            if not self._needs_install(preloaded):
+                continue
+            deps = preloaded.get("deps", {})
+            if deps.get("pip_constraints"):
+                constrained.update(deps.get("pip", []))
+                continue
+            for dep in deps.get("pip", []):
+                if dep in seen:
+                    continue
+                seen.add(dep)
+                packages.append(dep)
+
+        if len(packages) < 2:
+            return
+
+        log.verbose(f"Batch-installing {len(packages):,} pip packages for {len(modules):,} modules")
+        if await self.pip_install(packages):
+            # a spec shared with a module carrying custom constraints must still be
+            # resolved against those constraints, so it never counts as covered
+            self._batch_installed.update(set(packages) - constrained)
+
+    def _needs_install(self, preloaded):
+        """Whether _install() will actually reach install_module() for this module.
+
+        Batching deps for a module that is already recorded done would install
+        packages the locked path would never have touched.
+        """
+        if self.deps_behavior in ("disable", "force_install"):
+            return self.deps_behavior == "force_install"
+        success = self.setup_status.get(self._module_hash(preloaded), None)
+        if success is True:
+            satisfied, _ = self._pip_deps_satisfied(preloaded["deps"]["pip"])
+            return not satisfied
+        return success is None or self.deps_behavior == "retry_failed"
+
     def _pip_deps_satisfied(self, deps_pip):
         """Check whether a module's pip dependencies are currently installed in this environment.
 
@@ -527,17 +739,22 @@ async def install_core_deps(self):
         # install ansible community.general collection if needed
         overall_success = True
         if not self.setup_status.get("ansible:community.general", False):
-            log.info("Installing Ansible Community General Collection")
-            try:
-                command = ["ansible-galaxy", "collection", "install", "community.general"]
-                await self.parent_helper.run(command, check=True)
-                self.setup_status["ansible:community.general"] = True
-                log.info("Successfully installed Ansible Community General Collection")
-            except CalledProcessError as err:
-                log.warning(
-                    f"Failed to install Ansible Community.General Collection (return code {err.returncode}): {err.stderr}"
+            if self._local_pkg_mgrs_are_builtin():
+                log.debug(
+                    "Skipping Ansible Community General Collection (local package manager is built into ansible-core)"
                 )
-                overall_success = False
+            else:
+                log.info("Installing Ansible Community General Collection")
+                try:
+                    command = ["ansible-galaxy", "collection", "install", "community.general"]
+                    await self.parent_helper.run(command, check=True)
+                    self.setup_status["ansible:community.general"] = True
+                    log.info("Successfully installed Ansible Community General Collection")
+                except CalledProcessError as err:
+                    log.warning(
+                        f"Failed to install Ansible Community.General Collection (return code {err.returncode}): {err.stderr}"
+                    )
+                    overall_success = False
         # only run ansible if there's actually something to install
         if playbook:
             self._install_sudo_askpass()
@@ -554,6 +771,26 @@ async def install_core_deps(self):
             with suppress(Exception):
                 core_deps_cache_file.touch()
 
+    @staticmethod
+    def _local_pkg_mgrs_are_builtin():
+        """True if every package manager present on this host ships with ansible-core.
+
+        The only reason we install community.general is that the `package` action
+        dispatches to a per-manager module (pacman, apk, zypper...) that lives in
+        that collection. ansible-core bundles apt/dnf/dnf5, so on those hosts the
+        galaxy install is a pure no-op download.
+        """
+        try:
+            import ansible.modules
+            from ansible.module_utils.facts.system.pkg_mgr import PKG_MGRS
+        except Exception:
+            return False
+        bundled = Path(ansible.modules.__file__).resolve().parent
+        present = {p["name"] for p in PKG_MGRS if os.path.exists(p["path"])}
+        if not present:
+            return False
+        return all((bundled / f"{name}.py").is_file() for name in present)
+
     def _setup_sudo_cache(self):
         if not self._sudo_cache_setup:
             self._sudo_cache_setup = True
diff --git a/bbot/core/helpers/diff.py b/bbot/core/helpers/diff.py
index 126f122e68..afb0393569 100644
--- a/bbot/core/helpers/diff.py
+++ b/bbot/core/helpers/diff.py
@@ -1,5 +1,6 @@
 import logging
 import xmltodict
+from collections import Counter
 from deepdiff import DeepDiff
 from contextlib import suppress
 from xml.parsers.expat import ExpatError
@@ -8,6 +9,24 @@
 log = logging.getLogger("bbot.core.helpers.diff")
 
 
+def _is_flat_str_sequence(content):
+    return isinstance(content, list) and all(isinstance(i, str) for i in content)
+
+
+def _ordered_diff(content_1, content_2, **kwargs):
+    """DeepDiff with ignore_order, skipping pair-matching for flat string lists.
+
+    Pair-matching is O(n*m) nested DeepDiffs over the differing items, which is
+    what a split("\\n") body of a dynamic page looks like. For sequences whose
+    items are all strings there is nothing to match deeper, and equal-index
+    add/remove pairs are folded back into values_changed by DeepDiff itself, so
+    the result is unchanged.
+    """
+    if _is_flat_str_sequence(content_1) and _is_flat_str_sequence(content_2):
+        kwargs["cutoff_intersection_for_pairs"] = 0
+    return DeepDiff(content_1, content_2, ignore_order=True, view="tree", threshold_to_diff_deeper=0, **kwargs)
+
+
 class _BaselineSnapshot:
     """Lightweight stand-in for a blasthttp Response held by HttpCompare.
 
@@ -99,6 +118,7 @@ def __init__(
         self.headers = headers
         self.cookies = cookies
         self.timeout = 10
+        self.max_differing_lines = self.parent_helper.web_config.get("http_compare_max_differing_lines", 500) or 500
         # Optional async callback fired once with baseline_1 after the baseline is established.
         self.on_baseline_ready = on_baseline_ready
 
@@ -171,9 +191,7 @@ async def _baseline(self):
                 baseline_1_json = baseline_1.text.split("\n")
                 baseline_2_json = baseline_2.text.split("\n")
 
-            ddiff = DeepDiff(
-                baseline_1_json, baseline_2_json, ignore_order=True, view="tree", threshold_to_diff_deeper=0
-            )
+            ddiff = _ordered_diff(baseline_1_json, baseline_2_json)
             self.ddiff_filters = []
 
             for k in ddiff.keys():
@@ -235,17 +253,38 @@ def compare_headers(self, headers_1, headers_2):
                 differing_headers.append(header_value)
         return differing_headers
 
+    def _leaf_counts(self, content):
+        counts = Counter()
+        stack = [("root", content)]
+        while stack:
+            path, node = stack.pop()
+            if path in self.ddiff_filters:
+                continue
+            if isinstance(node, dict):
+                for key, value in node.items():
+                    stack.append((f"{path}[{key!r}]", value))
+            elif isinstance(node, list):
+                for index, value in enumerate(node):
+                    stack.append((f"{path}[{index}]", value))
+            else:
+                counts[str(node)] += 1
+        return counts
+
     def compare_body(self, content_1, content_2):
         if content_1 == content_2:
             return True
 
-        ddiff = DeepDiff(
+        counts_1 = self._leaf_counts(content_1)
+        counts_2 = self._leaf_counts(content_2)
+        differing = counts_1 - counts_2
+        differing.update(counts_2 - counts_1)
+        if sum(differing.values()) > self.max_differing_lines:
+            return False
+
+        ddiff = _ordered_diff(
             content_1,
             content_2,
-            ignore_order=True,
-            view="tree",
             exclude_paths=self.ddiff_filters,
-            threshold_to_diff_deeper=0,
         )
 
         if len(ddiff.keys()) == 0:
diff --git a/bbot/core/helpers/helper.py b/bbot/core/helpers/helper.py
index a4bf2a1572..99af85dded 100644
--- a/bbot/core/helpers/helper.py
+++ b/bbot/core/helpers/helper.py
@@ -7,7 +7,7 @@
 import logging
 from pathlib import Path
 import multiprocessing as mp
-from functools import partial
+from functools import partial, lru_cache
 from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
 
 from . import misc
@@ -30,6 +30,13 @@
 _PR_SET_PDEATHSIG = 1
 
 
+@lru_cache(maxsize=2)
+def _shared_cloudcheck(ssl_verify):
+    from cloudcheck import CloudCheck
+
+    return CloudCheck(verify_ssl=ssl_verify)
+
+
 def _pool_worker_init():
     """Set PR_SET_PDEATHSIG so pool workers die when the parent process dies.
 
@@ -83,10 +90,13 @@ class ConfigAwareHelper:
     def __init__(self, preset):
         self.preset = preset
         self.bbot_home = self.preset.bbot_home
-        self.cache_dir = self.bbot_home / "cache"
-        self.temp_dir = self.bbot_home / "temp"
-        self.tools_dir = self.bbot_home / "tools"
-        self.lib_dir = self.bbot_home / "lib"
+        # deps_home is bbot_home unless BBOT_SHARED_DEPS_DIR redirects the install-once
+        # dirs elsewhere; see BBOTCore.deps_home.
+        self.deps_home = self.preset.deps_home
+        self.cache_dir = self.deps_home / "cache"
+        self.tools_dir = self.deps_home / "tools"
+        self.lib_dir = self.deps_home / "lib"
+        self.temp_dir = self.deps_home / "temp"
         self.scans_dir = self.bbot_home / "scans"
         self.wordlist_dir = Path(__file__).parent.parent.parent / "wordlists"
         self.current_dir = Path.cwd()
@@ -153,10 +163,8 @@ def blasthttp(self):
     @property
     def cloudcheck(self):
         if self._cloudcheck is None:
-            from cloudcheck import CloudCheck
-
             ssl_verify = self.web_config.get("ssl_verify_infrastructure", True)
-            self._cloudcheck = CloudCheck(verify_ssl=ssl_verify)
+            self._cloudcheck = _shared_cloudcheck(ssl_verify)
         return self._cloudcheck
 
     def bloom_filter(self, size):
diff --git a/bbot/core/helpers/interactsh.py b/bbot/core/helpers/interactsh.py
index e178c6db8a..0d892c5129 100644
--- a/bbot/core/helpers/interactsh.py
+++ b/bbot/core/helpers/interactsh.py
@@ -78,6 +78,9 @@ class Interactsh:
         ```
     """
 
+    # grace period for in-flight interactions to reach the server before a final poll
+    SETTLE_INTERVAL = 5
+
     def __init__(self, parent_helper, poll_interval=10):
         self.parent_helper = parent_helper
         self.server = None
@@ -87,6 +90,14 @@ def __init__(self, parent_helper, poll_interval=10):
         self.poll_interval = poll_interval
         self._poll_task = None
 
+    async def settle(self):
+        """Wait for interactions triggered just before the scan ended to reach the server.
+
+        Round-trip latency is a property of the transport, so mocks override this to zero
+        rather than every caller hardcoding a sleep.
+        """
+        await asyncio.sleep(self.SETTLE_INTERVAL)
+
     async def register(self, callback=None):
         """
         Registers the instance with an interact.sh server and sets up polling.
diff --git a/bbot/core/helpers/misc.py b/bbot/core/helpers/misc.py
index 1baf3da426..28b65ea9e9 100644
--- a/bbot/core/helpers/misc.py
+++ b/bbot/core/helpers/misc.py
@@ -1441,6 +1441,23 @@ def search_dict_by_key(key, d):
             yield from search_dict_by_key(key, v)
 
 
+_PLACEHOLDER_REGEX = re.compile(r"#\{(\w+)\}")
+
+
+def _search_format_dict(d, replacements):
+    # replacements travels by reference; re-splatting it at each node rebuilds
+    # the whole mapping once per node, making the walk O(nodes * replacements).
+    if isinstance(d, dict):
+        return {k: _search_format_dict(v, replacements) for k, v in d.items()}
+    elif isinstance(d, list):
+        return [_search_format_dict(v, replacements) for v in d]
+    elif isinstance(d, str):
+        if "#{" not in d:
+            return d
+        return _PLACEHOLDER_REGEX.sub(lambda m: replacements.get(m.group(1), m.group(0)), d)
+    return d
+
+
 def search_format_dict(d, **kwargs):
     """Recursively format string values in a dictionary or list using the provided keyword arguments.
 
@@ -1455,15 +1472,7 @@ def search_format_dict(d, **kwargs):
         >>> search_format_dict({"test": "#{name} is awesome"}, name="keanu")
         {"test": "keanu is awesome"}
     """
-    if isinstance(d, dict):
-        return {k: search_format_dict(v, **kwargs) for k, v in d.items()}
-    elif isinstance(d, list):
-        return [search_format_dict(v, **kwargs) for v in d]
-    elif isinstance(d, str):
-        for find, replace in kwargs.items():
-            find = "#{" + str(find) + "}"
-            d = d.replace(find, replace)
-    return d
+    return _search_format_dict(d, kwargs)
 
 
 def search_dict_values(d, *regexes):
diff --git a/bbot/core/helpers/wordcloud.py b/bbot/core/helpers/wordcloud.py
index a5d9b9aaaf..9f3da3f949 100644
--- a/bbot/core/helpers/wordcloud.py
+++ b/bbot/core/helpers/wordcloud.py
@@ -6,12 +6,18 @@
 from pathlib import Path
 from contextlib import suppress
 from collections import OrderedDict
+from functools import lru_cache
 
 from .misc import tldextract, extract_words
 
 log = logging.getLogger("bbot.core.helpers.wordcloud")
 
 
+@lru_cache(maxsize=8)
+def _load_wordninja_model(wordlist_path):
+    return wordninja.LanguageModel(wordlist_path)
+
+
 class WordCloud(dict):
     """
     WordCloud is a specialized dictionary-like class for storing and aggregating
@@ -485,7 +491,7 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         wordlist_dir = Path(__file__).parent.parent.parent / "wordlists"
         wordninja_dns_wordlist = wordlist_dir / "wordninja_dns.txt.gz"
-        self.model = wordninja.LanguageModel(wordninja_dns_wordlist)
+        self.model = _load_wordninja_model(wordninja_dns_wordlist)
 
     def mutations(self, words, max_mutations=None):
         if isinstance(words, str):
diff --git a/bbot/core/modules.py b/bbot/core/modules.py
index cee81ef6f2..609a822968 100644
--- a/bbot/core/modules.py
+++ b/bbot/core/modules.py
@@ -467,8 +467,17 @@ def preload_cache(self):
     def preload_cache(self, value):
         self._preload_cache = value
         mkdir(self.preload_cache_file.parent)
-        with open(self.preload_cache_file, "wb") as f:
-            pickle.dump(self._preload_cache, f)
+        # concurrent scans can share the cache dir, so write via a private temp file
+        # and rename: a reader either sees the old cache or the new one, never a
+        # half-written pickle
+        tmp_file = self.preload_cache_file.with_suffix(f".{os.getpid()}.tmp")
+        try:
+            with open(tmp_file, "wb") as f:
+                pickle.dump(self._preload_cache, f)
+            os.replace(tmp_file, self.preload_cache_file)
+        finally:
+            with suppress(OSError):
+                tmp_file.unlink()
 
     def save_preload_cache(self):
         self.preload_cache = self.__preloaded
diff --git a/bbot/defaults.yml b/bbot/defaults.yml
index 3a3d53a55c..a3711f6d39 100644
--- a/bbot/defaults.yml
+++ b/bbot/defaults.yml
@@ -130,6 +130,7 @@ web:
   debug: false
   # Maximum number of HTTP redirects to follow
   http_max_redirects: 5
+  http_compare_max_differing_lines: 500
   # Whether to verify SSL certificates for target-directed traffic (probes, crawls, etc.)
   ssl_verify_target: false
   # Whether to verify SSL certificates for non-target traffic (APIs, wordlist downloads, etc.)
diff --git a/bbot/modules/base.py b/bbot/modules/base.py
index 4b911f3d89..33b256007b 100644
--- a/bbot/modules/base.py
+++ b/bbot/modules/base.py
@@ -720,6 +720,9 @@ async def _setup(self, deps_only=False):
                     break
             self.debug(f"Finished setting up module {self.name}")
         except Exception as e:
+            # a raising setup() must hard-fail, otherwise setup_modules() reads the
+            # last successful status and leaves an errored module in the scan
+            status = False
             self.set_error_state(f"Unexpected error during module setup: {e}", critical=True)
             msg = f"{e}"
             self.trace()
diff --git a/bbot/modules/dotnetnuke.py b/bbot/modules/dotnetnuke.py
index b99a7eeca0..8d754f84d3 100644
--- a/bbot/modules/dotnetnuke.py
+++ b/bbot/modules/dotnetnuke.py
@@ -210,8 +210,8 @@ async def cleanup(self):
                 self.warning(f"Interactsh failure: {e}")
 
     async def finish(self):
-        if self.interactsh_instance:
-            await self.helpers.sleep(5)
+        if self.interactsh_instance and self.interactsh_subdomain_tags:
+            await self.interactsh_instance.settle()
             try:
                 for r in await self.interactsh_instance.poll():
                     await self.interactsh_callback(r)
diff --git a/bbot/modules/generic_ssrf.py b/bbot/modules/generic_ssrf.py
index ae0e1baab0..0d040f14e5 100644
--- a/bbot/modules/generic_ssrf.py
+++ b/bbot/modules/generic_ssrf.py
@@ -253,8 +253,8 @@ async def cleanup(self):
                 self.warning(f"Interactsh failure: {e}")
 
     async def finish(self):
-        if self.scan.config.get("interactsh_disable", False) is False:
-            await self.helpers.sleep(5)
+        if self.scan.config.get("interactsh_disable", False) is False and self.interactsh_subdomain_tags:
+            await self.interactsh_instance.settle()
             try:
                 for r in await self.interactsh_instance.poll():
                     await self.interactsh_callback(r)
diff --git a/bbot/modules/host_header.py b/bbot/modules/host_header.py
index 02e9d60764..237e61cb21 100644
--- a/bbot/modules/host_header.py
+++ b/bbot/modules/host_header.py
@@ -61,8 +61,8 @@ async def interactsh_callback(self, r):
                 self.debug("skipping results because subdomain tag was missing")
 
     async def finish(self):
-        if self.scan.config.get("interactsh_disable", False) is False:
-            await self.helpers.sleep(5)
+        if self.scan.config.get("interactsh_disable", False) is False and self.subdomain_tags:
+            await self.interactsh_instance.settle()
             try:
                 for r in await self.interactsh_instance.poll():
                     await self.interactsh_callback(r)
diff --git a/bbot/modules/internal/excavate.py b/bbot/modules/internal/excavate.py
index 0d1651c375..cc5fb7d7af 100644
--- a/bbot/modules/internal/excavate.py
+++ b/bbot/modules/internal/excavate.py
@@ -34,11 +34,38 @@ def find_subclasses(obj, base_class):
     >>> find_subclasses(locals(), A)
     [<class '__main__.B'>, <class '__main__.C'>]
     """
-    subclasses = []
-    for name, member in inspect.getmembers(obj):
-        if inspect.isclass(member) and issubclass(member, base_class) and member is not base_class:
-            subclasses.append(member)
-    return subclasses
+    subclasses = {}
+    for namespace in _member_namespaces(obj):
+        for name, member in namespace.items():
+            if inspect.isclass(member) and issubclass(member, base_class) and member is not base_class:
+                subclasses.setdefault(name, member)
+    return [subclasses[name] for name in sorted(subclasses)]
+
+
+def _snapshot(mapping):
+    # a live module's setup() may be assigning attributes while we read it
+    while True:
+        try:
+            return dict(mapping)
+        except RuntimeError:
+            continue
+
+
+def _member_namespaces(obj):
+    """Yield the namespaces of ``obj`` without invoking attribute descriptors.
+
+    ``inspect.getmembers`` getattrs every name, which fires properties such as
+    ``BaseModule.memory_usage``. That walks the module's ``__dict__`` and blows
+    up with "dictionary changed size during iteration" when the module it is
+    reading is concurrently running its own ``setup()``.
+    """
+    if not isinstance(obj, type):
+        instance_dict = getattr(obj, "__dict__", None)
+        if isinstance(instance_dict, dict):
+            yield _snapshot(instance_dict)
+    classes = obj.__mro__ if isinstance(obj, type) else type(obj).__mro__
+    for cls in classes:
+        yield _snapshot(vars(cls))
 
 
 def _pick_select_value(options_html):
@@ -1372,11 +1399,11 @@ async def setup(self):
         self.yara_preprocess_dict = {}
         self.custom_yara_imports = []
 
-        modules_WEB_PARAMETER = [
-            module_name
-            for module_name, module in self.scan.modules.items()
-            if "WEB_PARAMETER" in module.watched_events
-        ]
+        # snapshot: setup_modules() pops failed modules from scan.modules while
+        # these setups run concurrently, so iterating the live dict can raise
+        scan_modules = list(self.scan.modules.values())
+
+        modules_WEB_PARAMETER = [module.name for module in scan_modules if "WEB_PARAMETER" in module.watched_events]
 
         self.parameter_extraction = bool(modules_WEB_PARAMETER)
         self.speculate_params = self.config.get("speculate_params", False)
@@ -1385,7 +1412,7 @@ async def setup(self):
         # at each YARA form-opening match. Caps worst-case Python re work per form.
         self.max_form_bytes = int(self.config.get("max_form_bytes", 262144))
 
-        for module in self.scan.modules.values():
+        for module in scan_modules:
             if not str(module).startswith("_"):
                 ExcavateRules = find_subclasses(module, ExcavateRule)
                 for e in ExcavateRules:
diff --git a/bbot/modules/lightfuzz/lightfuzz.py b/bbot/modules/lightfuzz/lightfuzz.py
index a2a3b71c5e..b1aac883dc 100644
--- a/bbot/modules/lightfuzz/lightfuzz.py
+++ b/bbot/modules/lightfuzz/lightfuzz.py
@@ -351,9 +351,9 @@ async def cleanup(self):
                 self.warning(f"Interactsh failure: {e}")
 
     async def finish(self):
-        if self.interactsh_instance:
-            self.debug("finish(): sleeping 5s before final interactsh poll")
-            await self.helpers.sleep(5)
+        if self.interactsh_instance and self.interactsh_subdomain_tags:
+            self.debug("finish(): settling before final interactsh poll")
+            await self.interactsh_instance.settle()
             try:
                 results = await self.interactsh_instance.poll()
                 self.debug(f"finish(): interactsh poll returned {len(results)} interaction(s)")
diff --git a/bbot/modules/lightfuzz/submodules/crypto.py b/bbot/modules/lightfuzz/submodules/crypto.py
index 085589c1d5..291a7ea208 100644
--- a/bbot/modules/lightfuzz/submodules/crypto.py
+++ b/bbot/modules/lightfuzz/submodules/crypto.py
@@ -567,6 +567,9 @@ async def padding_oracle_execute(self, original_data, encoding, block_size, cook
                     if char_diffs <= 5:
                         continue
                 differ_count += 1
+                # verdict is already fixed above this threshold; further probes cannot change it
+                if differ_count > block_size:
+                    break
         self.debug(f"padding_oracle_execute: finished loop. differ_count={differ_count}")
         # A padding oracle vulnerability can produce a small number of different responses.
         # The correct \x01 padding byte always differs, but also, multi-byte padding values (\x02\x02, \x03\x03\x03, etc.) can also produce valid padding if the intermediate state randomly aligns. At most 'block_size' of such values are possible.
@@ -648,6 +651,7 @@ async def padding_oracle(self, probe_value, cookies):
                         "context": context,
                     }
                 )
+                return  # Report first matching block size only
 
     async def error_string_search(self, text_dict, baseline_text):
         """
diff --git a/bbot/modules/lightfuzz/submodules/sqli.py b/bbot/modules/lightfuzz/submodules/sqli.py
index 545d54a494..23e944302a 100644
--- a/bbot/modules/lightfuzz/submodules/sqli.py
+++ b/bbot/modules/lightfuzz/submodules/sqli.py
@@ -309,6 +309,8 @@ async def fuzz(self):
                             ),
                         }
                     )
+                    # remaining templates only re-confirm the same injection point
+                    break
                 else:
                     self.verbose(
                         f"Stage 2 rejected {self.event.url}: d_high={d_high:.2f}s, "
diff --git a/bbot/modules/medusa.py b/bbot/modules/medusa.py
index 2be1333f9c..926e25a5e1 100644
--- a/bbot/modules/medusa.py
+++ b/bbot/modules/medusa.py
@@ -60,10 +60,12 @@ class Config(BaseModuleConfig):
         },
         {
             # The git repo will be copied because during build, files and subfolders get created. That prevents the Ansible git module to cache the repo.
+            # remote_src keeps the copy local; without it every file round-trips through the controller.
             "name": "Copy medusa repo",
             "copy": {
                 "src": "#{BBOT_TEMP}/medusa/gitrepo/",
                 "dest": "#{BBOT_TEMP}/medusa/workdir/",
+                "remote_src": True,
             },
         },
         {
diff --git a/bbot/modules/nuclei.py b/bbot/modules/nuclei.py
index a6d223770c..ee16f68390 100644
--- a/bbot/modules/nuclei.py
+++ b/bbot/modules/nuclei.py
@@ -1,14 +1,23 @@
 import asyncio
+import fcntl
 import json
 import os
 import shutil
 import yaml
 from typing import Literal
 from itertools import islice
+from contextlib import suppress
 
 from bbot.modules.base import BaseModule
+from bbot.core.helpers.misc import sha1
 from bbot.core.config.models import BaseModuleConfig, Field
 
+try:
+    # libyaml-backed loader, ~7x faster over the full template set
+    from yaml import CSafeLoader as YamlLoader
+except ImportError:
+    from yaml import SafeLoader as YamlLoader
+
 
 class nuclei(BaseModule):
     watched_events = ["URL"]
@@ -38,6 +47,13 @@ class Config(BaseModuleConfig):
             ),
         )
         etags: str = Field("", description="tags to exclude from the scan")
+        etypes: str = Field(
+            "",
+            description=(
+                "protocol types to exclude from the scan, comma-separated "
+                "(dns, file, http, headless, tcp, workflow, ssl, websocket, whois, code, javascript)"
+            ),
+        )
         budget: int = Field(1, description="Used in budget mode to set the number of allowed requests per host")
         silent: bool = Field(False, description="Don't display nuclei's banner or status messages")
         directory_only: bool = Field(True, description="Filter out 'file' URL event (default True)")
@@ -70,18 +86,8 @@ async def setup(self):
         self.nuclei_templates_dir = self.nuclei_state_dir / "templates"
         self.nuclei_config_dir.mkdir(parents=True, exist_ok=True)
         self.nuclei_cache_dir.mkdir(parents=True, exist_ok=True)
-        await self._update_templates()
-        # nuclei writes its version marker before the tarball finishes extracting,
-        # so a killed update can leave the marker pointing at an empty dir and
-        # every subsequent run reports "up-to-date." Verify and repair once.
-        if not self._templates_installed():
-            self.warning("Nuclei templates appear incomplete; wiping isolated state and re-downloading")
-            shutil.rmtree(self.nuclei_state_dir, ignore_errors=True)
-            self.nuclei_config_dir.mkdir(parents=True, exist_ok=True)
-            self.nuclei_cache_dir.mkdir(parents=True, exist_ok=True)
-            await self._update_templates()
-            if not self._templates_installed():
-                return False, "Failed to install nuclei templates after retry"
+        if not await self._ensure_templates():
+            return False, "Failed to install nuclei templates after retry"
         self.proxy = self.scan.web_config.get("http_proxy", "")
         self.mode = self.config.get("mode")
         self.ratelimit = self.config.get("ratelimit")
@@ -97,6 +103,9 @@ async def setup(self):
         self.etags = self.config.get("etags")
         if self.etags:
             self.info(f"Excluding the following nuclei tags: [{self.etags}]")
+        self.etypes = self.config.get("etypes")
+        if self.etypes:
+            self.info(f"Excluding the following nuclei protocol types: [{self.etypes}]")
         self.severity = self.config.get("severity")
         if self.mode != "severe" and self.severity != "":
             self.info(f"Limiting nuclei templates to the following severities: [{self.severity}]")
@@ -225,6 +234,8 @@ async def execute_nuclei(self, nuclei_input):
             "-stats-json",
             "-retries",
             self.retries,
+            "-timeout",
+            self.scan.http_timeout,
         ]
 
         if self.helpers.system_resolvers:
@@ -240,6 +251,9 @@ async def execute_nuclei(self, nuclei_input):
                 command.append(f"-{cli_option}")
                 command.append(option)
 
+        if self.etypes:
+            command += ["-exclude-type", self.etypes]
+
         if self.scan.config.get("interactsh_disable") is True:
             self.info("Disabling interactsh in accordance with global settings")
             command.append("-no-interactsh")
@@ -344,7 +358,61 @@ def _nuclei_env(self):
         env["XDG_CACHE_HOME"] = str(self.nuclei_cache_dir)
         return env
 
-    async def _update_templates(self):
+    async def _ensure_templates(self):
+        # 13k+ template files extract into one shared dir. Concurrent updaters
+        # (xdist workers, parallel scans) fight over the same tree and each
+        # re-does the other's work, so serialize on a lock beside it. A waiter
+        # that finds the tree already populated skips its own redundant update.
+        # The corruption repair below wipes the tree, so it has to run under the
+        # same lock: wiping while another worker is mid-extract destroys its
+        # output and makes it fail too.
+        uncontended = await self.helpers.run_in_executor_io(self._acquire_template_lock)
+        try:
+            if not uncontended and self._templates_installed():
+                self.info("Nuclei templates already up-to-date")
+                return True
+            outcome = await self._run_template_update()
+            if self._templates_installed():
+                return True
+            # A download that never produced files is a fetch failure, not the
+            # stale-marker corruption the wipe exists to repair. Wiping and
+            # re-downloading just doubles the requests against a source that is
+            # already failing, so only repair when the update claimed success.
+            if outcome == "failure":
+                return False
+            # nuclei writes its version marker before the tarball finishes
+            # extracting, so a killed update can leave the marker pointing at an
+            # empty dir and every subsequent run reports "up-to-date."
+            self.warning("Nuclei templates appear incomplete; wiping isolated state and re-downloading")
+            shutil.rmtree(self.nuclei_state_dir, ignore_errors=True)
+            self.nuclei_config_dir.mkdir(parents=True, exist_ok=True)
+            self.nuclei_cache_dir.mkdir(parents=True, exist_ok=True)
+            await self._run_template_update()
+            return self._templates_installed()
+        finally:
+            self._release_template_lock()
+
+    def _acquire_template_lock(self):
+        lock_path = self.helpers.tools_dir / "nuclei-templates.lock"
+        self._template_lock_file = open(lock_path, "w")
+        try:
+            fcntl.flock(self._template_lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            return True
+        except OSError:
+            fcntl.flock(self._template_lock_file, fcntl.LOCK_EX)
+            return False
+
+    def _release_template_lock(self):
+        lock_file = getattr(self, "_template_lock_file", None)
+        if lock_file is None:
+            return
+        self._template_lock_file = None
+        with suppress(OSError):
+            fcntl.flock(lock_file, fcntl.LOCK_UN)
+        with suppress(OSError):
+            lock_file.close()
+
+    async def _run_template_update(self):
         self.info("Updating Nuclei templates")
         # shield so an outer cancel can't kill the subprocess mid-extract and
         # corrupt the templates dir
@@ -361,6 +429,7 @@ async def _update_templates(self):
             self.info("Nuclei templates already up-to-date")
         else:
             self.warning(f"Failure while updating nuclei templates: {update_results.stderr or '<no stderr>'}")
+        return outcome
 
     # nuclei's success messaging has drifted across releases (installed / updated /
     # downloaded). Match any of them so a future rename doesn't silently downgrade
@@ -400,12 +469,48 @@ def __init__(self, nuclei_module):
         self._yaml_files = {}
         self.templates_dir = nuclei_module.nuclei_templates_dir
         self.yaml_list = self.get_yaml_list()
-        self.budget_paths = self.find_budget_paths(nuclei_module.budget)
-        self.collapsible_templates, self.severity_stats = self.find_collapsible_templates()
+        cached = self._cache_load(nuclei_module.budget)
+        if cached is not None:
+            self.budget_paths, self.collapsible_templates, self.severity_stats = cached
+        else:
+            self.budget_paths = self.find_budget_paths(nuclei_module.budget)
+            self.collapsible_templates, self.severity_stats = self.find_collapsible_templates()
+            self._cache_store(nuclei_module.budget)
+        # the parsed documents are ~340MB and are dead once both passes are done
+        self._yaml_files = {}
 
     def get_yaml_list(self):
         return list(self.templates_dir.rglob("*.yaml"))
 
+    def _cache_key(self, budget):
+        """Identify the template set by (path, size, mtime) so a template update invalidates it."""
+        sig = []
+        for f in sorted(self.yaml_list):
+            st = f.stat()
+            sig.append((str(f), st.st_size, st.st_mtime_ns))
+        return f"nuclei_budget_{sha1(repr((budget, sig))).hexdigest()}"
+
+    def _cache_load(self, budget):
+        with suppress(Exception):
+            raw = self.parent.helpers.cache_get(self._cache_key(budget))
+            if raw:
+                d = json.loads(raw)
+                return d["budget_paths"], d["collapsible_templates"], d["severity_stats"]
+        return None
+
+    def _cache_store(self, budget):
+        with suppress(Exception):
+            self.parent.helpers.cache_put(
+                self._cache_key(budget),
+                json.dumps(
+                    {
+                        "budget_paths": self.budget_paths,
+                        "collapsible_templates": self.collapsible_templates,
+                        "severity_stats": self.severity_stats,
+                    }
+                ),
+            )
+
     # Given the current budget setting, scan all of the templates for paths, sort them by frequency and select the first N (budget) items
     def find_budget_paths(self, budget):
         path_frequency = {}
@@ -484,11 +589,16 @@ def find_collapsible_templates(self):
 
     def parse_yaml(self, yamlfile):
         if yamlfile not in self._yaml_files:
-            with open(yamlfile, "r") as stream:
-                try:
-                    y = yaml.safe_load(stream)
-                    self._yaml_files[yamlfile] = y
-                except yaml.YAMLError as e:
-                    self.parent.warning(f"failed to load yaml file: {e}")
-                    return {}
+            with open(yamlfile, "rb") as stream:
+                raw = stream.read()
+            # a template only contributes if it has an http block; skipping the rest
+            # avoids parsing ~5.3k of 13.6k templates that can never yield a path
+            if not (raw.startswith(b"http:") or b"\nhttp:" in raw):
+                self._yaml_files[yamlfile] = {}
+                return self._yaml_files[yamlfile]
+            try:
+                self._yaml_files[yamlfile] = yaml.load(raw, Loader=YamlLoader)
+            except yaml.YAMLError as e:
+                self.parent.warning(f"failed to load yaml file: {e}")
+                return {}
         return self._yaml_files[yamlfile]
diff --git a/bbot/modules/trufflehog.py b/bbot/modules/trufflehog.py
index ee78e2ce41..13824264b4 100644
--- a/bbot/modules/trufflehog.py
+++ b/bbot/modules/trufflehog.py
@@ -149,6 +149,8 @@ async def execute_trufflehog(self, module, path=None, stdin_data=None):
         command = [
             "trufflehog",
             "--json",
+            # without this, overseer re-execs the binary and pays Go package init twice
+            "--local-dev",
             "--no-update",
         ]
         if self.verified:
diff --git a/bbot/modules/wayback.py b/bbot/modules/wayback.py
index e58f6d9a37..eef2da22a7 100644
--- a/bbot/modules/wayback.py
+++ b/bbot/modules/wayback.py
@@ -562,12 +562,16 @@ async def finish(self):
         skip_extensions.update(e.lower() for e in self.scan.config.get("url_extension_special", []))
 
         # build URL list and mapping back to metadata
+        # entries are consumed here: emitting events re-triggers the finish phase, and
+        # anything left in the cache would be fetched again on every re-entry
         url_metadata = {}
         for cleaned_url, value in list(self._archive_cache.items()):
             if not isinstance(value, tuple):
+                # unpaired entries stay put; a later parent event can still pair them
                 self.debug(f"Skipping unpaired archive entry: {cleaned_url}")
                 continue
             raw_url, parent_event = value
+            del self._archive_cache[cleaned_url]
             ext = get_file_extension(cleaned_url)
             if ext and ext in skip_extensions:
                 self.debug(f"Skipping archive fetch for {raw_url} (extension: .{ext})")
diff --git a/bbot/modules/webbrute.py b/bbot/modules/webbrute.py
index 441848f465..55454f2022 100644
--- a/bbot/modules/webbrute.py
+++ b/bbot/modules/webbrute.py
@@ -1,6 +1,8 @@
 import random
 import string
+import asyncio
 from typing import Union
+from collections import OrderedDict
 
 import blasthttp
 
@@ -45,6 +47,15 @@ class Config(BaseModuleConfig):
     in_scope_only = True
     _module_threads = 4
 
+    # bounds the per-shape baseline cache; each entry pins an HttpCompare and its baseline snapshot
+    _baseline_cache_size = 1000
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        # keyed by (url, prefix, suffix, ext); subclasses that override setup() inherit this
+        self._baseline_cache = OrderedDict()
+        self._baseline_locks = OrderedDict()
+
     async def setup_deps(self):
         self.wordlist = await self.helpers.wordlist(self.config.get("wordlist"))
         return True
@@ -128,16 +139,28 @@ def _build_batch_headers(self):
     # Host-level abort reasons that apply to ALL extensions, not just the one that triggered them.
     HOST_ABORT_REASONS = {"WAF_BLOCK_PAGE", "CONNECTIVITY_ISSUES", "BASELINE_CHANGED_CODES", "RECEIVED_429"}
 
-    async def baseline_fuzz(self, url, exts=None, prefix="", suffix=""):
-        if exts is None:
-            exts = [""]
-        filters = {}
-        host_abort = None
-
-        for ext in exts:
-            if host_abort is not None:
-                filters[ext] = host_abort
-                continue
+    async def _baseline_for(self, url, prefix, suffix, ext):
+        """Establish (or reuse) the baseline for one fuzz shape.
+
+        A baseline characterizes the response to a nonexistent path of a given
+        shape, so it depends only on (url, prefix, suffix, ext). Re-probing the
+        same shape costs two requests plus the 0.5s inter-sample sleep and
+        yields the same filter, so shapes are memoized per module instance and
+        raced callers are collapsed onto one probe.
+        """
+        key = (url, prefix, suffix, ext)
+        lock = self._baseline_locks.get(key)
+        if lock is None:
+            lock = self._baseline_locks[key] = asyncio.Lock()
+            while len(self._baseline_locks) > self._baseline_cache_size:
+                self._baseline_locks.popitem(last=False)
+
+        async with lock:
+            cached = self._baseline_cache.get(key)
+            if cached is not None:
+                self._baseline_cache.move_to_end(key)
+                self.debug(f"reusing baseline for URL [{url}] with ext [{ext}]")
+                return cached, None
 
             self.debug(f"running baseline for URL [{url}] with ext [{ext}]")
 
@@ -157,33 +180,44 @@ async def baseline_fuzz(self, url, exts=None, prefix="", suffix=""):
                 await compare._baseline()
             except HttpCompareError as e:
                 self.warning(f"Could not establish baseline for URL [{url}] ext [{ext}]: {e}")
-                abort = {"abort": True, "reason": "CONNECTIVITY_ISSUES"}
-                filters[ext] = abort
-                host_abort = abort
-                continue
+                return {"abort": True, "reason": "CONNECTIVITY_ISSUES"}, "CONNECTIVITY_ISSUES"
 
             baseline_status = compare.baseline.status_code
 
             if await self.helpers.yara.match(self.waf_yara_rules, compare.baseline.content):
                 self.warning(f"Baseline for URL [{url}] ext [{ext}] returned WAF block page, aborting.")
-                abort = {"abort": True, "reason": "WAF_BLOCK_PAGE"}
-                filters[ext] = abort
-                host_abort = abort
-                continue
+                return {"abort": True, "reason": "WAF_BLOCK_PAGE"}, "WAF_BLOCK_PAGE"
 
             if baseline_status == 429:
                 self.warning(
                     f"Received 429 (Too Many Requests) for URL [{url}]. A WAF or rate limiter is blocking requests, aborting."
                 )
-                abort = {"abort": True, "reason": "RECEIVED_429"}
-                filters[ext] = abort
-                host_abort = abort
-                continue
+                return {"abort": True, "reason": "RECEIVED_429"}, "RECEIVED_429"
 
             if baseline_status == 403:
                 self.warning("All baseline requests received 403. A WAF may be actively blocking traffic.")
 
-            filters[ext] = {"compare": compare}
+            result = {"compare": compare}
+            self._baseline_cache[key] = result
+            while len(self._baseline_cache) > self._baseline_cache_size:
+                self._baseline_cache.popitem(last=False)
+            return result, None
+
+    async def baseline_fuzz(self, url, exts=None, prefix="", suffix=""):
+        if exts is None:
+            exts = [""]
+        filters = {}
+        host_abort = None
+
+        for ext in exts:
+            if host_abort is not None:
+                filters[ext] = host_abort
+                continue
+
+            result, abort_reason = await self._baseline_for(url, prefix, suffix, ext)
+            filters[ext] = result
+            if abort_reason is not None:
+                host_abort = result
 
         return filters
 
diff --git a/bbot/modules/webbrute_shortnames.py b/bbot/modules/webbrute_shortnames.py
index 0883c22d3c..9bd52367a2 100644
--- a/bbot/modules/webbrute_shortnames.py
+++ b/bbot/modules/webbrute_shortnames.py
@@ -47,6 +47,8 @@ class Config(BaseModuleConfig):
 
     supplementary_words = ["html", "ajax", "xml", "json", "api"]
 
+    prefix_index_length = 2
+
     async def generate_templist(self, hint, shortname_type):
         words = await self.helpers.run_in_executor_cpu(self._generate_templist_sync, hint, shortname_type)
         return words, len(words)
@@ -103,13 +105,32 @@ async def setup(self):
         self.rate = self.config.get("rate", 0) or None
         self.concurrency = 50
 
+        prefix_index_length = self.prefix_index_length
+
         class MinimalWordPredictor:
+            # class-level defaults: unpickling restores state without calling __init__
+            prefix_index = {}
+            prefix_index_length = 0
+
             def __init__(self):
                 self.word_frequencies = {}
+                self.prefix_index = {}
+                self.prefix_index_length = 0
+
+            def build_prefix_index(self, prefix_length):
+                index = {}
+                for word in self.word_frequencies:
+                    index.setdefault(word[:prefix_length], []).append(word)
+                self.prefix_index = index
+                self.prefix_index_length = prefix_length
 
             def predict(self, prefix, top_n):
                 prefix = prefix.lower()
-                matches = [(word, freq) for word, freq in self.word_frequencies.items() if word.startswith(prefix)]
+                if self.prefix_index and len(prefix) >= self.prefix_index_length:
+                    candidates = self.prefix_index.get(prefix[: self.prefix_index_length], ())
+                    matches = [(word, self.word_frequencies[word]) for word in candidates if word.startswith(prefix)]
+                else:
+                    matches = [(word, freq) for word, freq in self.word_frequencies.items() if word.startswith(prefix)]
 
                 if not matches:
                     return []
@@ -146,6 +167,11 @@ def find_class(self, module, name):
             unpickler = CustomUnpickler(f)
             self.directory_predictor = unpickler.load()
 
+        # built once here, not lazily: predict() runs concurrently in the cpu
+        # executor, and a lazy build would be repeated by every racing caller
+        for predictor in (self.endpoint_predictor, self.directory_predictor):
+            await self.helpers.run_in_executor_cpu(predictor.build_prefix_index, prefix_index_length)
+
         self.subword_list = []
         if self.find_subwords:
             self.debug("Acquiring shortname subword list")
diff --git a/bbot/scanner/preset/preset.py b/bbot/scanner/preset/preset.py
index ed9e502c6f..6414962565 100644
--- a/bbot/scanner/preset/preset.py
+++ b/bbot/scanner/preset/preset.py
@@ -296,6 +296,10 @@ def __init__(
     def bbot_home(self):
         return Path(self.config.get("home", "~/.bbot")).expanduser().resolve()
 
+    @property
+    def deps_home(self):
+        return self.core.deps_home
+
     @property
     def target(self):
         if self._target is None:
diff --git a/bbot/scanner/scanner.py b/bbot/scanner/scanner.py
index 2379dcd940..93c9baa0db 100644
--- a/bbot/scanner/scanner.py
+++ b/bbot/scanner/scanner.py
@@ -238,6 +238,10 @@ def __init__(
             self.dispatcher = dispatcher
         self.dispatcher.set_scan(self)
 
+        # main loop polls fast when events are flowing, backs off when idle
+        self._main_loop_poll_min = 0.002
+        self._main_loop_poll_max = 0.1
+
         # scope distance
         self.scope_config = self.config.get("scope", {})
         self.scope_search_distance = max(0, int(self.scope_config.get("search_distance", 0)))
@@ -472,6 +476,7 @@ async def async_start(self):
             )
 
             # main scan loop
+            poll_interval = self._main_loop_poll_min
             while 1:
                 # abort if we're aborting
                 if self.aborting:
@@ -484,6 +489,7 @@ async def async_start(self):
                     for e in events:
                         yield e
                     if events:
+                        poll_interval = self._main_loop_poll_min
                         continue
 
                 # break if initialization finished and the scan is no longer active
@@ -492,8 +498,10 @@ async def async_start(self):
                     if not new_activity:
                         self._success = True
                         break
+                    poll_interval = self._main_loop_poll_min
 
-                await asyncio.sleep(0.1)
+                await asyncio.sleep(poll_interval)
+                poll_interval = min(poll_interval * 2, self._main_loop_poll_max)
 
             self._success = True
 
@@ -1208,6 +1216,18 @@ def dns_strings(self):
             self._dns_strings = dns_strings
         return self._dns_strings
 
+    def _generate_dns_regex_patterns(self, pattern):
+        """
+        Generates a list of DNS hostname regex source strings based on the provided pattern.
+
+        Args:
+            pattern (str):
+        Returns:
+            list[str]: A list of regex source strings.
+        """
+
+        return [f"{pattern}{re.escape(t)})" for t in self.dns_strings]
+
     def _generate_dns_regexes(self, pattern):
         """
         Generates a list of compiled DNS hostname regexes based on the provided pattern.
@@ -1220,10 +1240,9 @@ def _generate_dns_regexes(self, pattern):
         """
 
         dns_regexes = []
-        for t in self.dns_strings:
-            regex_pattern = re.compile(f"{pattern}{re.escape(t)})", re.I)
-            log.debug(f"Generated Regex [{regex_pattern.pattern}] for domain {t}")
-            dns_regexes.append(regex_pattern)
+        for regex_pattern in self._generate_dns_regex_patterns(pattern):
+            log.debug(f"Generated Regex [{regex_pattern}]")
+            dns_regexes.append(re.compile(regex_pattern, re.I))
         return dns_regexes
 
     @property
@@ -1245,10 +1264,10 @@ def dns_regexes(self):
     @property
     def dns_regexes_yara(self):
         """
-        Returns a list of DNS hostname regexes formatted specifically for compatibility with YARA rules.
+        Returns a list of DNS hostname regex source strings formatted specifically for compatibility with YARA rules.
         """
         if self._dns_regexes_yara is None:
-            self._dns_regexes_yara = self._generate_dns_regexes(r"(([a-z0-9-]+\.)*")
+            self._dns_regexes_yara = self._generate_dns_regex_patterns(r"(([a-z0-9-]+\.)*")
         return self._dns_regexes_yara
 
     @property
@@ -1256,7 +1275,7 @@ def dns_yara_rules_uncompiled(self):
         if self._dns_yara_rules_uncompiled is None:
             regexes_component_list = []
             for i, r in enumerate(self.dns_regexes_yara):
-                regexes_component_list.append(rf"$dns_name_{i} = /\b{r.pattern}/ nocase")
+                regexes_component_list.append(rf"$dns_name_{i} = /\b{r}/ nocase")
 
             # Chunk the regexes into groups of 10,000
             chunk_size = 10000
diff --git a/bbot/scripts/benchmark_report.py b/bbot/scripts/benchmark_report.py
index a5b06aec92..3f2290901c 100644
--- a/bbot/scripts/benchmark_report.py
+++ b/bbot/scripts/benchmark_report.py
@@ -6,6 +6,7 @@
 a comparison report showing performance differences between them.
 """
 
+import sys
 import json
 import argparse
 import subprocess
@@ -32,20 +33,6 @@ def get_current_branch() -> str:
     return result.stdout.strip()
 
 
-def checkout_branch(branch: str, repo_path: Path = None):
-    """Checkout a git branch, cleaning up generated and modified files first."""
-    # Reset modified tracked files (e.g. uv.lock changed by `uv sync`)
-    print("Resetting modified tracked files before checkout")
-    run_command(["git", "checkout", "--", "."], cwd=repo_path)
-    # Remove untracked files before checkout. Without this, files generated
-    # by one branch's toolchain (e.g. uv.lock from `uv run` on a Poetry
-    # branch) block checkout to a branch that tracks those same files.
-    print("Cleaning untracked files before checkout")
-    run_command(["git", "clean", "-fd"], cwd=repo_path)
-    print(f"Checking out branch: {branch}")
-    run_command(["git", "checkout", branch], cwd=repo_path)
-
-
 def run_benchmarks(output_file: Path, repo_path: Path = None) -> bool:
     """Run benchmarks and save results to JSON file.
 
@@ -82,9 +69,8 @@ def run_benchmarks(output_file: Path, repo_path: Path = None) -> bool:
 
     if result.returncode != 0:
         print(f"Pytest exited with code {result.returncode}")
+        return False
 
-    # pytest-benchmark writes JSON regardless of test failures;
-    # treat the run as successful if the output file has data
     if output_file.exists() and output_file.stat().st_size > 0:
         return True
 
@@ -379,15 +365,13 @@ def generate_report(current_data: Dict, base_data: Dict, current_branch: str, ba
     if not base_data:
         report = f"""## 🚀 Performance Benchmark Report
 
-> ℹ️ **No baseline benchmark data available**
-> 
-> Showing current results for **{current_branch}** only.
+> Results for **{current_branch}**.
 
 """
         current_benchmarks = current_data.get("benchmarks", [])
         if current_benchmarks:
             report += f"""<details>
-<summary>📊 Current Results ({current_branch}) - Click to expand</summary>
+<summary>📊 Results ({current_branch}) - Click to expand</summary>
 
 {generate_benchmark_table(current_benchmarks, "Results")}
 </details>"""
@@ -416,79 +400,49 @@ def generate_report(current_data: Dict, base_data: Dict, current_branch: str, ba
 
 
 def main():
-    parser = argparse.ArgumentParser(description="Compare benchmark performance between git branches")
-    parser.add_argument("--base", required=True, help="Base branch name (e.g., 'main', 'dev')")
-    parser.add_argument("--current", required=True, help="Current branch name (e.g., 'feature-branch', 'HEAD')")
+    parser = argparse.ArgumentParser(description="Run BBOT performance benchmarks and report the results")
     parser.add_argument("--output", type=Path, help="Output markdown file (default: stdout)")
     parser.add_argument("--keep-results", action="store_true", help="Keep intermediate JSON files")
 
     args = parser.parse_args()
 
-    # Get current working directory
     repo_path = Path.cwd()
 
-    # Save original branch to restore later
     try:
-        original_branch = get_current_branch()
-        print(f"Current branch: {original_branch}")
+        branch = get_current_branch()
     except subprocess.CalledProcessError:
-        print("Warning: Could not determine current branch")
-        original_branch = None
+        branch = "HEAD"
 
-    # Create temporary files for benchmark results
     with tempfile.TemporaryDirectory() as temp_dir:
-        temp_path = Path(temp_dir)
-        base_results_file = temp_path / "base_results.json"
-        current_results_file = temp_path / "current_results.json"
-
-        base_data = {}
-        current_data = {}
-
-        try:
-            # Run benchmarks on base branch
-            print(f"\n=== Running benchmarks on base branch: {args.base} ===")
-            checkout_branch(args.base, repo_path)
-            if run_benchmarks(base_results_file, repo_path):
-                base_data = load_benchmark_data(base_results_file)
-
-            # Run benchmarks on current branch
-            print(f"\n=== Running benchmarks on current branch: {args.current} ===")
-            checkout_branch(args.current, repo_path)
-            if run_benchmarks(current_results_file, repo_path):
-                current_data = load_benchmark_data(current_results_file)
-
-            # Generate report
-            print("\n=== Generating comparison report ===")
-            report = generate_report(current_data, base_data, args.current, args.base)
-
-            # Output report
-            if args.output:
-                with open(args.output, "w") as f:
-                    f.write(report)
-                print(f"Report written to {args.output}")
-            else:
-                print("\n" + "=" * 80)
-                print(report)
-
-            # Keep results if requested
-            if args.keep_results:
-                if base_data:
-                    with open("base_benchmark_results.json", "w") as f:
-                        json.dump(base_data, f, indent=2)
-                if current_data:
-                    with open("current_benchmark_results.json", "w") as f:
-                        json.dump(current_data, f, indent=2)
-                print("Benchmark result files saved.")
-
-        finally:
-            # Restore original branch
-            if original_branch:
-                print(f"\nRestoring original branch: {original_branch}")
-                try:
-                    checkout_branch(original_branch, repo_path)
-                except subprocess.CalledProcessError:
-                    print(f"Warning: Could not restore original branch {original_branch}")
+        results_file = Path(temp_dir) / "results.json"
+
+        print(f"\n=== Running benchmarks on {branch} ===")
+        if not run_benchmarks(results_file, repo_path):
+            print("Benchmarks produced no data")
+            return 1
+
+        data = load_benchmark_data(results_file)
+        if not data:
+            print(f"No benchmark data in {results_file}")
+            return 1
+
+        report = generate_report(data, {}, branch, None)
+
+        if args.output:
+            with open(args.output, "w") as f:
+                f.write(report)
+            print(f"Report written to {args.output}")
+        else:
+            print("\n" + "=" * 80)
+            print(report)
+
+        if args.keep_results:
+            with open("benchmark_results.json", "w") as f:
+                json.dump(data, f, indent=2)
+            print("Benchmark result files saved.")
+
+    return 0
 
 
 if __name__ == "__main__":
-    main()
+    sys.exit(main())
diff --git a/bbot/test/conftest.py b/bbot/test/conftest.py
index d7caf4aac5..c1d3819636 100644
--- a/bbot/test/conftest.py
+++ b/bbot/test/conftest.py
@@ -9,9 +9,10 @@
 from pathlib import Path
 from contextlib import suppress
 from pytest_httpserver import HTTPServer
-
+from werkzeug.wrappers import Response
 from bbot.test.worker import (
     BBOT_TEST_DIR,
+    BBOT_TEST_SHARED_DIR,
     HTTPSERVER_ALLINTERFACES_PORT,
     HTTPSERVER_PORT,
     HTTPSERVER_SSL_PORT,
@@ -21,6 +22,39 @@
 from bbot.core.helpers.misc import execute_sync_or_async
 from bbot.core.helpers.interactsh import server_list as interactsh_servers
 
+
+class FastShutdownHTTPServer(HTTPServer):
+    """socketserver polls every 0.5s for a shutdown request, so every teardown paid
+    half a second. The poll interval is the only thing standing between us and an
+    immediate stop, and thread_target is documented as the override point."""
+
+    SHUTDOWN_POLL_INTERVAL = 0.005
+
+    def thread_target(self) -> None:
+        self.server.serve_forever(poll_interval=self.SHUTDOWN_POLL_INTERVAL)
+
+    def respond_nohandler(self, request, extra_message=""):
+        """Serve an empty miss body instead of the upstream diagnostic dump.
+
+        Upstream builds the body from repr(request) plus a rendering of every
+        registered matcher, so the miss body is unique per URL and grows with the
+        handler count. Brute-force modules diff each miss against a baseline miss,
+        and a body that never repeats defeats that comparison: every response looks
+        like a difference and gets DeepDiffed in full. The dump was measurement
+        error, not fidelity.
+
+        The body stays empty rather than carrying placeholder text. Modules dedup
+        HTTP_RESPONSEs on (host, port, body_sha256) and base.py exempts the
+        empty-body hash from that check, so an empty miss is the only constant that
+        does not make every miss URL collide as duplicate content.
+
+        The accumulated assertion is dropped with it. bbot's httpserver fixtures
+        call clear() before check_assertions(), so no-handler assertions were
+        discarded unread and never failed a test.
+        """
+        return Response("", self.no_handler_status_code)
+
+
 # silence stdout + trace
 root_logger = logging.getLogger()
 pytest_debug_file = Path(__file__).parent.parent.parent / "pytest_debug.log"
@@ -38,6 +72,10 @@
 # and the sessionfinish cleanup below would delete a directory still in use.
 test_config["home"] = str(BBOT_TEST_DIR)
 
+# Deps install once into a shared dir instead of once per worker. Scan output stays
+# per-worker; the deps scratch dir does not, because it is baked into the module hash.
+os.environ.setdefault("BBOT_SHARED_DEPS_DIR", str(BBOT_TEST_SHARED_DIR))
+
 os.environ["BBOT_DEBUG"] = "True"
 CORE.logger.log_level = logging.DEBUG
 
@@ -114,12 +152,12 @@ def patched(self, module_name):
 def stop_server(server):
     server.stop()
     while server.is_running():
-        time.sleep(0.1)  # Wait a bit before checking again
+        time.sleep(0.005)
 
 
 @pytest.fixture
 def bbot_httpserver():
-    server = HTTPServer(host="127.0.0.1", port=HTTPSERVER_PORT, threaded=True)
+    server = FastShutdownHTTPServer(host="127.0.0.1", port=HTTPSERVER_PORT, threaded=True)
     server.start()
 
     yield server
@@ -138,7 +176,7 @@ def bbot_httpserver_ssl():
     keyfile = str(current_dir / "testsslkey.pem")
     certfile = str(current_dir / "testsslcert.pem")
     context.load_cert_chain(certfile, keyfile)
-    server = HTTPServer(host="127.0.0.1", port=HTTPSERVER_SSL_PORT, ssl_context=context, threaded=True)
+    server = FastShutdownHTTPServer(host="127.0.0.1", port=HTTPSERVER_SSL_PORT, ssl_context=context, threaded=True)
     server.start()
 
     yield server
@@ -245,7 +283,7 @@ async def patched_request_batch_stream(self, urls, threads=10, **kwargs):
 
 @pytest.fixture
 def bbot_httpserver_allinterfaces():
-    server = HTTPServer(host="0.0.0.0", port=HTTPSERVER_ALLINTERFACES_PORT, threaded=True)
+    server = FastShutdownHTTPServer(host="0.0.0.0", port=HTTPSERVER_ALLINTERFACES_PORT, threaded=True)
     server.start()
 
     yield server
@@ -277,6 +315,10 @@ async def register(self, callback=None):
             self.poll_task = asyncio.create_task(self.poll_loop(callback))
         return "fakedomain.fakeinteractsh.com"
 
+    async def settle(self):
+        # interactions are already queued in-process, so there is no propagation delay to wait out
+        return
+
     async def deregister(self, callback=None):
         await asyncio.sleep(1)
         self.stop = True
diff --git a/bbot/test/test_step_1/test_command.py b/bbot/test/test_step_1/test_command.py
index f8789f8588..dd9b21f7ec 100644
--- a/bbot/test/test_step_1/test_command.py
+++ b/bbot/test/test_step_1/test_command.py
@@ -121,30 +121,30 @@ async def test_command(bbot_scanner):
     # test sudo + existence of environment variables
     await scan1._prep()
     path_parts = os.environ.get("PATH", "").split(":")
-    assert f"{BBOT_TEST_DIR}/tools" in path_parts
+    assert str(scan1.helpers.tools_dir) in path_parts
     run_lines = (await scan1.helpers.run(["env"])).stdout.splitlines()
     assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_lines
     for line in run_lines:
         if line.startswith("PATH="):
             path_parts = line.split("=", 1)[-1].split(":")
-            assert f"{BBOT_TEST_DIR}/tools" in path_parts
+            assert str(scan1.helpers.tools_dir) in path_parts
     run_lines_sudo = (await scan1.helpers.run(["env"], sudo=True)).stdout.splitlines()
     assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_lines_sudo
     for line in run_lines_sudo:
         if line.startswith("PATH="):
             path_parts = line.split("=", 1)[-1].split(":")
-            assert f"{BBOT_TEST_DIR}/tools" in path_parts
+            assert str(scan1.helpers.tools_dir) in path_parts
     run_live_lines = [l async for l in scan1.helpers.run_live(["env"])]
     assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_live_lines
     for line in run_live_lines:
         if line.startswith("PATH="):
             path_parts = line.strip().split("=", 1)[-1].split(":")
-            assert f"{BBOT_TEST_DIR}/tools" in path_parts
+            assert str(scan1.helpers.tools_dir) in path_parts
     run_live_lines_sudo = [l async for l in scan1.helpers.run_live(["env"], sudo=True)]
     assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_live_lines_sudo
     for line in run_live_lines_sudo:
         if line.startswith("PATH="):
             path_parts = line.strip().split("=", 1)[-1].split(":")
-            assert f"{BBOT_TEST_DIR}/tools" in path_parts
+            assert str(scan1.helpers.tools_dir) in path_parts
 
     await scan1._cleanup()
diff --git a/bbot/test/test_step_1/test_depsinstaller.py b/bbot/test/test_step_1/test_depsinstaller.py
index 6d6ab8beee..363525c1db 100644
--- a/bbot/test/test_step_1/test_depsinstaller.py
+++ b/bbot/test/test_step_1/test_depsinstaller.py
@@ -1,3 +1,6 @@
+import time
+import subprocess
+from itertools import chain
 from importlib.metadata import version as installed_version
 
 from ..bbot_fixtures import *
@@ -139,11 +142,13 @@ async def mock_install_core_deps():
         }
 
     try:
-        # first run: nothing is cached yet, so both modules install their deps
+        # first run: nothing is cached yet, so both modules install their deps.
+        # the batch pass covers them in one resolver call, so assert on the set of
+        # specs that reached pip rather than on how many subprocesses carried them
         succeeded, failed = await installer.install(missing_module, present_module)
         assert not failed
         assert succeeded == sorted([missing_module, present_module])
-        assert sorted(pip_installs) == [["bbot-nonexistent-package"], ["pydantic"]]
+        assert sorted(chain(*pip_installs)) == ["bbot-nonexistent-package", "pydantic"]
 
         # second run: the cache is warm, but only pydantic is actually installed.
         # the missing package must be reinstalled instead of silently assumed present
@@ -151,8 +156,163 @@ async def mock_install_core_deps():
         succeeded, failed = await installer.install(missing_module, present_module)
         assert not failed
         assert succeeded == sorted([missing_module, present_module])
-        assert pip_installs == [["bbot-nonexistent-package"]]
+        assert sorted(chain(*pip_installs)) == ["bbot-nonexistent-package"]
     finally:
         for module_name in (missing_module, present_module):
             preloaded.pop(module_name, None)
         await scan._cleanup()
+
+
+@pytest.mark.asyncio
+async def test_depsinstaller_batch_covers_satisfied_deps(monkeypatch, bbot_scanner):
+    """
+    The batch pass must cover every spec it installs, including ones already present in
+    the environment. Leaving those uncovered made each owning module spawn its own no-op
+    pip subprocess, which dominated `bbot --install-all-deps`.
+    """
+    scan = bbot_scanner("127.0.0.1")
+    await scan._prep()
+    installer = scan.helpers.depsinstaller
+
+    pip_installs = []
+
+    async def mock_pip_install(packages, constraints=None):
+        pip_installs.append(list(packages))
+        return True
+
+    async def mock_install_core_deps():
+        return
+
+    monkeypatch.setattr(installer, "pip_install", mock_pip_install)
+    monkeypatch.setattr(installer, "install_core_deps", mock_install_core_deps)
+    monkeypatch.setattr(installer, "setup_status_cache", scan.helpers.temp_filename())
+    monkeypatch.setattr(installer, "setup_status", {})
+
+    preloaded = scan.preset.module_loader._preloaded
+    # pydantic is installed in every test environment, so these are the "satisfied" case
+    # distinct specs so the two modules hash differently; both are already installed,
+    # which is precisely the case the batch pass used to leave uncovered
+    shared_modules = ["deps_batch_a", "deps_batch_b"]
+    shared_specs = {"deps_batch_a": "pydantic", "deps_batch_b": "orjson"}
+    constrained_module = "deps_batch_constrained"
+    for module_name in shared_modules:
+        preloaded[module_name] = {
+            "sudo": False,
+            "deps": {
+                "apt": [],
+                "shell": [],
+                "pip": [shared_specs[module_name]],
+                "pip_constraints": [],
+                "common": [],
+                "ansible": [],
+            },
+        }
+    preloaded[constrained_module] = {
+        "sudo": False,
+        "deps": {
+            "apt": [],
+            "shell": [],
+            "pip": ["pydantic"],
+            "pip_constraints": ["pydantic>=1.0"],
+            "common": [],
+            "ansible": [],
+        },
+    }
+
+    try:
+        succeeded, failed = await installer.install(*shared_modules)
+        assert not failed
+        assert succeeded == sorted(shared_modules)
+        # one resolver pass covers both modules; neither may add a subprocess of its own
+        assert pip_installs == [["pydantic", "orjson"]]
+
+        # a spec shared with a module carrying custom constraints must still be resolved
+        # against those constraints, so the batch may not claim it
+        pip_installs.clear()
+        # install() re-reads the status cache under the lock, so a fresh file is what
+        # actually makes this a cold run
+        monkeypatch.setattr(installer, "setup_status_cache", scan.helpers.temp_filename())
+        monkeypatch.setattr(installer, "setup_status", {})
+        succeeded, failed = await installer.install(*shared_modules, constrained_module)
+        assert not failed
+        assert succeeded == sorted([*shared_modules, constrained_module])
+        # pydantic is claimed by the constrained module, so only orjson may be covered
+        assert installer._batch_installed == {"orjson"}
+        assert ["pydantic"] in pip_installs
+    finally:
+        for module_name in (*shared_modules, constrained_module):
+            preloaded.pop(module_name, None)
+        await scan._cleanup()
+
+
+@pytest.mark.asyncio
+async def test_depsinstaller_waiter_does_not_block_on_unrelated_installs(monkeypatch, bbot_scanner):
+    """
+    A lock holder installs its whole module list under a single hold. A waiter whose own
+    deps land partway through that chain must return as soon as they land, rather than
+    blocking for the remainder of the holder's unrelated work.
+    """
+    scan = bbot_scanner("127.0.0.1")
+    await scan._prep()
+    installer = scan.helpers.depsinstaller
+
+    async def mock_install_core_deps():
+        return
+
+    status_cache = Path(scan.helpers.temp_filename())
+    monkeypatch.setattr(installer, "install_core_deps", mock_install_core_deps)
+    monkeypatch.setattr(installer, "setup_status_cache", status_cache)
+    monkeypatch.setattr(installer, "setup_status", {})
+    monkeypatch.setattr(installer, "_setup_status_stamp", None)
+    monkeypatch.setattr(installer, "_core_deps_cached", lambda: True)
+
+    mine = "deps_test_waiter_mine"
+    preloaded = scan.preset.module_loader._preloaded
+    preloaded[mine] = {
+        "sudo": False,
+        "deps": {
+            "apt": [],
+            "shell": [],
+            "pip": [],
+            "pip_constraints": [],
+            "common": [],
+            "ansible": [{"name": f"{mine} task"}],
+        },
+    }
+    mine_hash = installer._module_hash(preloaded[mine])
+
+    holder_chain_secs = 6
+    lock_path = str(installer.data_dir / "install.lock")
+    # a real second process, because the convoy this guards against is cross-process
+    holder_src = (
+        "import fcntl,json,sys,time\n"
+        "f=open(sys.argv[1],'w')\n"
+        "fcntl.flock(f,fcntl.LOCK_EX)\n"
+        "print('locked',flush=True)\n"
+        "time.sleep(0.5)\n"
+        "open(sys.argv[2],'w').write(json.dumps({sys.argv[3]:True}))\n"
+        f"time.sleep({holder_chain_secs})\n"
+    )
+    holder = subprocess.Popen(
+        [sys.executable, "-c", holder_src, lock_path, str(status_cache), mine_hash],
+        stdout=subprocess.PIPE,
+        text=True,
+    )
+    try:
+        assert holder.stdout.readline().strip() == "locked"
+
+        start = time.time()
+        succeeded, failed = await installer.install(mine)
+        waited = time.time() - start
+
+        assert not failed
+        assert succeeded == [mine]
+        assert waited < holder_chain_secs, (
+            f"waiter blocked {waited:.2f}s on the holder's unrelated work; "
+            f"it should have returned once its own dep was published"
+        )
+    finally:
+        holder.kill()
+        holder.wait()
+        preloaded.pop(mine, None)
+        await scan._cleanup()
diff --git a/bbot/test/test_step_1/test_e2e.py b/bbot/test/test_step_1/test_e2e.py
index ee030b8375..aee0831153 100644
--- a/bbot/test/test_step_1/test_e2e.py
+++ b/bbot/test/test_step_1/test_e2e.py
@@ -8,6 +8,7 @@
 """
 
 import json
+import shutil
 import subprocess
 import sys
 import os
@@ -32,15 +33,22 @@ def _find_repo_root():
 
 @pytest.fixture(scope="module")
 def bbot_venv(tmp_path_factory):
-    """Create a fresh virtualenv and pip-install bbot from the local checkout."""
+    """Create a fresh virtualenv and install bbot from the local checkout."""
     venv_dir = tmp_path_factory.mktemp("bbot_e2e_venv")
     repo_root = _find_repo_root()
 
-    subprocess.check_call([sys.executable, "-m", "venv", str(venv_dir)])
-    pip = str(venv_dir / "bin" / "pip")
-    bbot = str(venv_dir / "bin" / "bbot")
+    # uv reuses its shared wheel cache, which CI already warmed via `uv sync`
+    uv = shutil.which("uv")
+    if uv:
+        subprocess.check_call([uv, "venv", str(venv_dir), "--python", sys.executable, "--seed", "--quiet"])
+        venv_python = str(venv_dir / "bin" / "python")
+        subprocess.check_call([uv, "pip", "install", "-e", repo_root, "--python", venv_python, "--quiet"], timeout=300)
+    else:
+        subprocess.check_call([sys.executable, "-m", "venv", str(venv_dir)])
+        pip = str(venv_dir / "bin" / "pip")
+        subprocess.check_call([pip, "install", "-e", repo_root, "--quiet"], timeout=300)
 
-    subprocess.check_call([pip, "install", "-e", repo_root, "--quiet"], timeout=300)
+    bbot = str(venv_dir / "bin" / "bbot")
     assert os.path.isfile(bbot), f"bbot CLI not found at {bbot}"
     return bbot
 
@@ -106,6 +114,9 @@ def run_bbot(bbot_bin, *args, timeout=180):
     )
 
 
+# bbot_venv is module-scoped but xdist scopes fixtures per worker, so splitting
+# these across workers rebuilds the venv once per worker
+@pytest.mark.xdist_group("e2e_venv")
 class TestE2E:
     def test_install_and_help(self, bbot_venv):
         """bbot installs from local source and -h works."""
diff --git a/bbot/test/test_step_1/test_helpers.py b/bbot/test/test_step_1/test_helpers.py
index d2a8505ba0..ef46648825 100644
--- a/bbot/test/test_step_1/test_helpers.py
+++ b/bbot/test/test_step_1/test_helpers.py
@@ -815,17 +815,81 @@ async def test_async_helpers():
     import random
     from bbot.core.helpers.misc import as_completed
 
+    # sleeps are scaled down; wall time here is sum(sleeps)/max_concurrent,
+    # so full-second sleeps buy nothing but 25s of idling
     async def do_stuff(r):
-        await asyncio.sleep(r)
+        await asyncio.sleep(r / 100)
         return r
 
     random_ints = [random.random() for _ in range(1000)]
     tasks = [do_stuff(r) for r in random_ints]
     results = set()
+    completion_order = []
     async for t in as_completed(tasks):
-        results.add(await t)
+        r = await t
+        results.add(r)
+        completion_order.append(r)
     assert len(results) == 1000
     assert sorted(random_ints) == sorted(results)
+    # results must arrive as they finish, not in submission order
+    assert completion_order != random_ints
+
+    # max_concurrent is an upper bound that actually gets saturated
+    for limit in (1, 5, 20, 50):
+        live = 0
+        peak = 0
+
+        async def probe(r):
+            nonlocal live, peak
+            live += 1
+            peak = max(peak, live)
+            try:
+                await asyncio.sleep(r / 100)
+                return r
+            finally:
+                live -= 1
+
+        vals = [random.random() for _ in range(200)]
+        completed = 0
+        async for t in as_completed([probe(r) for r in vals], max_concurrent=limit):
+            await t
+            completed += 1
+        assert completed == 200
+        assert peak == limit
+
+    # unlimited concurrency schedules everything at once
+    live = 0
+    peak = 0
+
+    async def unlimited(i):
+        nonlocal live, peak
+        live += 1
+        peak = max(peak, live)
+        try:
+            await asyncio.sleep(0)
+            return i
+        finally:
+            live -= 1
+
+    completed = 0
+    async for t in as_completed([unlimited(i) for i in range(200)], max_concurrent=None):
+        await t
+        completed += 1
+    assert completed == 200
+    assert peak == 200
+
+    # a raising coroutine is still yielded, and does not abort the rest
+    async def boom():
+        raise ValueError("boom")
+
+    async def fine(i):
+        await asyncio.sleep(0)
+        return i
+
+    yielded = [t async for t in as_completed([boom(), fine(1), boom(), fine(2)])]
+    assert len(yielded) == 4
+    assert sum(1 for t in yielded if isinstance(t.exception(), ValueError)) == 2
+    assert sorted(t.result() for t in yielded if t.exception() is None) == [1, 2]
 
 
 def test_portparse(helpers):
diff --git a/bbot/test/test_step_1/test_manager_scope_accuracy.py b/bbot/test/test_step_1/test_manager_scope_accuracy.py
index 66d80773bf..7a4b61e472 100644
--- a/bbot/test/test_step_1/test_manager_scope_accuracy.py
+++ b/bbot/test/test_step_1/test_manager_scope_accuracy.py
@@ -129,8 +129,6 @@ async def do_scan(*args, _config={}, _dns_mock={}, scan_callback=None, **kwargs)
         if scan_callback is not None:
             scan_callback(scan)
         output_events = [e async for e in scan.async_start()]
-        # let modules initialize
-        await asyncio.sleep(0.5)
         return (output_events, dummy_module.events, dummy_module_nodupes.events, dummy_graph_output_module.events, dummy_graph_batch_output_module.events)
 
     dns_mock_chain = {
diff --git a/bbot/test/test_step_1/test_modules_basic.py b/bbot/test/test_step_1/test_modules_basic.py
index 60e4080714..2d34ed41eb 100644
--- a/bbot/test/test_step_1/test_modules_basic.py
+++ b/bbot/test/test_step_1/test_modules_basic.py
@@ -345,36 +345,49 @@ async def test_modules_basic_perdomainonly(bbot_scanner, monkeypatch):
         force_start=True,
     )
 
-    await per_domain_scan._prep()
-    await per_domain_scan.setup_modules()
+    # postcheck only reads dedup state off each module, so loading is enough.
+    # _prep() additionally runs setup() on every module, and the ones that dial a
+    # service (rabbitmq, postgres, mysql) each burn their full connect-retry budget.
+    await per_domain_scan.load_modules()
     await per_domain_scan._set_status("RUNNING")
 
-    # ensure that multiple events to the same "host" (schema + host) are blocked and check the per host tracker
-
+    # ensure that a second event under an already-seen domain is deduped away
+    per_domain_seen = []
     for module_name, module in sorted(per_domain_scan.modules.items()):
         monkeypatch.setattr(module, "filter_event", BaseModule(per_domain_scan).filter_event)
 
-        if "URL" in module.watched_events:
-            url_1 = per_domain_scan.make_event(
-                "http://www.evilcorp.com/1", event_type="URL", parent=per_domain_scan.root_event, tags=["status-200"]
-            )
-            url_1.scope_distance = 0
-            url_2 = per_domain_scan.make_event(
-                "http://mail.evilcorp.com/2", event_type="URL", parent=per_domain_scan.root_event, tags=["status-200"]
-            )
-            url_2.scope_distance = 0
-            valid_1, reason_1 = await module._event_postcheck(url_1)
-            valid_2, reason_2 = await module._event_postcheck(url_2)
+        # every per_domain_only module watches DNS_NAME, so probing URL-only never
+        # reached the branch below
+        if "DNS_NAME" in module.watched_events:
+            event_type, host_1, host_2 = "DNS_NAME", "www.evilcorp.com", "mail.evilcorp.com"
+        elif "URL" in module.watched_events:
+            event_type, host_1, host_2 = "URL", "http://www.evilcorp.com/1", "http://mail.evilcorp.com/2"
+        else:
+            continue
+
+        event_1 = per_domain_scan.make_event(
+            host_1, event_type=event_type, parent=per_domain_scan.root_event, tags=["status-200"]
+        )
+        event_1.scope_distance = 0
+        event_2 = per_domain_scan.make_event(
+            host_2, event_type=event_type, parent=per_domain_scan.root_event, tags=["status-200"]
+        )
+        event_2.scope_distance = 0
+        valid_1, reason_1 = await module._event_postcheck(event_1)
+        valid_2, reason_2 = await module._event_postcheck(event_2)
+
+        if module.per_domain_only is True:
+            per_domain_seen.append(module_name)
+            assert valid_1 is True, f"{module_name}: first event rejected ({reason_1})"
+            assert valid_2 is False, f"{module_name}: second event accepted"
+            assert hash("evilcorp.com") in module._incoming_dup_tracker
+            assert reason_2 == "module has already seen it (per_domain_only=True)", f"{module_name}: {reason_2}"
 
-            if module.per_domain_only is True:
-                assert valid_1 is True
-                assert valid_2 is False
-                assert hash("evilcorp.com") in module._per_host_tracker
-                assert reason_2 == "per_domain_only enabled and already seen domain"
+        elif event_type == "URL":
+            assert valid_1 is True, f"{module_name}: {reason_1}"
+            assert valid_2 is True, f"{module_name}: {reason_2}"
 
-            else:
-                assert valid_1 is True
-                assert valid_2 is True
+    assert per_domain_seen, "no per_domain_only modules were exercised"
 
     await per_domain_scan._cleanup()
 
@@ -524,21 +537,24 @@ async def test_module_loading(bbot_scanner):
         config={i: True for i in available_internal_modules if i != "dnsresolve"},
         force_start=True,
     )
-    await scan2._prep()
-    await scan2._set_status("RUNNING")
+    # every assertion below reads class attributes off the instantiated modules, so
+    # loading them is enough. _prep() would additionally run setup() on all ~110,
+    # and the ones that talk to a service (rabbitmq, postgres, mysql) each burn
+    # their full connect-retry budget against a service that isn't there.
+    await scan2.load_modules()
 
     # attributes, descriptions, etc.
+    not_async = []
     for module_name, module in sorted(scan2.modules.items()):
         # flags
         assert module._type in ("internal", "output", "scan")
         # async stuff
-        not_async = []
         for func_name in ("setup", "ping", "filter_event", "handle_event", "finish", "report", "cleanup"):
             f = getattr(module, func_name)
             if not scan2.helpers.is_async_function(f):
                 log.error(f"{f.__qualname__}() is not async")
-                not_async.append(f)
-    assert not any(not_async)
+                not_async.append(f.__qualname__)
+    assert not not_async, f"non-async module methods: {not_async}"
 
     await scan2._cleanup()
 
diff --git a/bbot/test/test_step_1/test_python_api.py b/bbot/test/test_step_1/test_python_api.py
index cb685980ea..fee826ca5e 100644
--- a/bbot/test/test_step_1/test_python_api.py
+++ b/bbot/test/test_step_1/test_python_api.py
@@ -49,7 +49,8 @@ async def test_python_api(clean_default_config):
     bbot_home = str(worker_dir("/tmp/.bbot_python_api_test"))
     scan4 = Scanner("127.0.0.1", config={"home": bbot_home})
     await scan4._prep()
-    assert os.environ["BBOT_TOOLS"] == str(Path(bbot_home) / "tools")
+    assert scan4.helpers.scans_dir == Path(bbot_home) / "scans"
+    assert os.environ["BBOT_TOOLS"] == str(scan4.helpers.tools_dir)
 
     # output modules are additive
     scan5 = Scanner()
@@ -93,7 +94,8 @@ async def test_python_api_sync(clean_default_config):
     bbot_home = str(worker_dir("/tmp/.bbot_python_api_test"))
     scan3 = Scanner("127.0.0.1", config={"home": bbot_home})
     await scan3._prep()
-    assert os.environ["BBOT_TOOLS"] == str(Path(bbot_home) / "tools")
+    assert scan3.helpers.scans_dir == Path(bbot_home) / "scans"
+    assert os.environ["BBOT_TOOLS"] == str(scan3.helpers.tools_dir)
 
 
 def test_python_api_sync_no_pending_tasks():
diff --git a/bbot/test/test_step_1/test_web.py b/bbot/test/test_step_1/test_web.py
index b0d5ee2aad..a7143e119c 100644
--- a/bbot/test/test_step_1/test_web.py
+++ b/bbot/test/test_step_1/test_web.py
@@ -1,9 +1,12 @@
 import re
+import time
 
 from blasthttp import HTTPStatusError
 
 from ..bbot_fixtures import *
 
+from bbot.core.helpers.diff import _ordered_diff
+
 from bbot.test.worker import BBOT_TEST_DIR, HTTPSERVER_HOSTPORT
 
 
@@ -342,7 +345,12 @@ def sync_callback(data):
     assert response2.status_code == 200
     assert any(interactsh_domain2.endswith(f"{s}") for s in server_list)
 
-    await asyncio.sleep(10)
+    # poll until every asserted condition holds, capped at the old fixed budget
+    deadline = time.time() + 10
+    while time.time() < deadline:
+        if sync_correct_url and async_correct_url:
+            break
+        await asyncio.sleep(0.1)
 
     data_list = await interactsh_client.poll()
     data_list2 = await interactsh_client2.poll()
@@ -423,6 +431,116 @@ async def test_web_http_compare(blasthttp_mock, bbot_scanner):
     for mode in ("getparam", "header", "cookie"):
         assert await compare_helper.canary_check("http://www.example.com", mode=mode) is True
 
+    assert compare_helper.compare_body(["a", "b", "c"], ["a", "b", "c"]) is True
+    assert compare_helper.compare_body(["a", "b", "c"], ["a", "b", "x"]) is False
+    assert compare_helper.compare_body(["a", "b", "c"], ["c", "b", "a"]) is True
+    assert compare_helper.compare_body({"a": 1}, {"a": 2}) is False
+    assert compare_helper.compare_body({"a": 1}, {"a": 1}) is True
+
+    compare_helper.max_differing_lines = 500
+    base = [f"line {i}" for i in range(20000)]
+    similar = [f"line {i} X" if i < 5000 else f"line {i}" for i in range(20000)]
+    start = time.monotonic()
+    assert compare_helper.compare_body(base, similar) is False
+    assert (time.monotonic() - start) < 5
+
+    config_scan = bbot_scanner(config={"web": {"http_compare_max_differing_lines": 42}})
+    await config_scan._prep()
+    config_helper = config_scan.helpers.http_compare("http://www.example.com")
+    assert config_helper.max_differing_lines == 42
+    await config_scan._cleanup()
+
+    await scan._cleanup()
+
+
+@pytest.mark.asyncio
+async def test_web_http_compare_filtered_lines_not_counted(blasthttp_mock, bbot_scanner):
+    scan = bbot_scanner()
+    await scan._prep()
+    blasthttp_mock.add_response(url=re.compile(r"http://www\.example\.com.*"), text="wat")
+    compare_helper = scan.helpers.http_compare("http://www.example.com")
+    compare_helper.max_differing_lines = 500
+
+    static = [f"line {i}" for i in range(400)]
+    dynamic_a = [f"nonce {i} A" for i in range(600)]
+    dynamic_b = [f"nonce {i} B" for i in range(600)]
+    dynamic_c = [f"nonce {i} C" for i in range(600)]
+
+    baseline_1 = dynamic_a + static
+    baseline_2 = dynamic_b + static
+    subject = dynamic_c + static
+
+    ddiff = _ordered_diff(baseline_1, baseline_2)
+    compare_helper.ddiff_filters = [x.path() for k in ddiff.keys() for x in list(ddiff[k])]
+    assert len(compare_helper.ddiff_filters) == 600
+
+    assert compare_helper.compare_body(baseline_1, subject) is True
+
+    await scan._cleanup()
+
+
+@pytest.mark.asyncio
+async def test_web_http_compare_bounds_dict_bodies(blasthttp_mock, bbot_scanner):
+    scan = bbot_scanner()
+    await scan._prep()
+    blasthttp_mock.add_response(url=re.compile(r"http://www\.example\.com.*"), text="wat")
+    compare_helper = scan.helpers.http_compare("http://www.example.com")
+    compare_helper.max_differing_lines = 500
+    compare_helper.ddiff_filters = []
+
+    def rows(salt):
+        return [
+            f'<div class="r{i}" tok="{salt}{i}">item {i}{salt}</div>'
+            if i < 1000
+            else f'<div class="r{i}">item {i}</div>'
+            for i in range(4000)
+        ]
+
+    content_1 = {"html": {"body": {"div": rows("a")}}}
+    content_2 = {"html": {"body": {"div": rows("b")}}}
+
+    start = time.monotonic()
+    assert compare_helper.compare_body(content_1, content_2) is False
+    assert (time.monotonic() - start) < 5
+
+    await scan._cleanup()
+
+
+@pytest.mark.asyncio
+async def test_web_http_compare_threshold_boundary(blasthttp_mock, bbot_scanner):
+    scan = bbot_scanner()
+    await scan._prep()
+    blasthttp_mock.add_response(url=re.compile(r"http://www\.example\.com.*"), text="wat")
+    compare_helper = scan.helpers.http_compare("http://www.example.com")
+    compare_helper.max_differing_lines = 10
+
+    static = [f"line {i}" for i in range(100)]
+    at_threshold_1 = [f"nonce {i} A" for i in range(5)] + static
+    at_threshold_2 = [f"nonce {i} B" for i in range(5)] + static
+
+    ddiff = _ordered_diff(at_threshold_1, at_threshold_2)
+    compare_helper.ddiff_filters = [x.path() for k in ddiff.keys() for x in list(ddiff[k])]
+
+    assert compare_helper.compare_body(at_threshold_1, at_threshold_2) is True
+
+    over_threshold_1 = [f"nonce {i} A" for i in range(6)] + static
+    over_threshold_2 = [f"nonce {i} B" for i in range(6)] + static
+    compare_helper.ddiff_filters = []
+    assert compare_helper.compare_body(over_threshold_1, over_threshold_2) is False
+
+    await scan._cleanup()
+
+
+@pytest.mark.asyncio
+async def test_web_http_compare_null_threshold_config(blasthttp_mock, bbot_scanner):
+    scan = bbot_scanner(config={"web": {"http_compare_max_differing_lines": None}})
+    await scan._prep()
+    blasthttp_mock.add_response(url=re.compile(r"http://www\.example\.com.*"), text="wat")
+    compare_helper = scan.helpers.http_compare("http://www.example.com")
+    compare_helper.ddiff_filters = []
+
+    assert compare_helper.compare_body(["a", "b", "c"], ["a", "b", "x"]) is False
+
     await scan._cleanup()
 
 
diff --git a/bbot/test/test_step_2/module_tests/base.py b/bbot/test/test_step_2/module_tests/base.py
index 2c55ad6724..c0da1bd074 100644
--- a/bbot/test/test_step_2/module_tests/base.py
+++ b/bbot/test/test_step_2/module_tests/base.py
@@ -1,12 +1,15 @@
+import time
 import pytest
 import asyncio
 import logging
 import pytest_asyncio
+from contextlib import suppress
 
 from ...bbot_fixtures import *
 from bbot.scanner import Scanner
 from bbot.core.config.merge import deep_merge
 from bbot.core.helpers.misc import rand_string
+from bbot.test.worker import CONTAINER_READY_TIMEOUT, ORPHAN_CANCEL_TIMEOUT, _OOM_HINT
 
 log = logging.getLogger("bbot.test.modules")
 
@@ -126,7 +129,16 @@ async def module_test(
             self.log.debug(f"Cancelling {len(tasks)} orphaned tasks after {self.name}")
             for t in tasks:
                 t.cancel()
-            await asyncio.gather(*tasks, return_exceptions=True)
+            # Bounded: a task that absorbs its cancellation (an await shielded from
+            # it, whose resolver was itself cancelled) never completes, and an
+            # unbounded gather here then hangs the whole worker until the job
+            # limit. pytest-timeout cannot fire in fixture teardown.
+            _, pending = await asyncio.wait(tasks, timeout=ORPHAN_CANCEL_TIMEOUT)
+            if pending:
+                self.log.warning(
+                    f"{len(pending)} orphaned tasks did not exit within {ORPHAN_CANCEL_TIMEOUT}s "
+                    f"after {self.name}, abandoning them: {pending}"
+                )
 
     async def _execute_scan(self, module_test):
         """Execute the scan and collect events. Can be overridden by benchmark classes."""
@@ -178,18 +190,84 @@ async def setup_after_prep(self, module_test):
     async def _mock_http_wildcard(*args, **kwargs):
         return False
 
-    async def wait_for_port_open(self, port):
+    async def wait_for_port_open(self, port, timeout=CONTAINER_READY_TIMEOUT):
+        deadline = time.time() + timeout
         while not await self.is_port_open("localhost", port):
+            if time.time() > deadline:
+                raise RuntimeError(
+                    f"Port {port} did not open within {timeout}s, so the container never came up. {_OOM_HINT}"
+                )
             self.log.verbose(f"Waiting for port {port} to be open...")
             await asyncio.sleep(0.5)
         # allow an extra second for things to settle
         await asyncio.sleep(1)
 
-    async def is_port_open(self, host, port):
+    async def start_container(self, name, *args):
+        """Start a detached container, replacing any leftover of the same name.
+
+        A container the previous attempt failed to remove keeps its published
+        ports bound, so the retry's ``docker run`` fails and the test then waits
+        on a port nothing will ever listen on.
+        """
+        rm = await asyncio.create_subprocess_exec(
+            "docker",
+            "rm",
+            "-f",
+            name,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        await rm.communicate()
+
+        proc = await asyncio.create_subprocess_exec(
+            "docker",
+            "run",
+            "-d",
+            "--rm",
+            "--name",
+            name,
+            *args,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await proc.communicate()
+        if proc.returncode != 0:
+            raise RuntimeError(
+                f"Failed to start container {name} (exit {proc.returncode}): {stderr.decode(errors='replace').strip()}"
+            )
+        return stdout.decode(errors="replace").strip()
+
+    async def stop_container(self, name):
+        proc = await asyncio.create_subprocess_exec(
+            "docker",
+            "rm",
+            "-f",
+            name,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        await proc.communicate()
+
+    async def is_port_open(self, host, port, settle=0.5):
+        """Return True only once something is really listening behind ``port``.
+
+        docker publishes a port by binding it on the host at container create
+        time, so a bare connect succeeds while the containerized service is
+        still booting. The proxy then immediately EOFs that connection. Treat an
+        instant EOF as not-yet-listening; a connection that stays open, or one
+        that sends a banner, means the service is up.
+        """
+        writer = None
         try:
             reader, writer = await asyncio.open_connection(host, port)
-            writer.close()
-            await writer.wait_closed()
-            return True
+            try:
+                return await asyncio.wait_for(reader.read(1), timeout=settle) != b""
+            except asyncio.TimeoutError:
+                return True
         except (ConnectionRefusedError, OSError):
             return False
+        finally:
+            if writer is not None:
+                writer.close()
+                with suppress(ConnectionResetError, OSError):
+                    await writer.wait_closed()
diff --git a/bbot/test/test_step_2/module_tests/test_module_elastic.py b/bbot/test/test_step_2/module_tests/test_module_elastic.py
index 95cf083adc..1b91210c28 100644
--- a/bbot/test/test_step_2/module_tests/test_module_elastic.py
+++ b/bbot/test/test_step_2/module_tests/test_module_elastic.py
@@ -1,5 +1,4 @@
 import json
-import asyncio
 import ssl
 from urllib.request import urlopen, Request
 from urllib.error import URLError
@@ -37,12 +36,8 @@ class TestElastic(ModuleTestBase):
 
     async def setup_before_prep(self, module_test):
         # Start Elasticsearch container
-        await asyncio.create_subprocess_exec(
-            "docker",
-            "run",
-            "--name",
+        await self.start_container(
             "bbot-test-elastic",
-            "--rm",
             "-e",
             "ELASTIC_PASSWORD=bbotislife",
             "-e",
@@ -53,7 +48,6 @@ async def setup_before_prep(self, module_test):
             "cluster.routing.allocation.disk.watermark.flood_stage=98%",
             "-p",
             "9200:9200",
-            "-d",
             "docker.elastic.co/elasticsearch/elasticsearch:8.16.0",
         )
 
@@ -123,7 +117,4 @@ async def check(self, module_test, events):
             except URLError:
                 pass
             self.log.verbose("Deleted documents from index")
-            process = await asyncio.create_subprocess_exec(
-                "docker", "stop", "bbot-test-elastic", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
-            )
-            await process.communicate()
+            await self.stop_container("bbot-test-elastic")
diff --git a/bbot/test/test_step_2/module_tests/test_module_excavate.py b/bbot/test/test_step_2/module_tests/test_module_excavate.py
index 8f677ea56b..ee468d5105 100644
--- a/bbot/test/test_step_2/module_tests/test_module_excavate.py
+++ b/bbot/test/test_step_2/module_tests/test_module_excavate.py
@@ -6,6 +6,7 @@
 from bbot.modules.internal.excavate import ExcavateRule, split_yara_rules
 
 from pathlib import Path
+import threading
 import time
 import yara
 from bbot.test.worker import HTTPSERVER_PORT, HTTPSERVER_URL, LOCALHOST_URL
@@ -2186,3 +2187,52 @@ def check(self, module_test, events):
             "Both duplicate-content URLs were processed — content dedup failed for URL events"
         )
         assert len(consumer._content_dup_tracker) > 0, "Content dedup tracker should have entries"
+
+
+@pytest.mark.asyncio
+async def test_excavate_find_subclasses_does_not_touch_descriptors():
+    """excavate.setup() introspects sibling modules while their own setup() is
+    still assigning attributes. Reading through getattr fires BaseModule's
+    memory_usage property, which walks __dict__ and raises "dictionary changed
+    size during iteration" when it loses that race."""
+    from bbot.core.helpers.misc import get_size
+    from bbot.modules.internal.excavate import find_subclasses
+
+    class Sibling:
+        class SomeRule(ExcavateRule):
+            pass
+
+        def __init__(self):
+            for i in range(50):
+                setattr(self, f"attr{i}", i)
+
+        @property
+        def memory_usage(self):
+            return get_size(self, max_depth=3, seen=set())
+
+    sibling = Sibling()
+    assert [c.__name__ for c in find_subclasses(sibling, ExcavateRule)] == ["SomeRule"]
+
+    stop = []
+    errors = []
+
+    def concurrent_setup():
+        i = 0
+        while not stop:
+            setattr(sibling, f"cfg{i}", i)
+            delattr(sibling, f"cfg{i}")
+            i += 1
+
+    churn = threading.Thread(target=concurrent_setup, daemon=True)
+    churn.start()
+    try:
+        for _ in range(3000):
+            try:
+                assert find_subclasses(sibling, ExcavateRule) == [Sibling.SomeRule]
+            except RuntimeError as e:
+                errors.append(e)
+    finally:
+        stop.append(True)
+        churn.join(timeout=5)
+
+    assert not errors, f"find_subclasses raced against a concurrent setup(): {errors[0]}"
diff --git a/bbot/test/test_step_2/module_tests/test_module_gowitness.py b/bbot/test/test_step_2/module_tests/test_module_gowitness.py
index 8df6f9ae39..e5c74e2e39 100644
--- a/bbot/test/test_step_2/module_tests/test_module_gowitness.py
+++ b/bbot/test/test_step_2/module_tests/test_module_gowitness.py
@@ -20,7 +20,6 @@ class TestGowitness(ModuleTestBase):
     home_dir = worker_dir("/tmp/.bbot_gowitness_test")
     shutil.rmtree(home_dir, ignore_errors=True)
     config_overrides = {
-        "deps": {"behavior": "force_install"},
         "home": str(home_dir),
         "scope": {"report_distance": 2},
         "omit_event_types": [],
@@ -158,7 +157,6 @@ class TestGowitness_MultiPort(ModuleTestBase):
     home_dir = worker_dir("/tmp/.bbot_gowitness_multiport_test")
     shutil.rmtree(home_dir, ignore_errors=True)
     config_overrides = {
-        "deps": {"behavior": "force_install"},
         "home": str(home_dir),
         "omit_event_types": [],
     }
diff --git a/bbot/test/test_step_2/module_tests/test_module_kafka.py b/bbot/test/test_step_2/module_tests/test_module_kafka.py
index 451a551e5f..6f05c3d14d 100644
--- a/bbot/test/test_step_2/module_tests/test_module_kafka.py
+++ b/bbot/test/test_step_2/module_tests/test_module_kafka.py
@@ -16,42 +16,45 @@ class TestKafka(ModuleTestBase):
     skip_distro_tests = True
 
     async def setup_before_prep(self, module_test):
-        # Start Zookeeper
-        await asyncio.create_subprocess_exec(
-            "docker", "run", "-d", "--rm", "--name", "bbot-test-zookeeper", "-p", "2181:2181", "zookeeper:3.9"
-        )
-
-        # Wait for Zookeeper to be ready
-        await self.wait_for_port_open(2181)
-
-        # Start Kafka using wurstmeister/kafka
-        await asyncio.create_subprocess_exec(
-            "docker",
-            "run",
-            "-d",
-            "--rm",
-            "--name",
+        # KRaft mode: one broker, no zookeeper. The native image is ~150MB
+        # against ~785MB for wurstmeister/kafka plus zookeeper, and CI pulls
+        # cold on every run.
+        await self.start_container(
             "bbot-test-kafka",
-            "--link",
-            "bbot-test-zookeeper:zookeeper",
             "-e",
-            "KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181",
+            "KAFKA_NODE_ID=1",
             "-e",
-            "KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092",
+            "KAFKA_PROCESS_ROLES=broker,controller",
+            "-e",
+            "KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093",
             "-e",
             "KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092",
             "-e",
+            "KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER",
+            "-e",
+            "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT",
+            "-e",
+            "KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093",
+            "-e",
             "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1",
+            "-e",
+            "KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1",
+            "-e",
+            "KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1",
+            "-e",
+            "KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0",
             "-p",
             "9092:9092",
-            "wurstmeister/kafka",
+            "apache/kafka-native:4.1.2",
         )
 
-        # Wait for Kafka to be ready
+        # this runs before _prep(), which is what pip-installs the module's
+        # deps, so nothing here may import aiokafka. The broker opens its own
+        # listener as the last step of boot, after "Enabling request
+        # processing", so a connection that survives is_port_open's settle
+        # window is a ready broker rather than the docker proxy.
         await self.wait_for_port_open(9092)
 
-        await asyncio.sleep(1)
-
     async def check(self, module_test, events):
         from aiokafka import AIOKafkaConsumer
 
@@ -61,9 +64,12 @@ async def check(self, module_test, events):
             group_id="test_group",
             auto_offset_reset="earliest",
         )
-        await self.consumer.start()
 
         try:
+            # inside the try: a failure here must still tear the containers down,
+            # otherwise they hold port 9092 and every retry fails to bind it
+            await self.consumer.start()
+
             events_json = [e.json() for e in events]
             events_json.sort(key=lambda x: x["timestamp"])
 
@@ -88,12 +94,4 @@ async def _consume():
             # Clean up: Stop the Kafka consumer
             if hasattr(self, "consumer") and not self.consumer._closed:
                 await self.consumer.stop()
-            # Stop Kafka and Zookeeper containers
-            p1 = await asyncio.create_subprocess_exec(
-                "docker", "stop", "bbot-test-kafka", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
-            )
-            await p1.communicate()
-            p2 = await asyncio.create_subprocess_exec(
-                "docker", "stop", "bbot-test-zookeeper", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
-            )
-            await p2.communicate()
+            await self.stop_container("bbot-test-kafka")
diff --git a/bbot/test/test_step_2/module_tests/test_module_mongo.py b/bbot/test/test_step_2/module_tests/test_module_mongo.py
index f889dc1204..5e3262141c 100644
--- a/bbot/test/test_step_2/module_tests/test_module_mongo.py
+++ b/bbot/test/test_step_2/module_tests/test_module_mongo.py
@@ -1,5 +1,3 @@
-import asyncio
-
 from .base import ModuleTestBase
 from bbot.test.worker import wait_for_container
 
@@ -20,19 +18,14 @@ class TestMongo(ModuleTestBase):
     skip_distro_tests = True
 
     async def setup_before_prep(self, module_test):
-        await asyncio.create_subprocess_exec(
-            "docker",
-            "run",
-            "--name",
+        await self.start_container(
             "bbot-test-mongo",
-            "--rm",
             "-e",
             "MONGO_INITDB_ROOT_USERNAME=bbot",
             "-e",
             "MONGO_INITDB_ROOT_PASSWORD=bbotislife",
             "-p",
             "27017:27017",
-            "-d",
             "mongo",
         )
         # Wait for port to be available
@@ -147,7 +140,4 @@ async def check(self, module_test, events):
             await events_collection.delete_many({})
             # Close the MongoDB connection
             await client.aclose()
-            process = await asyncio.create_subprocess_exec(
-                "docker", "stop", "bbot-test-mongo", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
-            )
-            await process.communicate()
+            await self.stop_container("bbot-test-mongo")
diff --git a/bbot/test/test_step_2/module_tests/test_module_mysql.py b/bbot/test/test_step_2/module_tests/test_module_mysql.py
index de30c58f9f..77a34cee0c 100644
--- a/bbot/test/test_step_2/module_tests/test_module_mysql.py
+++ b/bbot/test/test_step_2/module_tests/test_module_mysql.py
@@ -1,5 +1,3 @@
-import asyncio
-
 from .base import ModuleTestBase
 
 
@@ -8,31 +6,20 @@ class TestMySQL(ModuleTestBase):
     skip_distro_tests = True
 
     async def setup_before_prep(self, module_test):
-        process = await asyncio.create_subprocess_exec(
-            "docker",
-            "run",
-            "--name",
+        await self.start_container(
             "bbot-test-mysql",
-            "--rm",
             "-e",
             "MYSQL_ROOT_PASSWORD=bbotislife",
             "-e",
             "MYSQL_DATABASE=bbot",
             "-p",
             "3306:3306",
-            "-d",
             "mysql",
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
         )
-        stdout, stderr = await process.communicate()
 
         # wait for the container to start
         await self.wait_for_port_open(3306)
 
-        if process.returncode != 0:
-            self.log.error(f"Failed to start MySQL server: {stderr.decode()}")
-
     async def check(self, module_test, events):
         import aiomysql
 
@@ -54,10 +41,4 @@ async def check(self, module_test, events):
                 assert len(targets) == 1, "No targets found in MySQL database"
         finally:
             conn.close()
-            process = await asyncio.create_subprocess_exec(
-                "docker", "stop", "bbot-test-mysql", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
-            )
-            stdout, stderr = await process.communicate()
-
-            if process.returncode != 0:
-                raise Exception(f"Failed to stop MySQL server: {stderr.decode()}")
+            await self.stop_container("bbot-test-mysql")
diff --git a/bbot/test/test_step_2/module_tests/test_module_nats.py b/bbot/test/test_step_2/module_tests/test_module_nats.py
index 41ae021cac..9d3e8fba68 100644
--- a/bbot/test/test_step_2/module_tests/test_module_nats.py
+++ b/bbot/test/test_step_2/module_tests/test_module_nats.py
@@ -1,5 +1,4 @@
 import json
-import asyncio
 from contextlib import suppress
 
 from .base import ModuleTestBase
@@ -18,9 +17,7 @@ class TestNats(ModuleTestBase):
 
     async def setup_before_prep(self, module_test):
         # Start NATS server
-        await asyncio.create_subprocess_exec(
-            "docker", "run", "-d", "--rm", "--name", "bbot-test-nats", "-p", "4222:4222", "nats:latest"
-        )
+        await self.start_container("bbot-test-nats", "-p", "4222:4222", "nats:latest")
 
         # Wait for NATS to be ready by checking the port
         await self.wait_for_port_open(4222)
@@ -61,7 +58,4 @@ async def check(self, module_test, events):
                     await self.nc.drain()
                 await self.nc.close()
             # Stop NATS server container
-            process = await asyncio.create_subprocess_exec(
-                "docker", "stop", "bbot-test-nats", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
-            )
-            await process.communicate()
+            await self.stop_container("bbot-test-nats")
diff --git a/bbot/test/test_step_2/module_tests/test_module_neo4j.py b/bbot/test/test_step_2/module_tests/test_module_neo4j.py
index be395206d3..27d00b0379 100644
--- a/bbot/test/test_step_2/module_tests/test_module_neo4j.py
+++ b/bbot/test/test_step_2/module_tests/test_module_neo4j.py
@@ -5,13 +5,14 @@ class TestNeo4j(ModuleTestBase):
     config_overrides = {"modules": {"neo4j": {"uri": "bolt://127.0.0.1:11111"}}}
 
     async def setup_after_prep(self, module_test):
-        # install neo4j
-        deps_pip = module_test.preloaded["neo4j"]["deps"]["pip"]
-        await module_test.scan.helpers.depsinstaller.pip_install(deps_pip)
-
         self.neo4j_used = False
 
     async def setup_before_prep(self, module_test):
+        # monkeypatching neo4j below imports it, so install it first rather than
+        # relying on another test in the session having already pulled it in
+        deps_pip = module_test.preloaded["neo4j"]["deps"]["pip"]
+        await module_test.scan.helpers.depsinstaller.pip_install(deps_pip)
+
         class MockResult:
             async def data(s):
                 self.neo4j_used = True
diff --git a/bbot/test/test_step_2/module_tests/test_module_nuclei.py b/bbot/test/test_step_2/module_tests/test_module_nuclei.py
index 1274a0c090..0f8790e924 100644
--- a/bbot/test/test_step_2/module_tests/test_module_nuclei.py
+++ b/bbot/test/test_step_2/module_tests/test_module_nuclei.py
@@ -1,6 +1,11 @@
 from ...bbot_fixtures import *
 from .base import ModuleTestBase
-from bbot.test.worker import HTTPSERVER_URL, BBOT_TEST_DIR
+from bbot.test.worker import HTTPSERVER_URL, BBOT_TEST_DIR, BBOT_TEST_TOOLS_DIR
+
+import copy
+import fcntl
+from types import SimpleNamespace
+from unittest.mock import patch
 
 
 class TestNucleiManual(ModuleTestBase):
@@ -16,8 +21,7 @@ class TestNucleiManual(ModuleTestBase):
             "nuclei": {
                 "mode": "manual",
                 "concurrency": 2,
-                "ratelimit": 10,
-                "templates": f"{BBOT_TEST_DIR}/tools/nuclei-state/templates/http/miscellaneous/",
+                "templates": f"{BBOT_TEST_TOOLS_DIR}/nuclei-state/templates/http/miscellaneous/",
                 "directory_only": False,
             }
         },
@@ -70,7 +74,7 @@ class TestNucleiSevere(TestNucleiManual):
             "nuclei": {
                 "mode": "severe",
                 "concurrency": 1,
-                "templates": f"{BBOT_TEST_DIR}/tools/nuclei-state/templates/http/vulnerabilities/generic/generic-env.yaml",
+                "templates": f"{BBOT_TEST_TOOLS_DIR}/nuclei-state/templates/http/vulnerabilities/generic/generic-env.yaml",
             }
         },
         "interactsh_disable": True,
@@ -90,9 +94,13 @@ def check(self, module_test, events):
 
 
 class TestNucleiTechnology(TestNucleiManual):
+    # etypes tcp: the fixture is a single HTTP port, so tcp-protocol templates
+    # (cassandra 9042, kafka 9092, ajp 8009) can only ever dial a closed port and
+    # wait out nuclei's fixed 5s read timeout. They match nothing here.
     config_overrides = {
+        "web": {"http_timeout": 7},
         "interactsh_disable": True,
-        "modules": {"nuclei": {"mode": "technology", "concurrency": 2, "tags": "apache"}},
+        "modules": {"nuclei": {"mode": "technology", "concurrency": 2, "tags": "apache", "etypes": "tcp"}},
     }
 
     async def setup_before_prep(self, module_test):
@@ -103,10 +111,31 @@ async def setup_before_prep(self, module_test):
         }
         module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args)
 
+    async def setup_after_prep(self, module_test):
+        self.commands = []
+        module = module_test.scan.modules["nuclei"]
+        original = module.run_process_live
+
+        def capture(*args, **kwargs):
+            command = args[0] if args and isinstance(args[0], (list, tuple)) else args
+            self.commands.append([str(c) for c in command])
+            return original(*args, **kwargs)
+
+        module.run_process_live = capture
+
     def check(self, module_test, events):
         assert any(e.type == "TECHNOLOGY" and "apache" in e.data["technology"].lower() for e in events)
         assert "Using Interactsh Server" not in open(module_test.scan.home / "debug.log").read()
 
+        assert self.commands, "nuclei was never executed"
+        for command in self.commands:
+            assert "-exclude-type" in command, f"-exclude-type missing from {command}"
+            assert command[command.index("-exclude-type") + 1] == "tcp"
+            # nuclei silently defaults to its own 10s, so an unpassed -timeout
+            # means web.http_timeout is ignored by this module alone
+            assert "-timeout" in command, f"-timeout missing from {command}"
+            assert command[command.index("-timeout") + 1] == "7"
+
 
 class TestNucleiBudget(TestNucleiManual):
     config_overrides = {
@@ -116,7 +145,7 @@ class TestNucleiBudget(TestNucleiManual):
                 "mode": "budget",
                 "concurrency": 1,
                 "tags": "spiderfoot",
-                "templates": f"{BBOT_TEST_DIR}/tools/nuclei-state/templates/exposed-panels/spiderfoot.yaml",
+                "templates": f"{BBOT_TEST_TOOLS_DIR}/nuclei-state/templates/exposed-panels/spiderfoot.yaml",
             }
         },
     }
@@ -233,9 +262,134 @@ def test_nuclei_classify_update_stderr():
     assert c(None) == "failure"
 
 
+@pytest.mark.asyncio
+async def test_nuclei_repair_wipe_holds_the_lock(tmp_path):
+    """Regression: the corruption repair wipes the whole nuclei state dir. If it
+    runs outside the template lock, a worker that wipes while another is
+    mid-extract destroys the other's output, so both fail and every nuclei test
+    on every xdist worker hard-fails with "Failed to install nuclei templates
+    after retry". Pin that the wipe only happens while the lock is held.
+    """
+    from bbot.modules.nuclei import nuclei
+
+    tools_dir = tmp_path / "tools"
+    tools_dir.mkdir()
+    state_dir = tools_dir / "nuclei-state"
+    templates_dir = state_dir / "templates"
+
+    mod = nuclei.__new__(nuclei)
+    mod.nuclei_state_dir = state_dir
+    mod.nuclei_config_dir = state_dir / "config"
+    mod.nuclei_cache_dir = state_dir / "cache"
+    mod.nuclei_templates_dir = templates_dir
+    mod.nuclei_config_dir.mkdir(parents=True, exist_ok=True)
+    mod.nuclei_cache_dir.mkdir(parents=True, exist_ok=True)
+
+    helpers = SimpleNamespace(
+        tools_dir=tools_dir,
+        run_in_executor_io=lambda fn, *a: asyncio.get_running_loop().run_in_executor(None, fn, *a),
+    )
+    for name in ("info", "warning", "success", "debug"):
+        setattr(mod, name, lambda *a, **kw: None)
+
+    held_during_wipe = []
+    original_rmtree = shutil.rmtree
+
+    def probe_rmtree(path, *args, **kwargs):
+        # a second process must not be able to take the lock while we wipe
+        probe = open(tools_dir / "nuclei-templates.lock", "w")
+        try:
+            try:
+                fcntl.flock(probe, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                held_during_wipe.append(False)
+                fcntl.flock(probe, fcntl.LOCK_UN)
+            except OSError:
+                held_during_wipe.append(True)
+        finally:
+            probe.close()
+        return original_rmtree(path, *args, **kwargs)
+
+    # first update produces nothing (simulates the killed/incomplete extract),
+    # second one populates the tree so the repair path is exercised end to end
+    calls = []
+
+    async def fake_update():
+        calls.append(1)
+        if len(calls) > 1:
+            (templates_dir / "http").mkdir(parents=True, exist_ok=True)
+            (templates_dir / "http" / "t.yaml").write_text("id: t")
+        # claimed success but produced nothing: the stale-marker corruption the
+        # wipe exists to repair
+        return "updated"
+
+    mod._run_template_update = fake_update
+
+    with patch.object(shutil, "rmtree", probe_rmtree), patch.object(nuclei, "helpers", helpers):
+        installed = await mod._ensure_templates()
+
+    assert installed, "repair path should report success once templates land"
+    assert calls == [1, 1], "repair should run exactly one retry update"
+    assert held_during_wipe == [True], "state dir was wiped without holding the template lock"
+    # lock must be released afterwards so the next worker can proceed
+    after = open(tools_dir / "nuclei-templates.lock", "w")
+    try:
+        fcntl.flock(after, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        fcntl.flock(after, fcntl.LOCK_UN)
+    finally:
+        after.close()
+
+
+@pytest.mark.asyncio
+async def test_nuclei_failed_download_does_not_wipe_and_refetch(tmp_path):
+    """Regression: when the template download itself fails, nuclei produced no
+    files, so there is no stale-marker corruption to repair. Wiping and
+    re-downloading doubles the requests against an already-failing source. In CI
+    this turned one bad fetch into 66 downloads and 33 hard failures across every
+    xdist worker. A "failure" outcome must fail fast without a second fetch.
+    """
+    from bbot.modules.nuclei import nuclei
+
+    tools_dir = tmp_path / "tools"
+    tools_dir.mkdir()
+    state_dir = tools_dir / "nuclei-state"
+
+    mod = nuclei.__new__(nuclei)
+    mod.nuclei_state_dir = state_dir
+    mod.nuclei_config_dir = state_dir / "config"
+    mod.nuclei_cache_dir = state_dir / "cache"
+    mod.nuclei_templates_dir = state_dir / "templates"
+    mod.nuclei_config_dir.mkdir(parents=True, exist_ok=True)
+    mod.nuclei_cache_dir.mkdir(parents=True, exist_ok=True)
+
+    helpers = SimpleNamespace(
+        tools_dir=tools_dir,
+        run_in_executor_io=lambda fn, *a: asyncio.get_running_loop().run_in_executor(None, fn, *a),
+    )
+    for name in ("info", "warning", "success", "debug"):
+        setattr(mod, name, lambda *a, **kw: None)
+
+    calls = []
+    wiped = []
+
+    async def failing_update():
+        calls.append(1)
+        return "failure"
+
+    mod._run_template_update = failing_update
+
+    with patch.object(shutil, "rmtree", lambda *a, **kw: wiped.append(1)), patch.object(nuclei, "helpers", helpers):
+        installed = await mod._ensure_templates()
+
+    assert installed is False, "a failed download must not report success"
+    assert calls == [1], "a failed download must not trigger a second fetch"
+    assert wiped == [], "a failed download must not wipe state that holds no corruption"
+
+
 class TestNucleiCustomHeaders(TestNucleiManual):
     custom_headers = {"testheader1": "test1", "testheader2": "test2"}
-    config_overrides = TestNucleiManual.config_overrides
+    # deep copy: a shallow dict() still shares the nested "web" dict, so writing
+    # http_headers into it would reach TestNucleiManual and every sibling subclass.
+    config_overrides = copy.deepcopy(TestNucleiManual.config_overrides)
     config_overrides["web"]["http_headers"] = custom_headers
 
     async def setup_after_prep(self, module_test):
diff --git a/bbot/test/test_step_2/module_tests/test_module_postgres.py b/bbot/test/test_step_2/module_tests/test_module_postgres.py
index 8c52eabebe..e3d1e47457 100644
--- a/bbot/test/test_step_2/module_tests/test_module_postgres.py
+++ b/bbot/test/test_step_2/module_tests/test_module_postgres.py
@@ -1,5 +1,3 @@
-import asyncio
-
 from .base import ModuleTestBase
 
 
@@ -8,28 +6,20 @@ class TestPostgres(ModuleTestBase):
     skip_distro_tests = True
 
     async def setup_before_prep(self, module_test):
-        process = await asyncio.create_subprocess_exec(
-            "docker",
-            "run",
-            "--name",
+        await self.start_container(
             "bbot-test-postgres",
-            "--rm",
             "-e",
             "POSTGRES_PASSWORD=bbotislife",
             "-e",
             "POSTGRES_USER=postgres",
             "-p",
             "5432:5432",
-            "-d",
             "postgres",
         )
 
         # wait for the container to start
         await self.wait_for_port_open(5432)
 
-        if process.returncode != 0:
-            self.log.error("Failed to start PostgreSQL server")
-
     async def check(self, module_test, events):
         import asyncpg
 
@@ -45,10 +35,4 @@ async def check(self, module_test, events):
             assert len(targets) == 1, "No targets found in PostgreSQL database"
         finally:
             await conn.close()
-            process = await asyncio.create_subprocess_exec(
-                "docker", "stop", "bbot-test-postgres", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
-            )
-            stdout, stderr = await process.communicate()
-
-            if process.returncode != 0:
-                raise Exception(f"Failed to stop PostgreSQL server: {stderr.decode()}")
+            await self.stop_container("bbot-test-postgres")
diff --git a/bbot/test/test_step_2/module_tests/test_module_rabbitmq.py b/bbot/test/test_step_2/module_tests/test_module_rabbitmq.py
index 0e76fd05db..0ec01ec307 100644
--- a/bbot/test/test_step_2/module_tests/test_module_rabbitmq.py
+++ b/bbot/test/test_step_2/module_tests/test_module_rabbitmq.py
@@ -1,5 +1,4 @@
 import json
-import asyncio
 
 from .base import ModuleTestBase
 
@@ -16,21 +15,8 @@ class TestRabbitMQ(ModuleTestBase):
     skip_distro_tests = True
 
     async def setup_before_prep(self, module_test):
-        # Remove any leftover container from a previous failed run
-        proc = await asyncio.create_subprocess_exec(
-            "docker",
-            "rm",
-            "-f",
-            "bbot-test-rabbitmq",
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-        await proc.wait()
-
         # Start RabbitMQ
-        await asyncio.create_subprocess_exec(
-            "docker", "run", "-d", "--rm", "--name", "bbot-test-rabbitmq", "-p", "5672:5672", "rabbitmq:3-management"
-        )
+        await self.start_container("bbot-test-rabbitmq", "-p", "5672:5672", "rabbitmq:3-management")
 
         # Wait for RabbitMQ to be ready by checking the port
         await self.wait_for_port_open(5672)
@@ -65,7 +51,4 @@ async def check(self, module_test, events):
             # Clean up: Close the RabbitMQ connection
             await connection.close()
             # Stop RabbitMQ container
-            process = await asyncio.create_subprocess_exec(
-                "docker", "stop", "bbot-test-rabbitmq", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
-            )
-            await process.communicate()
+            await self.stop_container("bbot-test-rabbitmq")
diff --git a/bbot/test/test_step_2/module_tests/test_module_websocket.py b/bbot/test/test_step_2/module_tests/test_module_websocket.py
index 26cd28ecd8..c4eedb7d03 100644
--- a/bbot/test/test_step_2/module_tests/test_module_websocket.py
+++ b/bbot/test/test_step_2/module_tests/test_module_websocket.py
@@ -1,5 +1,4 @@
 import json
-import asyncio
 import logging
 from websockets.asyncio.server import serve
 
@@ -17,20 +16,21 @@ async def websocket_handler(websocket):
         results["events"].append(message)
 
 
-# Define a coroutine for the server
-async def server_coroutine():
-    async with serve(websocket_handler, "127.0.0.1", WEBSOCKET_PORT) as server:
-        await server.serve_forever()
-
-
 class TestWebsocket(ModuleTestBase):
     config_overrides = {"modules": {"websocket": {"url": f"ws://127.0.0.1:{WEBSOCKET_PORT}/testing"}}}
 
     async def setup_before_prep(self, module_test):
-        self.server_task = asyncio.create_task(server_coroutine())
+        self.server = await serve(websocket_handler, "127.0.0.1", WEBSOCKET_PORT)
+
+    async def _execute_scan(self, module_test):
+        # shut down on the fixture's loop, which is the one that created the server
+        try:
+            await super()._execute_scan(module_test)
+        finally:
+            self.server.close()
+            await self.server.wait_closed()
 
     def check(self, module_test, events):
         assert results["path"] == "/testing"
         decoded_events = [json.loads(e) for e in results["events"]]
         assert any(e["type"] == "SCAN" for e in decoded_events)
-        self.server_task.cancel()
diff --git a/bbot/test/worker.py b/bbot/test/worker.py
index 8901dc65fb..5368b95c38 100644
--- a/bbot/test/worker.py
+++ b/bbot/test/worker.py
@@ -48,6 +48,13 @@ def worker_dir(base=BASE_BBOT_TEST_DIR):
     return Path(f"{base}_{worker}") if worker else Path(base)
 
 
+# Dependency installs are keyed on bbot_home, so a per-worker home makes every
+# worker reinstall all ~126 modules' deps from scratch. Point the install-once
+# dirs at one shared location; scan output and temp files stay per-worker.
+BBOT_TEST_SHARED_DIR = Path(f"{BASE_BBOT_TEST_DIR}_shared")
+BBOT_TEST_TOOLS_DIR = BBOT_TEST_SHARED_DIR / "tools"
+
+
 HTTPSERVER_PORT = worker_port(8888)
 HTTPSERVER_SSL_PORT = worker_port(9999)
 HTTPSERVER_ALLINTERFACES_PORT = worker_port(5556)
@@ -83,6 +90,10 @@ def worker_dir(base=BASE_BBOT_TEST_DIR):
 
 _OOM_HINT = "A container that exits with code 137 was OOM-killed; it competes with the other test workers for memory."
 
+# A cancelled task that never completes would otherwise block fixture teardown
+# forever, and pytest-timeout does not cover teardown under timeout_func_only.
+ORPHAN_CANCEL_TIMEOUT = int(os.environ.get("BBOT_TEST_ORPHAN_TIMEOUT", "60"))
+
 
 async def wait_for_container(name, connect, timeout=CONTAINER_READY_TIMEOUT):
     """Retry ``connect`` until it succeeds, or raise once ``timeout`` elapses.
diff --git a/docs/modules/nuclei.md b/docs/modules/nuclei.md
index f9f9075c14..82c8ed0107 100644
--- a/docs/modules/nuclei.md
+++ b/docs/modules/nuclei.md
@@ -43,6 +43,7 @@ The Nuclei module has many configuration options:
 | modules.nuclei.concurrency    | int                                                 | maximum number of templates to be executed in parallel (default 25)                                                                                                                                                                                                                                                            | 25        |
 | modules.nuclei.directory_only | bool                                                | Filter out 'file' URL event (default True)                                                                                                                                                                                                                                                                                     | True      |
 | modules.nuclei.etags          | str                                                 | tags to exclude from the scan                                                                                                                                                                                                                                                                                                  |           |
+| modules.nuclei.etypes         | str                                                 | protocol types to exclude from the scan, comma-separated (dns, file, http, headless, tcp, workflow, ssl, websocket, whois, code, javascript)                                                                                                                                                                                   |           |
 | modules.nuclei.mode           | Literal['manual', 'technology', 'severe', 'budget'] | manual &#124; technology &#124; severe &#124; budget. Technology: Only activate based on technology events that match nuclei tags (nuclei -as mode). Manual (DEFAULT): Fully manual settings. Severe: Only critical and high severity templates without intrusive. Budget: Limit Nuclei to a specified number of HTTP requests | manual    |
 | modules.nuclei.module_timeout | int                                                 | Max time in seconds to spend handling each batch of events                                                                                                                                                                                                                                                                     | 21600     |
 | modules.nuclei.ratelimit      | int                                                 | maximum number of requests to send per second (default 150)                                                                                                                                                                                                                                                                    | 150       |
diff --git a/docs/scanning/configuration.md b/docs/scanning/configuration.md
index b392fbdb7f..6aae71afcd 100644
--- a/docs/scanning/configuration.md
+++ b/docs/scanning/configuration.md
@@ -519,6 +519,7 @@ In addition to the stated options for each module, the following universal optio
 | modules.nuclei.concurrency                             | int                                                  | maximum number of templates to be executed in parallel (default 25)                                                                                                                                                                                                                                                            | 25                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
 | modules.nuclei.directory_only                          | bool                                                 | Filter out 'file' URL event (default True)                                                                                                                                                                                                                                                                                     | True                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
 | modules.nuclei.etags                                   | str                                                  | tags to exclude from the scan                                                                                                                                                                                                                                                                                                  |                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
+| modules.nuclei.etypes                                  | str                                                  | protocol types to exclude from the scan, comma-separated (dns, file, http, headless, tcp, workflow, ssl, websocket, whois, code, javascript)                                                                                                                                                                                   |                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
 | modules.nuclei.mode                                    | Literal['manual', 'technology', 'severe', 'budget']  | manual &#124; technology &#124; severe &#124; budget. Technology: Only activate based on technology events that match nuclei tags (nuclei -as mode). Manual (DEFAULT): Fully manual settings. Severe: Only critical and high severity templates without intrusive. Budget: Limit Nuclei to a specified number of HTTP requests | manual                                                                                                                                                                                                                                                                                                                                                                                                                                                  |
 | modules.nuclei.module_timeout                          | int                                                  | Max time in seconds to spend handling each batch of events                                                                                                                                                                                                                                                                     | 21600                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
 | modules.nuclei.ratelimit                               | int                                                  | maximum number of requests to send per second (default 150)                                                                                                                                                                                                                                                                    | 150                                                                                                                                                                                                                                                                                                                                                                                                                                                     |