Merge remote-tracking branch 'origin/main' into fix/346-test-ci-hygiene

On Windows py3.13 CI (PR #405) a one-file duplicate cleanup missed its
10 s WAIT. The test already waits on the job's own task; the 10 s bound
acted as a deadline on disk-bound work. In the same 50 s window the
following tests' setups took up to 9 s against 0.1-0.6 s around it
while the test bodies stayed fast, which is consistent with a
process-wide disk I/O stall on the runner rather than a lease, handle,
executor or job stall (the log has no thread dump to show more).

tests/disk_waits.py holds the one disk-bound hang guard (90 s per wait,
under the 120 s per-test watchdog), shared with RESTORE_SECONDS (#382).
Waits on find and cleanup job tasks in the duplicate-files tests use
it; gates and server-state polls keep WAIT. The keeper-validation test
turns the safety backup off, since it does not test backups.

.github/workflows/test.yml

@@ -221,7 +221,10 @@ jobs:
       - uses: actions/setup-python@v7
         with:
           python-version: "3.12"
-      - run: pip install black flake8
+          cache: pip
+          cache-dependency-path: requirements-dev.txt
+      # The exact black and flake8 pins from requirements-dev.txt, so a new release cannot turn CI red.
+      - run: pip install $(grep -E '^(black|flake8)==' requirements-dev.txt)
       - run: black --check .
       - run: flake8 .
 
@@ -235,6 +238,8 @@ jobs:
       - uses: actions/setup-node@v7
         with:
           node-version: "24"
+          cache: npm
+          cache-dependency-path: tools/vendor/package-lock.json
 
       - name: Check shipped extension and page modules parse
         run: |

.pre-commit-config.yaml

@@ -13,9 +13,11 @@ repos:
         language: system
         types: [python]
 
+      # Fails at the severity CI enforces (code-quality.yml: -lll, high only), never stricter.
+      # pre-commit appends the staged files, so no shell wrapper can drop the first one.
       - id: bandit
         name: bandit
-        entry: bash -c 'bandit "$@" -ll -q 2>&1; true'
+        entry: bandit -lll -q
         language: system
         types: [python]
         exclude: ^(tests/|reference/)

requirements-dev.txt

@@ -8,8 +8,8 @@ coverage>=7.16.2
 numpy>=2.2.6  # WD14 engine tests
 
 # Code quality
-black>=26.10.1
-flake8>=7.4.1
+black==26.10.1
+flake8==7.4.1
 
 # Security scanning
 bandit[toml]>=1.9.4

tests/disk_waits.py

@@ -0,0 +1,16 @@
+"""Hang guards for waits on disk-bound work (tests only).
+
+A test that waits for a job or handoff whose work is file and SQLite I/O
+(a safety backup with fsync, a restore, hashing and unlinking files) waits
+on that state; the ceiling below only ends a real hang. Hosted Windows
+runners stall disk I/O for tens of seconds at a time: a restore took over
+10 s (merge queue, #382), and on PR #405 every test setup in a 50 s window
+took up to 9 s against 0.1-0.6 s around it while the test bodies stayed
+fast, so a one-file duplicate cleanup missed a 10 s bound (#409).
+
+The ceiling is per wait and stays under the suite's 120 s per-test hang
+watchdog. A test with several such waits in a row can reach the watchdog
+first; either way a real hang fails, the watchdog with every thread's stack.
+"""
+
+DISK_BOUND_SECONDS = 90

tests/private_temp.py

@@ -0,0 +1,24 @@
+"""A temp SQLite database in a folder no other test process shares (#346).
+
+``PromptManagerAPI`` sweeps its database's folder at startup for the stages,
+uploads and safety-backup copies an interrupted restore or backup left. A
+fixture database directly in the system temp folder made every other test
+process that started an API (another worktree's suite, a pre-push hook) delete
+this process's in-flight restore files, and left safety backups behind there.
+"""
+
+import tempfile
+
+
+def private_temp_db(case):
+    """Return a closed ``NamedTemporaryFile`` handle for an empty ``.db`` file.
+
+    The file lives in a new folder that ``case`` removes at cleanup, after its
+    own teardown has closed and unlinked the database (cleanups run after
+    ``tearDown``), together with anything a restore or backup left next to it.
+    """
+    folder = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
+    case.addCleanup(folder.cleanup)
+    handle = tempfile.NamedTemporaryFile(delete=False, suffix=".db", dir=folder.name)
+    handle.close()
+    return handle

tests/test_api.py

@@ -8,7 +8,6 @@ without requiring a full ComfyUI server.
 import json
 import os
 import sys
-import tempfile
 import unittest
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@@ -18,6 +17,7 @@ from aiohttp.test_utils import AioHTTPTestCase
 
 from database.operations import PromptDatabase
 from py.api import PromptManagerAPI
+from tests.private_temp import private_temp_db
 from utils.hashing import generate_prompt_hash
 
 
@@ -25,8 +25,7 @@ class APITestCase(AioHTTPTestCase):
     """Base class that stands up a test aiohttp app with PromptManager routes."""
 
     async def get_application(self):
-        self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
-        self._temp_db.close()
+        self._temp_db = private_temp_db(self)
 
         app = web.Application()
         routes = web.RouteTableDef()

tests/test_api_autotag.py

@@ -35,6 +35,7 @@ from aiohttp.test_utils import AioHTTPTestCase
 
 from database.operations import PromptDatabase
 from py.api import PromptManagerAPI
+from tests.private_temp import private_temp_db
 from py.api.autotag_routes import MODEL_BUSY_ERROR
 from py.api.jobs import JobBusy
 from utils.hashing import generate_prompt_hash
@@ -143,8 +144,7 @@ class AutotagAPITestCase(AioHTTPTestCase):
     fake engine."""
 
     async def get_application(self):
-        self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
-        self._temp_db.close()
+        self._temp_db = private_temp_db(self)
         self.tmp_root = Path(tempfile.mkdtemp())
         self.output_dir = self.tmp_root / "output"
         self.output_dir.mkdir()

tests/test_api_duplicate_files.py

@@ -18,15 +18,19 @@ from pathlib import Path
 from unittest.mock import patch
 
 try:
+    from tests.disk_waits import DISK_BOUND_SECONDS
     from tests.open_handles import open_handles_under
     from tests.test_api_maintenance import ApplyTestCase, _digest
 except ImportError:  # discovered with tests/ as the top-level directory
+    from disk_waits import DISK_BOUND_SECONDS
     from open_handles import open_handles_under
     from test_api_maintenance import ApplyTestCase, _digest
 
 from py.api import duplicate_files  # noqa: E402
 from py.config import GalleryConfig, PromptManagerConfig  # noqa: E402
 
+# Gates and server-state polls. Waits on a find or cleanup job's task use
+# DISK_BOUND_SECONDS: those jobs hash, back up and unlink files (#409).
 WAIT = 10
 FIND = "/prompt_manager/jobs/duplicates"
 GROUPS = "/prompt_manager/maintenance/duplicates/"
@@ -83,7 +87,7 @@ class DuplicateFilesTestCase(ApplyTestCase):
         self.assertEqual(response.status, 202, text)
         self.assertNotIn(self.tmpdir, text)
         job = self.api._jobs.get(json.loads(text)["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
         return job
 
     async def groups(self, job_id, status=200, **query):
@@ -127,7 +131,7 @@ class DuplicateFilesTestCase(ApplyTestCase):
         plan = await self.review(job.id, [self.keep_of(group, path)])
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         return plan, cleanup
 
     async def until(self, predicate):
@@ -178,7 +182,7 @@ class TestFindJob(DuplicateFilesTestCase):
         gate.set()
         body = await (await legacy).text()
         job = self.api._jobs.get(summary["id"])
-        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
         events = _sse_payloads(body)
         done = events[-1]
         self.assertEqual(done["type"], "complete")
@@ -241,7 +245,7 @@ class TestFindJob(DuplicateFilesTestCase):
         self.assertTrue((await cancel.json())["cancellation_requested"])
         gate.set()
         job = self.api._jobs.get(job_id)
-        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
         self.assertEqual(job.state, "cancelled")
         last = _sse_payloads(await (await stream).text())[-1]
         self.assertEqual(last, {"type": "error", "message": "Duplicate scan cancelled"})
@@ -341,6 +345,8 @@ class TestReview(DuplicateFilesTestCase):
         self.assertEqual(plan["protected"], {"count": 0, "reasons": []})
 
     async def test_cannot_delete_keeper_or_outside_group(self):
+        # About which files may go, not the backup: skip its fsync-heavy copy.
+        PromptManagerConfig.SAFETY_BACKUP = False
         self.trio()
         self.write("e.png", b"other pair")
         self.write("f.png", b"other pair")
@@ -372,7 +378,7 @@ class TestReview(DuplicateFilesTestCase):
         self.assertEqual([row["label"] for row in plan["details"]], ["b.png"])
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.counters["removed"], 1)
         self.assertTrue((self.output_dir / "a.png").is_file())
         self.assertTrue((self.output_dir / "sub" / "c.png").is_file())
@@ -496,7 +502,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         plan = await self.review(job.id, keep)
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertTrue(kept.is_file())
         self.assertFalse(gone.exists())
         self.assertFalse(video_thumb.exists())
@@ -508,7 +514,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         self.write("b.png", b"new bytes")  # changed since the review
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.counters, {"removed": 1, "kept": 2, "errors": 0})
         self.assertEqual((self.output_dir / "b.png").read_bytes(), b"new bytes")
         self.assertFalse((self.output_dir / "sub" / "c.png").exists())
@@ -522,7 +528,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         self.write("a.png", b"edited keeper")
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.counters["removed"], 0)
         self.assertTrue((self.output_dir / "b.png").is_file())
         self.assertTrue((self.output_dir / "sub" / "c.png").is_file())
@@ -549,7 +555,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         with patch.object(duplicate_files, "_unlink", side_effect=unlink):
             body = await self.apply(plan["token"])
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.state, "failed")
         self.assertEqual(cleanup.counters, {"removed": 1, "kept": 1, "errors": 1})
         self.assertEqual(
@@ -584,7 +590,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         ):
             body = await self.apply(plan["token"])
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.counters, {"removed": 0, "kept": 3, "errors": 1})
         self.assertEqual(
             cleanup.outcomes["errors"][0]["detail"], duplicate_files.DELETE_FAILED
@@ -619,7 +625,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
             with context:
                 body = await self.apply(plan["token"])
                 cleanup = self.api._jobs.get(body["job"]["id"])
-                await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+                await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         finally:
             if handle is not None:  # pragma: no cover - Windows
                 handle.close()
@@ -642,7 +648,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         with patch.object(duplicate_files, "_unlink", side_effect=unlink):
             body = await self.apply(plan["token"])
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.counters["removed"], 2)
         self.assertEqual(open_at_unlink, [])
 
@@ -703,7 +709,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         with patch.object(duplicate_files, "_unlink", side_effect=unlink):
             body = await self.apply(plan["token"])
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         # Its records moved to the identical kept copy before the unlink failed.
         [image] = self.api.db.get_prompt_images(prompt)
         self.assertTrue(image["image_path"].endswith("a.png"))
@@ -719,7 +725,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         self.assertEqual((busy["code"], busy["running"]["id"]), ("busy", find["id"]))
         gate.set()
         await asyncio.wait_for(
-            asyncio.shield(self.api._jobs.get(find["id"]).task), WAIT
+            asyncio.shield(self.api._jobs.get(find["id"]).task), DISK_BOUND_SECONDS
         )
         # The token survived the refusal; the cleanup holds off a new find.
         release = asyncio.Event()
@@ -737,7 +743,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
             self.assertEqual(legacy.status, 409)
             release.set()
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
 
     async def test_cleanup_refused_up_front_when_the_backup_does_not_fit(self):
         from py.api import maintenance
@@ -965,7 +971,7 @@ class TestReviewHelpers(DuplicateFilesTestCase):
         GalleryConfig.MONITORING_DIRECTORIES = [self._root(self.second_source())]
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.state, "failed")
         self.assertEqual(cleanup.last_event["code"], "sources-changed")
         self.assertEqual(self.backups(), [])
@@ -985,7 +991,7 @@ class TestReviewHelpers(DuplicateFilesTestCase):
         ):
             body = await self.apply(plan["token"])
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.state, "failed")
         self.assertEqual(
             cleanup.outcomes["errors"][-1]["detail"], duplicate_files.RECORDS_FAILED

tests/test_api_duplicate_safety.py

@@ -19,17 +19,17 @@ from pathlib import Path
 from unittest.mock import patch
 
 try:
+    from tests.disk_waits import DISK_BOUND_SECONDS
     from tests.open_handles import open_handles_under
     from tests import test_api_duplicate_files as dup_tests
 except ImportError:  # discovered with tests/ as the top-level directory
     import test_api_duplicate_files as dup_tests
+    from disk_waits import DISK_BOUND_SECONDS
     from open_handles import open_handles_under
 
 from py.api import duplicate_files  # noqa: E402
 from py.config import GalleryConfig  # noqa: E402
 
-WAIT = 10
-
 
 def _locator(path):
     return os.path.normcase(os.path.realpath(str(path)))
@@ -49,7 +49,7 @@ class SafetyTestCase(dup_tests.DuplicateFilesTestCase):
     async def apply_plan(self, plan):
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         return cleanup
 
 

tests/test_api_history.py

@@ -1169,3 +1169,47 @@ class HistoryHostCleanupTests(unittest.IsolatedAsyncioTestCase):
                     fence.validate(epoch, generation)
                 with self.assertRaises(sqlite3.Error):
                     api._history_fence()
+
+
+class SharedTempFolderRestoreTests(ImageAPITestCase):
+    """#346: a restore survives another suite starting in the shared temp folder.
+
+    Every ``PromptManagerAPI`` sweeps its database's folder for stages and
+    uploads an interrupted restore left. A fixture database directly in the
+    shared temp folder let any other test process (another worktree's suite,
+    a pre-push hook) delete this restore's files mid-request. The shared
+    folder here is a scratch stand-in for the system temp folder, so the test
+    never sweeps the real one.
+    """
+
+    async def get_application(self):
+        shared = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
+        # Cleanups run last-in first-out: this one after the fixture's own.
+        self.addCleanup(shared.cleanup)
+        self.shared_tmp = shared.name
+        with patch.object(tempfile, "tempdir", shared.name):
+            return await super().get_application()
+
+    async def test_restore_survives_another_suite_sweeping_the_shared_folder(self):
+        from database import restore_stage
+        from database.models import PromptModel
+
+        self.api.db.save_prompt("kept by the upload")
+        source = self.comfy_dir / "upload.db"
+        self.assertTrue(self.api.db.model.backup_database(str(source)))
+        stage = PromptModel.stage_restore
+        swept = []
+
+        def stage_then_other_suite_starts(model, src_path):
+            staged = stage(model, src_path)
+            # What another suite's PromptManagerAPI does at startup.
+            other = os.path.join(self.shared_tmp, "other-suite.db")
+            swept.extend(restore_stage.sweep_restore_temps(other))
+            return staged
+
+        form = FormData()
+        form.add_field("database_file", source.read_bytes(), filename="upload.db")
+        with patch.object(PromptModel, "stage_restore", stage_then_other_suite_starts):
+            response = await self.client.post("/prompt_manager/restore", data=form)
+        self.assertEqual(response.status, 200, await response.text())
+        self.assertEqual(swept, [], "another suite's sweep reached this restore")

tests/test_api_images.py

@@ -42,6 +42,7 @@ from py import lora_utils  # noqa: E402
 from py import lora_manager_layout as layout_module  # noqa: E402
 from py.api.thumbnail_settings import ThumbnailSettings  # noqa: E402
 from py.api import PromptManagerAPI  # noqa: E402
+from tests.private_temp import private_temp_db  # noqa: E402
 from py.api.prompts import MAX_PAGE_LIMIT  # noqa: E402
 from py.config import GalleryConfig  # noqa: E402
 from utils.hashing import generate_prompt_hash  # noqa: E402
@@ -60,8 +61,7 @@ class ImageAPITestCase(AioHTTPTestCase):
     """App with PromptManager routes, a temp DB and a temp output directory."""
 
     async def get_application(self):
-        self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
-        self._temp_db.close()
+        self._temp_db = private_temp_db(self)
 
         self._output_tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
         # A fake ComfyUI tree: <tmp>/output is the gallery root and <tmp> is the

tests/test_api_lora.py

@@ -36,6 +36,7 @@ from aiohttp.test_utils import AioHTTPTestCase
 
 from database.operations import PromptDatabase
 from py.api import PromptManagerAPI
+from tests.private_temp import private_temp_db
 from py import config_store, lora_import, lora_utils
 from py.config import IntegrationConfig, PromptManagerConfig
 from py.api import settings_routes as settings_module
@@ -57,8 +58,7 @@ class LoraAPITestCase(AioHTTPTestCase):
     """Stands up the PromptManager routes with a temp DB and neutral LoRA config."""
 
     async def get_application(self):
-        self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
-        self._temp_db.close()
+        self._temp_db = private_temp_db(self)
         self.tmp_root = Path(tempfile.mkdtemp())
         self.custom_nodes = self.tmp_root / "custom_nodes"
         self.custom_nodes.mkdir()

tests/test_api_prompts.py

@@ -5,7 +5,6 @@ Route tests for the prompt API (py/api/prompts.py).
 import json
 import os
 import sys
-import tempfile
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -29,6 +28,7 @@ except ImportError:  # discovered with tests/ as the top-level directory
 
 from database.operations import PromptDatabase  # noqa: E402
 from py.api import PromptManagerAPI  # noqa: E402
+from tests.private_temp import private_temp_db  # noqa: E402
 import py.api.prompts as prompts_module  # noqa: E402
 from py.api.prompts import (  # noqa: E402
     BULK_CATEGORY_MAX_PROMPTS,
@@ -43,8 +43,7 @@ class PromptAPITestCase(AioHTTPTestCase):
     """App with PromptManager routes and a temporary SQLite database."""
 
     async def get_application(self):
-        self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
-        self._temp_db.close()
+        self._temp_db = private_temp_db(self)
 
         app = web.Application()
         routes = web.RouteTableDef()

tests/test_execution_history_writer.py

@@ -31,6 +31,12 @@ from utils.execution_history_writer import HistoryWriter
 # 0.19 s on macOS py3.13 (PR #351).
 NON_BLOCKING_CEILING = 1.0
 
+# Saturation tests block the worker while they queue 1023 more items. The
+# block must hold until the test releases it (#346/#352): a 3 s timer let a
+# loaded runner unblock the worker mid-fill and drain it. This is only a hang
+# guard; the test, or cleanup, always sets the release.
+SATURATION_HOLD_SECONDS = 60
+
 
 class ManualClock:
     def __init__(self):
@@ -84,6 +90,26 @@ class HistoryWriterTests(unittest.TestCase):
             threading.Event().wait(0.005)
         self.fail("finite deadline expired waiting for worker behavior")
 
+    def wait_for_worker_pass(self, writer):
+        """Wait until the idle worker has run one whole loop pass from now.
+
+        A worker with no accepted work ends every pass by parking on its wake
+        event. The first park counted may end a pass already under way; the
+        second ends a pass that began after this call, so whatever the worker
+        decides at the current clock value has happened (state, not a
+        wall-clock pause that a loaded runner can outlast).
+        """
+        wake = writer._wake
+        park = wake.wait
+        parks = []
+
+        def counted(timeout=None):
+            parks.append(timeout)
+            return park(timeout)
+
+        with patch.object(wake, "wait", counted):
+            self.wait_for(lambda: len(parks) >= 2)
+
     def writer(self, factory=None, ready=True):
         writer = HistoryWriter(factory or (lambda: self.db), clock=self.clock)
         self.writers.append(writer)
@@ -301,7 +327,7 @@ class HistoryWriterTests(unittest.TestCase):
 
         def write(command):
             entered.set()
-            release.wait(3)
+            release.wait(SATURATION_HOLD_SECONDS)
             return original(command)
 
         with patch.object(self.db, "write_execution_start", write):
@@ -348,9 +374,12 @@ class HistoryWriterTests(unittest.TestCase):
         first = self.sealed(writer)
         charge = first.lease.byte_count
 
+        left = threading.Event()
+
         def write(*args, **kwargs):
             entered.set()
             release.wait(3)
+            left.set()
             return original(*args, **kwargs)
 
         with patch.object(self.db, "write_execution_contexts_for_finish", write):
@@ -396,8 +425,21 @@ class HistoryWriterTests(unittest.TestCase):
                 before = monotonic()
                 self.assertFalse(writer.close(0.02))
                 elapsed = monotonic() - before
+            # close() waits until its own monotonic deadline, so the lower
+            # bound holds on any runner.
             self.assertGreaterEqual(elapsed, 0.02)
-            self.assertLess(elapsed, 0.2)
+            if early_join:
+                # Manual clock: only the patched join/sleep advance it, so this
+                # exact bound proves early joins do not spin forever.
+                self.assertLess(elapsed, 0.2)
+            else:
+                # Real clock: a loaded runner may run late, but close() must
+                # not wait anywhere near the 3 s the blocked write holds.
+                self.assertLess(elapsed, 0.02 + NON_BLOCKING_CEILING)
+            # Bounded on the real clock too: close() gave up while the
+            # in-flight write was still blocked, instead of waiting it out.
+            self.assertFalse(left.is_set())
+            self.assertTrue(writer.status().alive)
             self.assertEqual(pool.charged_bytes, charge)
             replacement = HistoryWriter(lambda: self.db, context_budget=pool)
             self.writers.append(replacement)
@@ -771,7 +813,7 @@ class HistoryWriterTests(unittest.TestCase):
             self.wait_for(lambda: not writer.status().accepting)
             self.assertFalse(writer.submit(fresh))
             self.clock.advance(4.99)
-            threading.Event().wait(0.08)
+            self.wait_for_worker_pass(writer)
             self.assertEqual(probes, [0])
             self.clock.advance(0.01)
             self.wait_for(lambda: writer.status().accepting)
@@ -794,12 +836,12 @@ class HistoryWriterTests(unittest.TestCase):
         writer = self.writer(factory, ready=False)
         self.wait_for(lambda: len(calls) == 1)
         self.clock.advance(4.99)
-        threading.Event().wait(0.08)
+        self.wait_for_worker_pass(writer)
         self.assertEqual(calls, [0])
         self.clock.advance(0.01)
         self.wait_for(lambda: len(calls) == 2)
         self.clock.advance(4.99)
-        threading.Event().wait(0.08)
+        self.wait_for_worker_pass(writer)
         self.assertEqual(calls, [0, 5])
         self.clock.advance(0.01)
         self.wait_for(lambda: writer.status().accepting)
@@ -853,7 +895,7 @@ class HistoryWriterTests(unittest.TestCase):
                 )
             writer.invalidate()
             release.set()
-            threading.Event().wait(0.08)
+            self.wait_for_worker_pass(writer)
             self.assertFalse(writer.status().accepting)
             self.assertIsNone(writer.status().epoch)
             self.assertFalse(writer.submit(self.start_command(epoch=stale_epoch)))
@@ -871,7 +913,7 @@ class HistoryWriterTests(unittest.TestCase):
         lock.execute("BEGIN IMMEDIATE")
         before = time.monotonic()
         self.assertTrue(writer.submit(command))
-        self.assertLess(time.monotonic() - before, 0.1)
+        self.assertLess(time.monotonic() - before, NON_BLOCKING_CEILING)
         self.wait_for(lambda: not writer.status().accepting)
         lock.rollback()
         self.clock.advance(5)

tests/test_execution_output_writer.py

@@ -18,6 +18,7 @@ from utils.execution_output_capture import seal_outputs
 from utils.execution_outputs import snapshot_output_roots
 from utils.execution_history_writer import HistoryWriter
 from tests.history_waits import drain_writer
+from tests.disk_waits import DISK_BOUND_SECONDS
 
 try:
     from tests import test_execution_history_writer as fixtures
@@ -28,11 +29,17 @@ except ImportError:
 # Upper bound a blocked fake write waits for its release. Tests prove ordering
 # with events, so this only caps a failing run and never decides a pass.
 BLOCK_SECONDS = 10
+
+# Saturation tests block the worker while they queue 1023 more items. The
+# block must hold until the test releases it (#346/#352): a 3 s timer let a
+# loaded runner unblock the worker mid-fill and drain it. This is only a hang
+# guard; the test, or cleanup, always sets the release.
+SATURATION_HOLD_SECONDS = 60
+
 # A restore is a full verify/migrate/backup/replace cycle and is bound by the
 # disk: a busy Windows runner took over 10 s (merge queue, #382). Waits that
-# span a restore use this ceiling, which stays under the runner's 120 s hang
-# watchdog, so a real hang still fails with the thread dump.
-RESTORE_SECONDS = 90
+# span a restore use the shared disk-bound hang guard (tests/disk_waits.py).
+RESTORE_SECONDS = DISK_BOUND_SECONDS
 
 
 class OutputWriterTests(fixtures.unittest.TestCase):
@@ -591,7 +598,7 @@ class OutputWriterTests(fixtures.unittest.TestCase):
 
         def write(command):
             entered.set()
-            release.wait(3)
+            release.wait(SATURATION_HOLD_SECONDS)
             return original(command)
 
         with patch.object(self.db, "write_execution_start", write):

tests/vendor/ci-workflow.test.mjs

@@ -247,3 +247,32 @@ test('the Python matrix and coverage run the suite through the watchdog runner',
         assert.doesNotMatch(found.get(id), /unittest discover/, `${id} no longer runs the bare unittest discover`);
     }
 });
+
+// CI hygiene (#352): lint tools are pinned, the bandit hook can fail, the JS job caches npm.
+const readRepoFile = path => readFileSync(new URL(`../../${path}`, import.meta.url), 'utf8');
+
+test('black and flake8 are pinned exactly and the lint job installs those pins', () => {
+    const requirements = readRepoFile('requirements-dev.txt');
+    assert.match(requirements, /^black==\d+\.\d+\.\d+$/m, 'black is pinned exactly');
+    assert.match(requirements, /^flake8==\d+\.\d+\.\d+$/m, 'flake8 is pinned exactly');
+    const lint = jobs(readWorkflow('test.yml')).get('lint');
+    assert.match(lint, /pip install \$\(grep -E '\^\(black\|flake8\)==' requirements-dev\.txt\)/);
+    assert.doesNotMatch(lint, /pip install black flake8/, 'lint no longer installs the latest releases');
+});
+
+test('the frontend modules job caches npm keyed on the vendor lockfile', () => {
+    const setupNode = steps(jobs(readWorkflow('test.yml')).get('js-test'))
+        .find(step => /uses: actions\/setup-node@v\d+/.test(step));
+    assert.match(setupNode, /^ {10}cache: npm$/m);
+    assert.match(setupNode, /^ {10}cache-dependency-path: tools\/vendor\/package-lock\.json$/m);
+});
+
+test('the bandit pre-commit hook fails at the severity CI enforces and passes every file', () => {
+    const quality = readWorkflow('code-quality.yml');
+    assert.match(quality, /^ {8}run: bandit -r \. -lll -q --exclude /m, 'CI fails on high severity');
+    const hook = /- id: bandit\n([\s\S]*?)(?=\n {6}(?:- id:|#)|\n*$)/.exec(readRepoFile('.pre-commit-config.yaml'))?.[1];
+    assert.ok(hook, 'the bandit hook exists');
+    assert.match(hook, /^ {8}entry: bandit -lll -q$/m, 'the hook enforces the same -lll level as CI');
+    assert.doesNotMatch(hook, /true|exit-zero|bash -c/, 'nothing masks the exit code or drops a file');
+    assert.doesNotMatch(hook, /pass_filenames: false/, 'the hook scans the staged files');
+});