diff --git a/tests/background_stops.py b/tests/background_stops.py new file mode 100644 index 0000000..9aeb07a --- /dev/null +++ b/tests/background_stops.py @@ -0,0 +1,27 @@ +"""Final stops for the background threads a test starts (#301, #302). + +Every real ``ImageGenerationHandler``, ``ImageMonitor`` or ``PromptTracker`` +a test constructs ends with its final stop and an assertion, before the +test's folder and databases go: ``close()`` for a handler, +``stop_monitoring()`` for a monitor, ``shutdown()`` for a tracker. +""" + +from utils.prompt_tracker import PromptTracker + + +def stop_tracker(testcase, tracker): + """Shut a real tracker's cleanup thread down; mocks are left alone.""" + if isinstance(tracker, PromptTracker): + testcase.assertTrue(tracker.shutdown(), "prompt cleanup thread still running") + + +def close_handler(testcase, handler): + """Close ``handler`` for good, then stop the tracker it was built with.""" + testcase.assertTrue(handler.close(), "image worker still running") + stop_tracker(testcase, getattr(handler, "prompt_tracker", None)) + + +def stop_monitor(testcase, monitor): + """Stop ``monitor`` (its observer and worker), then its tracker.""" + testcase.assertTrue(monitor.stop_monitoring(), "image monitor still running") + stop_tracker(testcase, getattr(monitor, "prompt_tracker", None)) diff --git a/tests/browser/workspace_system_fixture.py b/tests/browser/workspace_system_fixture.py index 0fc36d9..9900abc 100644 --- a/tests/browser/workspace_system_fixture.py +++ b/tests/browser/workspace_system_fixture.py @@ -97,15 +97,25 @@ def register_system_commands(app, routes, api, root: Path): real_move = library_move.move_library keep = PromptManagerConfig.KEEP_LIBRARY_IN_NODE_FOLDER config_bytes = config_path.read_bytes() if config_path.exists() else None - host_values = ( - PromptManagerConfig.RESULT_TIMEOUT, - PromptManagerConfig.WEBUI_DISPLAY_MODE, - PromptManagerConfig.WORKER_THREADS, - GalleryConfig.MONITORING_ENABLED, - PromptManagerConfig.THUMBNAIL_FORMAT, - PromptManagerConfig.THUMBNAIL_QUALITY, - PromptManagerConfig.THUMBNAIL_EDGE, + from py.api.settings_routes import ( + _AUTOTAG_SETTINGS, + _GALLERY_SETTINGS, + _SETTINGS_FILE_KEYS, ) + + # Every value POST /settings can change, so a reset restores each one + # (a saved safety_backup=false once leaked into a later spec's review). + # The tagging settings are the tagging fixture's to reset. + host_values = { + (GalleryConfig if name in _GALLERY_SETTINGS else PromptManagerConfig, name): ( + getattr( + GalleryConfig if name in _GALLERY_SETTINGS else PromptManagerConfig, + name, + ) + ) + for name in _SETTINGS_FILE_KEYS + if name not in _AUTOTAG_SETTINGS + } from py.api import thumbnail_encode monitor = getattr(image_monitor, "_monitor_instance", None) @@ -135,15 +145,8 @@ def register_system_commands(app, routes, api, root: Path): unplugged.rmdir() GalleryConfig.MONITORING_DIRECTORIES = list(configured) api._health_report = None - ( - PromptManagerConfig.RESULT_TIMEOUT, - PromptManagerConfig.WEBUI_DISPLAY_MODE, - PromptManagerConfig.WORKER_THREADS, - GalleryConfig.MONITORING_ENABLED, - PromptManagerConfig.THUMBNAIL_FORMAT, - PromptManagerConfig.THUMBNAIL_QUALITY, - PromptManagerConfig.THUMBNAIL_EDGE, - ) = host_values + for (owner, name), value in host_values.items(): + setattr(owner, name, value) thumbnail_encode.reset_encoders() image_monitor._monitor_instance = monitor api.__dict__.pop("_library_resolver", None) diff --git a/tests/comfyui/fixtures/ledger_nodes.py b/tests/comfyui/fixtures/ledger_nodes.py index c65199c..80914ff 100644 --- a/tests/comfyui/fixtures/ledger_nodes.py +++ b/tests/comfyui/fixtures/ledger_nodes.py @@ -1,5 +1,6 @@ """Deterministic nodes and queue controls installed only in an owned test host.""" +import asyncio import os import threading import time @@ -291,7 +292,8 @@ async def image_monitor(request: web.Request) -> web.Response: config.MONITORING_ENABLED = True monitor.start_monitoring([folder_paths.get_output_directory()]) else: - monitor.stop_monitoring() + # Joins the image worker and the observer (up to 5 s each): off the loop. + await asyncio.to_thread(monitor.stop_monitoring) config.MONITORING_ENABLED = False return web.json_response({"monitoring": monitor.is_monitoring}) diff --git a/tests/test_execution_history_hooks.py b/tests/test_execution_history_hooks.py index ae790d2..b0f5454 100644 --- a/tests/test_execution_history_hooks.py +++ b/tests/test_execution_history_hooks.py @@ -28,6 +28,8 @@ except ImportError: Status = namedtuple("ExecutionStatus", "status_str completed messages") JOB = "a0c26da3-eb26-405c-b143-c662a84c3112" +# A cross-thread wait or a duration check fails only past this (#370). +CEILING_SECONDS = 30.0 class HostQueue: @@ -209,6 +211,9 @@ class HistoryHooksTests(unittest.TestCase): self.queue.get = unrelated_after self.dequeue(0) self.queue.task_done(0, {}, status()) + # The writer drains first: this test checks wrappers and the session, + # not the 2 s bound, which a slow runner can spend on the write (#370). + self.drain(first) self.assertTrue(first.dispose()) self.assertTrue(first.dispose()) self.assertIs(self.queue.get, unrelated_after) @@ -339,15 +344,17 @@ class HistoryHooksTests(unittest.TestCase): with patch.object(self.db, "write_execution_start", blocked): registration = self.register() self.dequeue(0) - self.assertTrue(entered.wait(1)) + # Ceilings, not tight wall clocks (#370): the return values show + # the bound; a ceiling still fails a call that waits unbounded. + self.assertTrue(entered.wait(CEILING_SECONDS)) began = time.monotonic() self.queue.task_done(0, {}, status()) self.dequeue(1) - self.assertLess(time.monotonic() - began, 0.2) + self.assertLess(time.monotonic() - began, CEILING_SECONDS) began = time.monotonic() self.assertFalse(registration.dispose()) self.assertGreaterEqual(time.monotonic() - began, 1.8) - self.assertLess(time.monotonic() - began, 2.3) + self.assertLess(time.monotonic() - began, CEILING_SECONDS) release.set() self.assertTrue(registration.dispose(2)) diff --git a/tests/test_execution_history_writer.py b/tests/test_execution_history_writer.py index ec6327d..1d93dcb 100644 --- a/tests/test_execution_history_writer.py +++ b/tests/test_execution_history_writer.py @@ -642,6 +642,8 @@ class HistoryWriterTests(unittest.TestCase): == "succeeded" ) self.assertTrue(writer.submit(self.start_command(2))) + # Drain first: the 2 s bound is not under test here (#370). + drain_writer(writer) self.assertTrue(writer.close(2)) self.assertEqual(len(set(threads)), 1) self.assertNotEqual(threads[0], threading.get_ident()) diff --git a/tests/test_execution_output_hooks.py b/tests/test_execution_output_hooks.py index cf27a8c..567a596 100644 --- a/tests/test_execution_output_hooks.py +++ b/tests/test_execution_output_hooks.py @@ -245,6 +245,8 @@ class OutputHooksTests(fixtures.unittest.TestCase): registration = self.register() self.dequeue(0) late = registration._task_done + # Drain first: the 2 s bound is not under test here (#370). + self.drain(registration) self.assertTrue(registration.dispose()) thread = threading.Thread( target=lambda: late((0, history(), fixtures.status()), {}) diff --git a/tests/test_execution_output_writer.py b/tests/test_execution_output_writer.py index c91a49e..c9f89d4 100644 --- a/tests/test_execution_output_writer.py +++ b/tests/test_execution_output_writer.py @@ -17,6 +17,7 @@ from utils.execution_contexts import MAX_ATTEMPT_BYTES 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 try: from tests import test_execution_history_writer as fixtures @@ -161,6 +162,8 @@ class OutputWriterTests(fixtures.unittest.TestCase): start = replace(self.start_command(), run_id=str(fixtures.uuid4())) self.assertTrue(writer.submit(start)) self.finalization(writer, start) + # Drain first: the 2 s bound is not under test here (#370). + drain_writer(writer) self.assertTrue(writer.close(2)) self.assertEqual( self.db.get_execution_run(start.run_id)["outcome"], "succeeded" diff --git a/tests/test_hashing.py b/tests/test_hashing.py index 9a262be..ad02cdf 100644 --- a/tests/test_hashing.py +++ b/tests/test_hashing.py @@ -12,6 +12,7 @@ import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from tests.background_stops import stop_tracker # noqa: E402 from utils.hashing import ( generate_prompt_hash, is_duplicate_prompt, @@ -49,6 +50,7 @@ class TestSingleHashImplementation(unittest.TestCase): db = mock.Mock() db.get_prompt_by_hash.return_value = {"id": 42} tracker = PromptTracker(db) + self.addCleanup(stop_tracker, self, tracker) tracker.set_current_prompt(" Hello World ") db.get_prompt_by_hash.assert_called_once_with( generate_prompt_hash(" Hello World ") diff --git a/tests/test_image_linking.py b/tests/test_image_linking.py index 65cc015..3cc8586 100644 --- a/tests/test_image_linking.py +++ b/tests/test_image_linking.py @@ -8,6 +8,7 @@ import unittest.mock as mock sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from tests.background_stops import close_handler # noqa: E402 from database.operations import PromptDatabase from utils.hashing import generate_prompt_hash from utils.image_monitor import ImageGenerationHandler @@ -52,6 +53,7 @@ class LinkingTestCase(unittest.TestCase): ) def tearDown(self): + close_handler(self, self.handler) # and the tracker's cleanup thread self.db.close_all() # Windows cannot unlink an open database file for path in ( self.image, diff --git a/tests/test_image_monitor_singleton.py b/tests/test_image_monitor_singleton.py index c77811c..0b90716 100644 --- a/tests/test_image_monitor_singleton.py +++ b/tests/test_image_monitor_singleton.py @@ -136,13 +136,21 @@ class TestMonitorSingletonLifecycle(unittest.TestCase): class FakeObserver: - """Records what the monitor does with the watchdog observer.""" + """Records what the monitor does with the watchdog observer. - def __init__(self): + ``calls`` lists ``stop`` and ``join`` in order (shared with a test's own + recorders when given); ``stay_alive`` keeps it alive after ``join``; + ``stop_error`` makes ``stop`` raise. + """ + + def __init__(self, calls=None, stay_alive=False, stop_error=None): self.scheduled = [] self.started = False self.stopped = False self.joined = False + self.calls = calls if calls is not None else [] + self.stay_alive = stay_alive + self.stop_error = stop_error def schedule(self, handler, path, recursive=False): self.scheduled.append((handler, path, recursive)) @@ -151,12 +159,18 @@ class FakeObserver: self.started = True def stop(self): + self.calls.append(("stop",)) + if self.stop_error is not None: + raise self.stop_error self.stopped = True def join(self, timeout=None): + self.calls.append(("join", timeout)) self.joined = True def is_alive(self): + if self.stay_alive: + return self.started return self.started and not self.stopped @@ -171,6 +185,8 @@ class TestStartStop(unittest.TestCase): self.missing = os.path.join(self.tmp.name, "does-not-exist") def tearDown(self): + # Before the folder is removed (addCleanup runs after tearDown). + self.assertTrue(self.monitor.stop_monitoring()) im_mod._monitor_instance = None def test_only_nonexistent_directories_leaves_monitoring_off(self): @@ -225,9 +241,8 @@ class TestStartStop(unittest.TestCase): worker = handler._worker self.assertTrue(worker.is_alive()) - self.monitor.stop_monitoring() + self.assertTrue(self.monitor.stop_monitoring()) - worker.join(timeout=2) self.assertFalse(worker.is_alive()) self.assertEqual(handler.pending_count(), 0) # A restart gets a fresh handler and observer @@ -303,6 +318,61 @@ class TestStartStop(unittest.TestCase): self.assertEqual(self.monitor.monitored_directories, [self.tmp.name]) self.monitor.stop_monitoring() + def _start_with(self, fake): + with patch.object(im_mod, "Observer", return_value=fake): + self.monitor.start_monitoring([self.tmp.name]) + handler = self.monitor.handler + real_close = handler.close + + def close(timeout=im_mod.WORKER_JOIN_SECONDS): + fake.calls.append(("close",)) + return real_close(timeout) + + handler.close = close + return handler + + def test_stop_monitoring_joins_observer_and_worker(self): + fake = FakeObserver() + handler = self._start_with(fake) + + self.assertTrue(self.monitor.stop_monitoring(timeout=3.0)) + + self.assertEqual([c[0] for c in fake.calls], ["close", "stop", "join"]) + join_timeout = fake.calls[2][1] + self.assertIsNotNone(join_timeout) + self.assertLessEqual(join_timeout, 3.0) + self.assertTrue(handler._closed) + + def test_stop_monitoring_reports_an_observer_still_running(self): + fake = FakeObserver(stay_alive=True) + self._start_with(fake) + with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs: + self.assertFalse(self.monitor.stop_monitoring(timeout=0.05)) + self.assertIn("watchdog observer still running", "\n".join(logs.output)) + self.assertFalse(self.monitor.is_monitoring) + fake.stay_alive = False # it ends; tearDown's stop waits for it + + def test_a_second_stop_monitoring_waits_for_the_observer(self): + fake = FakeObserver(stay_alive=True) + self._start_with(fake) + with self.assertLogs("prompt_manager.image_monitor", "WARNING"): + self.assertFalse(self.monitor.stop_monitoring(timeout=0.05)) + self.assertFalse(self.monitor.stop_monitoring(timeout=0.05)) + fake.stay_alive = False + self.assertTrue(self.monitor.stop_monitoring()) + self.assertFalse(self.monitor._unfinished) + + def test_stop_monitoring_reports_an_observer_stop_that_raised(self): + fake = FakeObserver(stop_error=OSError("descriptor gone")) + self._start_with(fake) + with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs: + self.assertFalse(self.monitor.stop_monitoring()) + self.assertIn("Error while stopping image monitoring", "\n".join(logs.output)) + fake.stop_error = None # tearDown's stop calls stop() again + + def test_stop_monitoring_without_an_observer_returns_true(self): + self.assertTrue(self.monitor.stop_monitoring()) + class FakeEvent: def __init__(self, path, is_directory=False): @@ -335,6 +405,10 @@ class TestHandlerScheduling(unittest.TestCase): self.processed.append((path, prompt_snapshot)) ) + def tearDown(self): + # Before the folder is removed (addCleanup runs after tearDown). + self.assertTrue(self.handler.close()) + def _image(self, name, content=b"x" * 16): path = os.path.join(self.tmp.name, name) with open(path, "wb") as f: @@ -399,23 +473,127 @@ class TestHandlerScheduling(unittest.TestCase): empty = self._image("empty.png", content=b"") self.assertFalse(self.handler.wait_until_settled(empty)) - def test_shutdown_stops_the_worker_and_clears_pending(self): + def test_shutdown_joins_the_worker_within_the_bound(self): self.handler.processing_delay = 10 # keep the first image waiting self.handler.schedule(self._image("s1.png")) self.handler.schedule(self._image("s2.png")) worker = self.handler._worker self.assertTrue(worker.is_alive()) - self.handler.shutdown() + self.assertTrue(self.handler.shutdown()) - worker.join(timeout=2) self.assertFalse(worker.is_alive()) self.assertEqual(self.handler.pending_count(), 0) + def test_shutdown_reports_a_worker_still_busy(self): + entered, release = threading.Event(), threading.Event() + + def busy(path, prompt_snapshot=None): + entered.set() + release.wait(30) + + self.handler.process_new_image = busy + self.handler.schedule(self._image("busy.png")) + worker = self.handler._worker + self.addCleanup(worker.join, 30) + self.addCleanup(release.set) + self.assertTrue(entered.wait(30), "the worker never took the image") + + with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs: + self.assertFalse(self.handler.shutdown(timeout=0.05)) + self.assertIn("image worker still running", "\n".join(logs.output)) + + release.set() + worker.join(30) + self.assertFalse(worker.is_alive()) + + def test_a_second_stop_waits_for_a_worker_still_finishing(self): + entered, release = threading.Event(), threading.Event() + + def busy(path, prompt_snapshot=None): + entered.set() + release.wait(30) + + self.handler.process_new_image = busy + self.handler.schedule(self._image("twice.png")) + worker = self.handler._worker + self.addCleanup(worker.join, 30) + self.addCleanup(release.set) + self.assertTrue(entered.wait(30), "the worker never took the image") + + with self.assertLogs("prompt_manager.image_monitor", "WARNING"): + self.assertFalse(self.handler.close(timeout=0.05)) + self.assertFalse(self.handler.close(timeout=0.05)) + release.set() + self.assertTrue(self.handler.close(timeout=30)) + self.assertFalse(worker.is_alive()) + + def test_shutdown_on_the_worker_itself_only_signals(self): + entered, result = threading.Event(), {} + + def stop_from_inside(path, prompt_snapshot=None): + result["value"] = self.handler.shutdown() + entered.set() + + self.handler.process_new_image = stop_from_inside + self.handler.schedule(self._image("self.png")) + worker = self.handler._worker + self.assertTrue(entered.wait(30), "the worker never took the image") + worker.join(30) + self.assertFalse(result["value"]) + self.assertFalse(worker.is_alive()) + + def _live_workers(self): + return { + t for t in threading.enumerate() if t.name == "PromptManagerImageWorker" + } + + def _seal(self, handler, name): + handler._await_history(os.path.join(self.tmp.name, name)) + handler._on_outputs_sealed( + "run-1", + [{"folder_type": "output", "subfolder": "", "filename": name}], + ) + + def test_close_refuses_new_work(self): + path = self._image("late.png") + with patch.object(im_mod, "host_output_directory", return_value=self.tmp.name): + self.assertTrue(self.handler.close()) + before = self._live_workers() + self.assertFalse(self.handler.schedule(path)) + # The image in hand can still ask for a history re-check and its + # run can be sealed afterwards, on the history writer's thread. + self.handler._await_history(path) + self.assertEqual(self.handler.awaiting_history(), []) + self._seal(self.handler, "late.png") + self.handler._schedule_recheck(path) + self.assertIsNone(self.handler._worker) + self.assertEqual(list(self.handler._history_due), []) + self.assertEqual(self._live_workers() - before, set()) + + def test_a_sealed_run_starts_a_worker_on_an_open_handler(self): + self._image("open.png") + with patch.object(im_mod, "host_output_directory", return_value=self.tmp.name): + self._seal(self.handler, "open.png") + self.assertIsNotNone(self.handler._worker) + self.assertTrue(self.handler.close()) + + def test_close_is_idempotent(self): + self.handler.schedule(self._image("c1.png")) + self.assertTrue(self.handler.close()) + self.assertTrue(self.handler.close()) + + def test_on_created_after_close_takes_no_lock(self): + self.assertTrue(self.handler.close()) + self.handler.on_created(FakeEvent(self._image("after.png"))) + self.tracker.keyed_snapshot.assert_not_called() + self.assertEqual(self.handler.pending_count(), 0) + def test_shutdown_is_idempotent_and_safe_before_start(self): fresh = im_mod.ImageGenerationHandler(MagicMock(), self.tracker) - fresh.shutdown() - fresh.shutdown() + self.assertTrue(fresh.shutdown()) + self.assertTrue(fresh.shutdown()) + self.assertTrue(fresh.close()) self.handler.schedule(self._image("s3.png")) self.assertTrue(wait_for(lambda: self.handler.pending_count() == 0)) self.handler.shutdown() @@ -434,7 +612,7 @@ class TestHandlerScheduling(unittest.TestCase): self.assertTrue(wait_for(lambda: self.handler.pending_count() == 0)) self.assertIn(path, [p for p, _ in self.processed]) self.assertIsNot(self.handler._worker, old_worker) - self.handler.shutdown() + self.assertTrue(self.handler.close()) def test_processing_errors_do_not_kill_the_worker(self): def explode(path, prompt_snapshot=None): @@ -449,6 +627,47 @@ class TestHandlerScheduling(unittest.TestCase): self.assertEqual(self.processed, [first, second]) +class TestCloseReleasesHandles(unittest.TestCase): + """``close`` joins the worker, so the test folder can be removed (#302).""" + + def test_close_releases_handles(self): + from PIL import Image + + from database.operations import PromptDatabase + from tests.open_handles import assert_closed + from utils.prompt_tracker import PromptTracker + + with tempfile.TemporaryDirectory() as folder: + db = PromptDatabase(os.path.join(folder, "prompts.db")) + tracker = PromptTracker(db) + handler = im_mod.ImageGenerationHandler(db, tracker) + handler.processing_delay = 0 + handler.settle_interval = 0.01 + done = threading.Event() + real_process = handler.process_new_image + + def process(path, prompt_snapshot=None): + try: + real_process(path, prompt_snapshot=prompt_snapshot) + finally: + done.set() + + handler.process_new_image = process + path = os.path.join(folder, "real.png") + Image.new("RGB", (8, 8), "red").save(path) + try: + self.assertTrue(handler.schedule(path)) + worker = handler._worker + self.assertTrue(done.wait(30), "the worker never processed the image") + finally: + closed, shut = handler.close(), tracker.shutdown() + db.model.close_all() + self.assertTrue(closed) + self.assertTrue(shut) + self.assertFalse(worker.is_alive()) + assert_closed(self, folder) + + class TestOutputListener(unittest.TestCase): """Every settled output is reported to the Images listing (#348, PR 2).""" @@ -456,7 +675,7 @@ class TestOutputListener(unittest.TestCase): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.handler = im_mod.ImageGenerationHandler(MagicMock(), MagicMock()) - self.addCleanup(self.handler.shutdown) + self.addCleanup(lambda: self.assertTrue(self.handler.close())) self.handler.processing_delay = 0 self.handler.settle_interval = 0.01 self.handler.process_new_image = lambda path, prompt_snapshot=None: None @@ -549,7 +768,7 @@ class TestOutputListener(unittest.TestCase): self.assertIsNone(monitor.output_listener) self.assertIsNone(monitor.handler.output_listener) finally: - monitor.stop_monitoring() + self.assertTrue(monitor.stop_monitoring()) class TestModuleIsolation(unittest.TestCase): diff --git a/tests/test_library_move_review.py b/tests/test_library_move_review.py index 14e5d52..57f61c3 100644 --- a/tests/test_library_move_review.py +++ b/tests/test_library_move_review.py @@ -29,6 +29,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from database import library_move as lm # noqa: E402 from database import models # noqa: E402 from database.operations import PromptDatabase # noqa: E402 +from tests.background_stops import close_handler # noqa: E402 from tests.open_handles import assert_closed # noqa: E402 from utils import library_location as ll # noqa: E402 from utils import usage_tracking # noqa: E402 @@ -204,6 +205,7 @@ class TestBackgroundWritesSurviveTheSwitch(_MoveCase): image = os.path.join(self.comfy, "output", "late.png") writer_db = PromptDatabase(self.old_db) handler = ImageGenerationHandler(writer_db, mock.MagicMock()) + self.addCleanup(close_handler, self, handler) started = threading.Event() probe = _CrossThreadCloseProbe() outcome = {} @@ -256,6 +258,7 @@ class TestBackgroundWritesSurviveTheSwitch(_MoveCase): db = PromptDatabase(self.old_db) handler = ImageGenerationHandler(db, mock.MagicMock()) + self.addCleanup(close_handler, self, handler) image = os.path.join(self.comfy, "output", "orphan.png") with mock.patch.object(handler, "logger") as log: handler.link_image_to_prompt(image, {"id": 99999}, {}) diff --git a/tests/test_nonfinite_metadata.py b/tests/test_nonfinite_metadata.py index 60f536d..b49729d 100644 --- a/tests/test_nonfinite_metadata.py +++ b/tests/test_nonfinite_metadata.py @@ -24,6 +24,7 @@ from unittest.mock import patch from PIL import Image, PngImagePlugin from database.operations import PromptDatabase +from tests.background_stops import close_handler from tests.history_waits import drain_writer from tests.test_api_admin import AdminAPITestCase from tests.test_api_images import ImageAPITestCase @@ -224,6 +225,7 @@ class TestMonitorIngestsNonFiniteMetadata(StrictRows, unittest.TestCase): self.handler.logger.disabled = True def tearDown(self): + close_handler(self, self.handler) self.db.close_all() assert_closed(self, self.temp.name) self.temp.cleanup() diff --git a/tests/test_prompt_attribution.py b/tests/test_prompt_attribution.py index 6171771..7cf3d10 100644 --- a/tests/test_prompt_attribution.py +++ b/tests/test_prompt_attribution.py @@ -23,6 +23,7 @@ try: except ImportError: # discovered with tests/ as the top-level directory from open_handles import assert_closed +from tests.background_stops import close_handler # noqa: E402 from database.operations import PromptDatabase from prompt_manager import PromptManager from prompt_manager_pair import PromptManagerPair @@ -278,6 +279,7 @@ class MonitorCase(DatabaseCase): fd, self.image = tempfile.mkstemp(suffix=".png", dir=self.tmp.name) os.close(fd) self.handler = ImageGenerationHandler(self.db, self.tracker) + self.addCleanup(close_handler, self, self.handler) self.linked = [] self.handler.link_image_to_prompt = lambda path, prompt, meta: ( self.linked.append(prompt["id"]) diff --git a/tests/test_prompt_tracker.py b/tests/test_prompt_tracker.py index a9bfdc8..35a5c9c 100644 --- a/tests/test_prompt_tracker.py +++ b/tests/test_prompt_tracker.py @@ -5,10 +5,12 @@ Tests PromptTracker, PromptExecutionContext, and the singleton get_prompt_tracker. Uses a mock db_manager to avoid database deps. """ +import logging import os import sys import time import threading +import types import unittest from unittest.mock import MagicMock, patch @@ -312,5 +314,93 @@ class TestPromptQueue(unittest.TestCase): self.assertEqual(popped_ids, ids) +def _real_tracker(): + """A PromptTracker built by its own ``__init__``, so its cleanup thread runs.""" + mock_db = MagicMock() + mock_db.get_prompt_by_hash.return_value = None + return PromptTracker(mock_db) + + +class TestTrackerShutdown(unittest.TestCase): + """The cleanup thread stops on a signal and is joined within a bound (#301).""" + + def test_shutdown_stops_the_cleanup_thread_within_the_bound(self): + tracker = _real_tracker() + thread = tracker.cleanup_thread + self.assertTrue(thread.is_alive()) + self.assertEqual(thread.name, "PromptManagerPromptCleanup") + + self.assertTrue(tracker.shutdown()) + self.assertFalse(thread.is_alive()) + + def test_shutdown_is_idempotent(self): + tracker = _real_tracker() + self.assertTrue(tracker.shutdown()) + self.assertTrue(tracker.shutdown()) + + def test_cleanup_error_wait_is_interruptible(self): + import utils.prompt_tracker as tracker_mod + + real_time = time.time + calls = {"n": 0} + + def failing_once(): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("clock unavailable") + return real_time() + + stub = types.SimpleNamespace( + time=failing_once, sleep=time.sleep, monotonic=time.monotonic + ) + logged = threading.Event() + logger = logging.getLogger("prompt_manager.prompt_tracker") + real_error = logger.error + + def error(*args, **kwargs): + real_error(*args, **kwargs) + logged.set() + + with ( + patch.object(tracker_mod, "time", stub), + patch.object(logger, "error", side_effect=error), + ): + tracker = _real_tracker() + self.addCleanup(tracker.shutdown) + self.assertTrue(logged.wait(30), "the cleanup error was never logged") + # The thread is now in its 60 s error wait. + self.assertTrue(tracker.shutdown()) + self.assertFalse(tracker.cleanup_thread.is_alive()) + + def test_tracker_works_after_shutdown(self): + tracker = _real_tracker() + self.assertTrue(tracker.shutdown()) + tracker.set_current_prompt("still tracked", {"prompt_id": 9}) + self.assertEqual(tracker.pop_next_prompt()["id"], 9) + + def test_shutdown_of_a_tracker_built_without_init(self): + self.assertTrue(_make_tracker().shutdown()) + + def test_shutdown_on_the_cleanup_thread_only_signals(self): + tracker = _real_tracker() + self.addCleanup(lambda: self.assertTrue(tracker.shutdown())) + result = {} + with patch.object(tracker, "cleanup_thread", threading.current_thread()): + result["value"] = tracker.shutdown() + self.assertFalse(result["value"]) + self.assertTrue(tracker._stop.is_set()) + + def test_shutdown_reports_a_thread_still_running(self): + tracker = _real_tracker() + self.assertTrue(tracker.shutdown()) + busy = MagicMock() + busy.is_alive.return_value = True + tracker.cleanup_thread = busy + with self.assertLogs("prompt_manager.prompt_tracker", "WARNING") as logs: + self.assertFalse(tracker.shutdown(timeout=0.05)) + busy.join.assert_called_once_with(0.05) + self.assertIn("prompt cleanup thread still running", "\n".join(logs.output)) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_review.py b/tests/test_review.py index a313e31..ce79a32 100644 --- a/tests/test_review.py +++ b/tests/test_review.py @@ -27,6 +27,7 @@ except ImportError: # discovered with tests/ as the top-level directory from PIL import Image, PngImagePlugin +from tests.background_stops import close_handler, stop_monitor from database.operations import PromptDatabase from database.review_schema import canonical_evidence, evidence_digest from py.api import PromptManagerAPI @@ -790,6 +791,7 @@ class TestReviewEdges(ReviewTestCase): class TestMonitorRecording(ReviewTestCase): def _handler(self): handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db)) + self.addCleanup(close_handler, self, handler) return handler def _linked_prompt(self, path): @@ -953,7 +955,7 @@ class TestMonitorRecording(ReviewTestCase): monitor.set_review_recorder(None) self.assertIsNone(monitor.handler.review_recorder) finally: - monitor.stop_monitoring() + stop_monitor(self, monitor) if __name__ == "__main__": @@ -1092,6 +1094,7 @@ class TestCaptureStoreShutdown(ReviewTestCase): self._scan_record(path) # The monitor's pass records nothing, so it needs no store. handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db)) + self.addCleanup(close_handler, self, handler) handler.review_recorder = self.api.record_review_file handler._record_review(str(path)) self.assertEqual(self.api.record_review_file(str(path)), "skipped") diff --git a/tests/test_review_image_links_tool.py b/tests/test_review_image_links_tool.py index a88fb10..7941e53 100644 --- a/tests/test_review_image_links_tool.py +++ b/tests/test_review_image_links_tool.py @@ -17,6 +17,7 @@ except ImportError: # discovered with tests/ as the top-level directory from open_handles import assert_closed from database.operations import PromptDatabase +from tests.background_stops import close_handler from tests.test_api_images import ImageAPITestCase from tests.test_prompt_attribution import issue_graph, pair_evidence from tools import review_image_links as tool @@ -158,8 +159,8 @@ class ReviewToolTests(ToolFixture): def test_conflicting_sources_are_left_to_the_maintainer(self): db = PromptDatabase(self.db_path) + handler = tool.ImageGenerationHandler(db, None) try: - handler = tool.ImageGenerationHandler(db, None) row = { "id": 1, "prompt_id": self.coffee, @@ -170,6 +171,7 @@ class ReviewToolTests(ToolFixture): ): verdict = tool.judge(db, handler, row, self.output) finally: + close_handler(self, handler) db.close_all() self.assertEqual(verdict["verdict"], "conflict") diff --git a/tests/test_review_inspection.py b/tests/test_review_inspection.py index 36cba58..33e4f3b 100644 --- a/tests/test_review_inspection.py +++ b/tests/test_review_inspection.py @@ -28,6 +28,7 @@ except ImportError: # discovered with tests/ as the top-level directory from test_api_review import ReviewRoutesTestCase from test_review import unlinkable_graph, write_png +from tests.background_stops import close_handler from utils import inspection_semantics, review_capture from utils.hashing import generate_prompt_hash from utils.image_monitor import ImageGenerationHandler @@ -437,6 +438,7 @@ class TestApplyInspectionCandidate(InspectionTestCase): class TestMonitorNeverQueues(InspectionTestCase): def _handler(self): handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db)) + self.addCleanup(close_handler, self, handler) handler.review_recorder = self.api.record_review_file return handler @@ -760,6 +762,7 @@ class TestRecordingEdges(InspectionTestCase): text="latest", prompt_hash=generate_prompt_hash("latest") ) handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db)) + self.addCleanup(close_handler, self, handler) handler.link_image_to_prompt( str(path), {"id": prompt_id, "text": "latest", "fallback": True}, diff --git a/tests/test_review_outcome.py b/tests/test_review_outcome.py index b24a524..e83e77b 100644 --- a/tests/test_review_outcome.py +++ b/tests/test_review_outcome.py @@ -38,6 +38,7 @@ except ImportError: # discovered with tests/ as the top-level directory write_png, ) +from tests.background_stops import close_handler from utils.hashing import generate_prompt_hash from utils.image_monitor import ImageGenerationHandler from utils.prompt_tracker import PromptTracker @@ -299,6 +300,7 @@ class TestQueueSelfHeals(OutcomeTestCase): class TestMonitorRule(OutcomeTestCase): def _handler(self): handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db)) + self.addCleanup(close_handler, self, handler) handler.review_recorder = self.api.record_review_file return handler diff --git a/tests/test_run_attribution.py b/tests/test_run_attribution.py index 5ec53c6..ce3c60c 100644 --- a/tests/test_run_attribution.py +++ b/tests/test_run_attribution.py @@ -11,6 +11,7 @@ from unittest import mock from database.run_attribution import stored_run_prompt from tests import test_execution_history_database as history_tests +from tests.background_stops import close_handler from tests.test_api_admin import AdminAPITestCase from tests.test_execution_context_database import snapshot from tests.test_prompt_attribution import issue_graph, llm_text_graph, make_tracker @@ -151,6 +152,7 @@ class TestStoredRunPrompt(RecordedRuns, unittest.TestCase): prompt_id = self.save(LLM_TEXT) self.record_run(self.db, self.output, [("sub", "img_00001_.png")], [LLM_TEXT]) handler = ImageGenerationHandler(self.db, make_tracker(self.db)) + self.addCleanup(close_handler, self, handler) linked = [] handler.link_image_to_prompt = lambda path, prompt, meta: linked.append( prompt["id"] @@ -332,7 +334,7 @@ class TestMonitorHistoryRecheck(RecordedRuns, unittest.TestCase): self.handler = ImageGenerationHandler(self.db, make_tracker(self.db)) self.handler.processing_delay = 0 self.handler.settle_interval = 0.01 - self.addCleanup(self.handler.shutdown) + self.addCleanup(close_handler, self, self.handler) self.linked = [] self.handler.link_image_to_prompt = lambda path, prompt, meta: ( self.linked.append((os.path.basename(path), prompt["id"])) @@ -650,7 +652,7 @@ class TestLinkedOutputRoot(RecordedRuns, unittest.TestCase): handler = ImageGenerationHandler(db, make_tracker(db)) handler.processing_delay = 0 handler.settle_interval = 0.01 - self.addCleanup(handler.shutdown) + self.addCleanup(close_handler, self, handler) linked = [] handler.link_image_to_prompt = lambda path, prompt, meta: linked.append( prompt["id"] diff --git a/tests/test_thumbnail_auto.py b/tests/test_thumbnail_auto.py index d21677e..20ed0a5 100644 --- a/tests/test_thumbnail_auto.py +++ b/tests/test_thumbnail_auto.py @@ -32,6 +32,7 @@ from PIL import Image # noqa: E402 import py.api.images as images_module # noqa: E402 import py.api.thumbnail_auto as auto # noqa: E402 +from tests.background_stops import close_handler, stop_monitor # noqa: E402 from py.api.jobs import THUMBNAILS_JOB_KIND # noqa: E402 SERVE = "/prompt_manager/images/serve/" @@ -980,6 +981,7 @@ class TestMonitorQueuesThumbnails(AutoThumbnailCase): from utils.image_monitor import ImageGenerationHandler handler = ImageGenerationHandler(self.api.db, unittest.mock.MagicMock()) + self.addCleanup(close_handler, self, handler) handler.processing_delay = 0 handler.settle_interval = 0 return handler @@ -1022,7 +1024,7 @@ class TestMonitorQueuesThumbnails(AutoThumbnailCase): monitor.set_thumbnail_queue(None) self.assertIsNone(monitor.handler.thumbnail_queue) finally: - monitor.stop_monitoring() + stop_monitor(self, monitor) class TestScanHandsOffThumbnails(AutoThumbnailCase): diff --git a/tests/test_usage_tracking.py b/tests/test_usage_tracking.py index 11c7282..f4a9354 100644 --- a/tests/test_usage_tracking.py +++ b/tests/test_usage_tracking.py @@ -16,6 +16,7 @@ import uuid sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from tests.background_stops import stop_tracker # noqa: E402 from database.operations import PromptDatabase from prompt_manager_base import PromptManagerBase from utils.hashing import generate_prompt_hash @@ -596,6 +597,7 @@ class TestNodeRoles(DbTestCase): class TestTrackerQueueFlag(unittest.TestCase): def setUp(self): self.tracker = PromptTracker(mock.Mock()) + self.addCleanup(stop_tracker, self, self.tracker) def test_push_to_queue_false_keeps_the_batch_queue_empty(self): self.tracker.set_current_prompt("a cat", {"prompt_id": 1}, push_to_queue=False) diff --git a/utils/image_monitor.py b/utils/image_monitor.py index 773af07..d887186 100644 --- a/utils/image_monitor.py +++ b/utils/image_monitor.py @@ -70,6 +70,11 @@ MAX_AWAITING_HISTORY = 1000 AWAIT_HISTORY_SECONDS = 3600 # ``schedule`` snapshot meaning "history-only re-check" (not a new image) HISTORY_ONLY = object() +# How long a stop waits for the image worker, and for the watchdog observer +# after its stop (#302). Every wait before the image in hand is interrupted +# at once, so a bound is reached only while one image is still being linked. +WORKER_JOIN_SECONDS = 5.0 +OBSERVER_JOIN_SECONDS = 5.0 class ImageGenerationHandler(FileSystemEventHandler): @@ -143,6 +148,11 @@ class ImageGenerationHandler(FileSystemEventHandler): self._history_due: "OrderedDict[str, None]" = OrderedDict() self._awaiting_lock = threading.Lock() self._sealed_subscribed = False + # Set by ``close``: no new work, no new worker, ever (#302). + self._closed = False + # Workers a stop signalled that had not ended yet: every later stop + # waits for them again, so it never reports True too early. + self._finishing: List[threading.Thread] = [] def on_created(self, event): """Handle filesystem creation events. @@ -157,6 +167,10 @@ class ImageGenerationHandler(FileSystemEventHandler): Args: event: FileSystemEvent object containing event details """ + # Closed: return before any log or tracker lock, so watchdog's + # ``observer.stop()`` never waits on this dispatch. + if self._closed: + return if event.is_directory or not self.is_image_file(event.src_path): return self.logger.info(f"New image detected: {event.src_path}") @@ -180,9 +194,12 @@ class ImageGenerationHandler(FileSystemEventHandler): and once its size has stopped changing. Returns: - False when the path is already pending or the pending set is full + False when the path is already pending, the pending set is full or + the handler is closed """ with self._pending_lock: + if self._closed: + return False if image_path in self._pending: self.logger.debug(f"Already pending, not rescheduled: {image_path}") return False @@ -203,12 +220,19 @@ class ImageGenerationHandler(FileSystemEventHandler): with self._pending_lock: return len(self._pending) - def shutdown(self) -> None: - """Stop the worker thread and drop every image still waiting. + def shutdown(self, timeout: float = WORKER_JOIN_SECONDS) -> bool: + """Stop the worker thread, drop every image still waiting, and wait + for the worker within ``timeout`` seconds. - The worker wakes from any delay or settle wait, finishes nothing more - and exits; the next ``schedule`` starts a fresh worker with its own - queue. Safe to call repeatedly and before anything was scheduled. + The worker wakes from any delay or settle wait, finishes at most the + image in hand and exits; the next ``schedule`` starts a fresh worker + with its own queue (use ``close`` for a final stop). Safe to call + repeatedly and before anything was scheduled. + + Returns: + True when no worker is left running. False when the worker is + still finishing its image after ``timeout`` (a warning is + logged), or when called on the worker itself (it only signals). """ with self._awaiting_lock: self._awaiting.clear() @@ -218,9 +242,40 @@ class ImageGenerationHandler(FileSystemEventHandler): worker, self._worker = self._worker, None work, self._work = self._work, queue.Queue() stop, self._stop = self._stop, threading.Event() + if worker is not None: + self._finishing.append(worker) + finishing = [w for w in self._finishing if w.is_alive()] + self._finishing = finishing stop.set() if worker is not None and worker.is_alive(): work.put(None) # sentinel: wake a worker idle on an empty queue + if threading.current_thread() in finishing: + self.logger.debug("shutdown called on the image worker itself") + return False + deadline = time.monotonic() + timeout + for thread in finishing: + thread.join(max(0.0, deadline - time.monotonic())) + with self._pending_lock: + self._finishing = [w for w in self._finishing if w.is_alive()] + still_running = bool(self._finishing) + if still_running: + self.logger.warning( + f"image worker still running after {timeout} s; " + "it stops after the image in hand" + ) + return False + return True + + def close(self, timeout: float = WORKER_JOIN_SECONDS) -> bool: + """Final stop: ``shutdown`` and refuse every later ``schedule``. + + After ``close`` no worker starts again, not even from a sealed run's + history re-check on the history writer's thread. Returns what + ``shutdown`` returns. + """ + with self._pending_lock: + self._closed = True + return self.shutdown(timeout) def _ensure_worker(self) -> None: """Start the single daemon worker thread if it isn't running. @@ -521,7 +576,12 @@ class ImageGenerationHandler(FileSystemEventHandler): self._awaiting.popitem(last=False) def _await_history(self, image_path: str) -> None: - """Re-check ``image_path`` once, history only, when its run is sealed.""" + """Re-check ``image_path`` once, history only, when its run is sealed. + + A closed handler re-checks nothing. + """ + if self._closed: + return if not self._sealed_subscribed: subscribe_outputs_sealed(self._on_outputs_sealed) self._sealed_subscribed = True @@ -570,10 +630,19 @@ class ImageGenerationHandler(FileSystemEventHandler): self._schedule_recheck(image_path) def _schedule_recheck(self, image_path: str) -> None: - """Schedule the history-only re-check, or keep it due for later.""" + """Schedule the history-only re-check, or keep it due for later. + + A closed handler drops it. + """ + if self._closed: + return if self.schedule(image_path, HISTORY_ONLY): return with self._awaiting_lock: + # ``close`` sets the flag before ``shutdown`` clears this set, so + # a close that raced the refusal above is seen here. + if self._closed: + return self._history_due.pop(image_path, None) self._history_due[image_path] = None while len(self._history_due) > MAX_AWAITING_HISTORY: @@ -872,6 +941,9 @@ class ImageMonitor: self.observer = None self.handler = None self.monitored_directories = [] + # (observer, handler) pairs a stop left running: the next stop waits + # for them again instead of reporting True. + self._unfinished: List[Tuple[Any, Any]] = [] self.logger = get_logger("prompt_manager.image_monitor") self.review_recorder: Optional[Callable[[str], str]] = None self.thumbnail_queue: Optional[Callable[[str], str]] = None @@ -999,29 +1071,69 @@ class ImageMonitor: """Whether a filesystem observer is running.""" return self.observer is not None - def stop_monitoring(self): - """Stop the image monitoring system. + def stop_monitoring(self, timeout: float = OBSERVER_JOIN_SECONDS) -> bool: + """Stop the image monitoring system within one ``timeout`` deadline. - Cleanly shuts down the filesystem watcher and clears all monitoring state. - This method should be called before program exit to ensure proper cleanup. - Safe to call when monitoring never started. + Closes the handler first (its worker is joined and its dispatch + returns at once), then stops the watchdog observer and joins it. + watchdog's ``observer.stop()`` joins its emitters without a limit; + it counts against the deadline but is not bounded here. A part an + earlier call left running is waited for again. Safe to call when + monitoring never started and more than once. + + Returns: + True when every worker and observer it stopped has ended. """ + deadline = time.monotonic() + timeout observer, self.observer = self.observer, None handler, self.handler = self.handler, None self.monitored_directories = [] + pairs, self._unfinished = self._unfinished, [] + if observer is not None: + pairs.append((observer, handler)) + if not pairs: + return True + + def remaining() -> float: + return max(0.0, deadline - time.monotonic()) + + for pair_observer, pair_handler in pairs: + handler_ok = self._close_handler(pair_handler, remaining()) + observer_ok = self._stop_observer(pair_observer, remaining()) + if not observer_ok: + self.logger.warning( + f"watchdog observer still running after {timeout} s" + ) + if not (handler_ok and observer_ok): + self._unfinished.append( + ( + None if observer_ok else pair_observer, + None if handler_ok else pair_handler, + ) + ) + self.logger.debug("Image monitoring stopped") + return not self._unfinished + + def _close_handler(self, handler, timeout: float) -> bool: + if handler is None: + return True + try: + return handler.close(timeout) + except Exception as e: + self.logger.warning(f"Error while stopping the image worker: {e}") + return False + + def _stop_observer(self, observer, timeout: float) -> bool: + """``observer.stop()`` (safe to repeat), then a bounded join.""" if observer is None: - return + return True try: observer.stop() - observer.join() + observer.join(timeout) except Exception as e: self.logger.warning(f"Error while stopping image monitoring: {e}") - if handler is not None: - try: - handler.shutdown() - except Exception as e: - self.logger.warning(f"Error while stopping the image worker: {e}") - self.logger.debug("Image monitoring stopped") + return False + return not observer.is_alive() def detect_comfyui_output_dirs(self) -> list: """Auto-detect ComfyUI output directories. diff --git a/utils/prompt_tracker.py b/utils/prompt_tracker.py index fcc1c21..67f7c79 100644 --- a/utils/prompt_tracker.py +++ b/utils/prompt_tracker.py @@ -44,6 +44,9 @@ except ImportError: from utils.hashing import generate_prompt_hash from utils.logging_config import get_logger +# How long ``PromptTracker.shutdown`` waits for the cleanup thread (#301). +TRACKER_JOIN_SECONDS = 2.0 + class PromptTracker: """Thread-safe tracking of current prompt executions. @@ -92,9 +95,12 @@ class PromptTracker: self.cleanup_interval = 300 # 5 minutes self.prompt_timeout = 600 # 10 minutes - # Start cleanup thread + # Start cleanup thread; ``shutdown`` sets ``_stop`` to end it. + self._stop = threading.Event() self.cleanup_thread = threading.Thread( - target=self._cleanup_expired_prompts, daemon=True + target=self._cleanup_expired_prompts, + name="PromptManagerPromptCleanup", + daemon=True, ) self.cleanup_thread.start() self.logger.debug("PromptTracker initialization completed") @@ -464,9 +470,10 @@ class PromptTracker: Runs continuously as a daemon thread, periodically removing expired prompt contexts from the global tracking dictionary. This prevents - memory leaks from accumulating old prompt data. + memory leaks from accumulating old prompt data. Every wait ends at + once when ``shutdown`` sets the stop event. """ - while True: + while not self._stop.is_set(): try: current_time = time.time() expired_ids = [] @@ -497,11 +504,37 @@ class PromptTracker: if pruned: self.logger.debug(f"Pruned {pruned} expired queue entries") - time.sleep(self.cleanup_interval) + if self._stop.wait(self.cleanup_interval): + break except Exception as e: self.logger.error(f"Error in cleanup thread: {e}") - time.sleep(60) # Wait a minute before retrying + if self._stop.wait(60): # Wait a minute before retrying + break + + def shutdown(self, timeout: float = TRACKER_JOIN_SECONDS) -> bool: + """Stop the cleanup thread and wait for it within ``timeout`` seconds. + + Returns True when the thread has ended (or never existed). The tracker + keeps working afterwards; only the background pruning of expired + entries stops. Safe to call more than once. Called on the cleanup + thread itself it only signals and returns False. + """ + stop = getattr(self, "_stop", None) + thread = getattr(self, "cleanup_thread", None) + if stop is None or thread is None: + return True # built without __init__: no thread was started + stop.set() + if thread is threading.current_thread(): + self.logger.debug("shutdown called on the prompt cleanup thread") + return False + thread.join(timeout) + if thread.is_alive(): + self.logger.warning( + f"prompt cleanup thread still running after {timeout} s" + ) + return False + return True def get_active_prompts(self) -> Dict[str, Dict[str, Any]]: """