test: isolate API fixture databases from other suites; CI lint pins, failing bandit hook, npm cache (#346, #352, #409) (#411)

* ci: pin black and flake8, let the bandit hook fail, cache npm in the JS job (#352)

The lint job installs black==26.10.1 and flake8==7.4.1 from
requirements-dev.txt (the versions CI resolved on main today), so a new
release cannot turn CI red with no code change.

The local bandit hook ran 'bash -c ... "$@"; true': it could never
fail, and bash bound the first staged file to $0 so it was never
scanned. It now runs 'bandit -lll -q' directly, failing on high
severity exactly like code-quality.yml.

The frontend modules job's setup-node caches npm keyed on
tools/vendor/package-lock.json.

* test(history): wait on the writer's state, not short wall-clock pauses (#352)

Four 0.08 s pauses before asserting that the worker had not probed yet
now wait until the idle worker has run a whole loop pass at the current
manual clock value (three clock reads), so a slow runner cannot end the
pause before the worker decides. A mutation that probes 20 ms early now
fails two tests deterministically.

The real-clock bounded close no longer asserts close(0.02) returned in
under 0.2 s wall time; it asserts the in-flight write is still blocked
and the worker alive, like the output writer test. The manual-clock
bound stays exact. The last 0.1 s non-blocking submit check uses the
1 s NON_BLOCKING_CEILING from #351.

* test(api): give each API fixture database its own temp folder (#346)

Every PromptManagerAPI sweeps its database's folder at startup for the
stages, uploads and safety-backup copies an interrupted restore left.
The API test fixtures created their database directly in the system
temp folder, so any other test process starting an API there (another
worktree's suite, a pre-push hook, an agent's targeted run) deleted
this process's in-flight restore files. The restore, epoch and
retirement tests then failed with 'staged restore changed', 'unable to
open database file' or 503 instead of 409. Busy machines fail because
they are running other suites, not because they are slow: the same
CPU load with a private TMPDIR passes.

private_temp_db() puts the fixture database in a new folder removed at
cleanup (with any safety backups a restore left beside it). A
regression test sweeps a stand-in shared folder in the middle of a
multipart restore, as another suite's startup would.

* test(history): count worker parks, hold saturation blocks until released (#352)

wait_for_worker_pass now counts the idle worker's parks on its wake
event (two parks from now = one whole pass begun after the call)
instead of assuming a fixed number of clock reads per pass, which a
future clock read would have made vacuous. The real-clock bounded
close keeps a loose upper bound (0.02 s + NON_BLOCKING_CEILING).

The two saturation tests (history and output writer) blocked the
worker with release.wait(3) while they queued 1023 more items. On a
loaded runner the fill took longer than 3 s, the worker unblocked on
its own and drained part of the queue (pending 868..1017 != 1024),
and cleanup then reported a leaked worker. The block now holds until
the test releases it; 60 s is only a hang guard, and cleanup always
sets the release. Under one core with 8 busy loops: 8/8 runs failed
before, 6/6 pass after.

* test(maintenance): wait on duplicate jobs with a disk-bound hang guard (#409)

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');
+});