diff --git a/py/lora_manager.py b/py/lora_manager.py index 727b4fc2..c745c16c 100644 --- a/py/lora_manager.py +++ b/py/lora_manager.py @@ -251,11 +251,14 @@ class LoraManager: # Startup sweep: purge pending-delete batches that expired during a # previous run. Non-blocking (fire-and-forget); purge_expired only # removes already-expired batches, so a staged undo that survived a - # restart stays restorable. Covers both plugin and standalone modes - # (StandaloneLoraManager reuses this classmethod). + # restart stays restorable. scan_roots=True runs the reconciliation + # pass first so leftover batches (the in-process registry is empty + # after a restart) are re-discovered on disk. Covers both plugin + # and standalone modes (StandaloneLoraManager reuses this + # classmethod). pending_delete_service = await get_pending_delete_service() asyncio.create_task( - pending_delete_service.purge_expired(), + pending_delete_service.purge_expired(scan_roots=True), name="pending_delete_startup_sweep", ) diff --git a/py/services/pending_delete_service.py b/py/services/pending_delete_service.py index 91fc620f..4267d480 100644 --- a/py/services/pending_delete_service.py +++ b/py/services/pending_delete_service.py @@ -97,6 +97,11 @@ class PendingDeleteService: # ServiceRegistry roots during sweeps so undo/purge work even before # every scanner is registered. self._known_roots: List[str] = [] + # MODEL batches only; recipe batches live in the fixed settings-dir + # parent. Registry access is short critical sections; the lock-free + # reconciliation scan registers concurrently. + self._known_batch_dirs: Dict[str, str] = {} + self._registry_lock = asyncio.Lock() # ------------------------------------------------------------------ # Public API @@ -176,6 +181,7 @@ class PendingDeleteService: ) self._write_manifest_atomic(batch_dir, manifest) self._remember_root(root) + await self._remember_batch(batch_id, batch_dir) # Arm the per-batch purge timer. Safe inside the lock: task # creation does not await, and purge_batch re-reads the # manifest's expires_at at fire time, so stale timers no-op. @@ -296,7 +302,7 @@ class PendingDeleteService: # Track (entry, original_staged_path, loser_dir) for rollback. moved: List[Tuple[Dict[str, Any], str, str]] = [] - processed_losers: List[str] = [] + processed_losers: List[Tuple[str, str]] = [] # (loser_id, loser_dir) try: for loser_id in batch_ids[1:]: @@ -332,7 +338,7 @@ class PendingDeleteService: entry["staged"] = os.path.abspath(new_staged) winner_manifest["entries"].append(entry) moved.append((entry, original_staged, loser_dir)) - processed_losers.append(loser_dir) + processed_losers.append((loser_id, loser_dir)) except OSError as exc: logger.warning( "Merge of %s failed after moving files: %s; rolling back", @@ -357,10 +363,15 @@ class PendingDeleteService: self._rollback_merge_moves(moved) return None - # All moves committed: remove loser dirs (must be empty by now). - for loser_dir in processed_losers: + # All moves committed: remove loser dirs (must be empty by now) + # and drop them from the registry. Skipped losers (missing / + # corrupted / same-dir) stay registered so the sweep still + # quarantines them, exactly as before the registry existed. + for loser_id, loser_dir in processed_losers: self._remove_manifest(loser_dir) self._remove_empty_dir(loser_dir) + await self._forget_batch(loser_id) + await self._remember_batch(winner_id, winner_dir) # Arm a fresh purge timer for the winner with the re-anchored # expiry (the winner's original timer fires at the OLD expiry and @@ -438,33 +449,101 @@ class PendingDeleteService: # Remove the manifest + batch dir only after all entries restored. self._remove_manifest(batch_dir) self._remove_empty_dir(batch_dir) + await self._forget_batch(batch_id) logger.info("Restored pending-delete batch %s", batch_id) return self._undo_result(manifest) - async def purge_expired(self) -> int: - """Purge every expired batch across ALL model roots and the recipe dir. + async def purge_expired(self, scan_roots: bool = False) -> int: + """Purge every expired batch. - Lock-free by design: enumerates staging parents (all scanner types via - the ServiceRegistry plus the global recipe staging dir) and delegates - each batch to :meth:`purge_batch`, which acquires the ops lock. Never - call this while holding the ops lock. + Default (registry-only): iterates a SNAPSHOT of the in-process MODEL + batch registry plus a shallow check of the fixed recipe staging + parent - cheap, no tree walk per delete. With ``scan_roots=True`` + (startup sweep only) a reconciliation pass re-discovers every batch + on disk under the model roots and registers it FIRST, so crash + leftovers and externally created batches are covered too. + + Lock-free by design: delegates each batch to :meth:`purge_batch`, + which acquires the ops lock. Never call this while holding the ops + lock. """ purged = 0 - for parent in await self._get_all_staging_parents(): - if not os.path.isdir(parent): + if scan_roots: + await self._reconcile_scan_roots() + # MODEL batches: snapshot so purge_batch can remove entries + # mid-iteration without a dict-changed-size error. + batch_ids: List[str] = [ + batch_id for batch_id, _dir in await self._registered_batch_dirs() + ] + # RECIPE batches: fixed settings-dir parent, shallow check as before. + recipe_parent = self._recipe_staging_parent() + for name in self._list_dir_names(recipe_parent): + if name.endswith(ORPHANED_SUFFIX): + # Quarantine is terminal - never re-rename or delete. continue - for name in self._list_dir_names(parent): - if name.endswith(ORPHANED_SUFFIX): - # Quarantine is terminal - never re-rename or delete. - continue - try: - await self.purge_batch(name) - purged += 1 - except Exception as exc: # defensive - sweep must not crash - logger.warning("Failed to purge batch %s: %s", name, exc) + if name not in batch_ids: + batch_ids.append(name) + for batch_id in batch_ids: + try: + await self.purge_batch(batch_id) + purged += 1 + except Exception as exc: # defensive - sweep must not crash + logger.warning("Failed to purge batch %s: %s", batch_id, exc) return purged + async def _reconcile_scan_roots(self) -> None: + """Register every pending-delete batch found under the model roots. + + Runs at startup (``purge_expired(scan_roots=True)``) to re-discover + batches left over from a previous process or created externally. + Registers ALL non-orphaned batch dirs regardless of manifest validity: + malformed/manifest-less dirs must reach ``_purge_batch_dir`` so it can + QUARANTINE them (preserving the pre-registry sweep semantics). The + walk only descends into dirs literally named ``.lm-pending-delete``, + so false positives are structurally limited. + """ + from .model_scanner import _is_excluded_dir + + for root in await self._get_all_model_roots(): + if not os.path.isdir(root): + continue + visited: Set[str] = set() + for dirpath, dirnames, _files in os.walk( + root, followlinks=True, topdown=True + ): + real_dir = os.path.realpath(dirpath) + if real_dir in visited: + # Symlink cycle: prune descent and move on. + dirnames[:] = [] + continue + visited.add(real_dir) + if os.path.basename(dirpath) == PENDING_DELETE_DIR_NAME: + # The current dir IS a staging parent (reachable only when + # a model root itself is one): register its batches. + await self._register_batch_candidates(dirpath) + dirnames[:] = [] + continue + next_dirs: List[str] = [] + for name in dirnames: + if name == PENDING_DELETE_DIR_NAME: + await self._register_batch_candidates( + os.path.join(dirpath, name) + ) + elif _is_excluded_dir(name): + continue + else: + next_dirs.append(name) + dirnames[:] = next_dirs + + async def _register_batch_candidates(self, staging_parent: str) -> None: + """Register every non-orphaned batch subdir of a staging parent.""" + for name in self._list_dir_names(staging_parent): + if name.endswith(ORPHANED_SUFFIX): + # Quarantine is terminal - never re-register. + continue + await self._remember_batch(name, os.path.join(staging_parent, name)) + async def purge_batch(self, batch_id: str) -> None: """Purge one batch. Silent no-op for missing/undone/not-yet-expired. @@ -476,7 +555,10 @@ class PendingDeleteService: batch_dir = await self._find_batch_dir(batch_id) if not batch_dir: return - self._purge_batch_dir(batch_dir) + if self._purge_batch_dir(batch_dir): + # Both purge and quarantine remove the batch dir (quarantine + # renames it to *.orphaned), so the registry entry is stale. + await self._forget_batch(batch_id) # ------------------------------------------------------------------ # Internals @@ -500,6 +582,29 @@ class PendingDeleteService: if root and root not in self._known_roots: self._known_roots.append(root) + async def _remember_batch(self, batch_id: str, batch_dir: str) -> None: + """Register a MODEL batch in the in-process registry (idempotent). + + Short critical section (dict mutation only, no I/O while holding the + lock) so the lock-free reconciliation scan can register concurrently. + """ + async with self._registry_lock: + self._known_batch_dirs[batch_id] = batch_dir + + async def _forget_batch(self, batch_id: str) -> None: + """Remove a MODEL batch from the in-process registry (idempotent).""" + async with self._registry_lock: + self._known_batch_dirs.pop(batch_id, None) + + async def _registered_batch_dirs(self) -> List[Tuple[str, str]]: + """Return a SNAPSHOT of (batch_id, batch_dir) registry pairs. + + The snapshot lets purge iterate safely while purge_batch removes + entries mid-loop (no dict-changed-size error). + """ + async with self._registry_lock: + return list(self._known_batch_dirs.items()) + def _find_model_root(self, scanner: Any, original_file_path: Optional[str]) -> Optional[str]: """Return the configured root containing ``original_file_path``.""" finder = getattr(scanner, "_find_root_for_file", None) @@ -818,18 +923,6 @@ class PendingDeleteService: settings_paths.get_settings_dir(create=True), PENDING_DELETE_DIR_NAME ) - async def _get_all_staging_parents(self) -> List[str]: - """Model staging parents for every scanner type + the recipe parent.""" - parents: List[str] = [] - for root in await self._get_all_model_roots(): - parent = os.path.join(root, PENDING_DELETE_DIR_NAME) - if parent not in parents: - parents.append(parent) - recipe_parent = self._recipe_staging_parent() - if recipe_parent not in parents: - parents.append(recipe_parent) - return parents - async def _get_all_model_roots(self) -> List[str]: """Collect every configured model root across all scanner types. @@ -878,13 +971,81 @@ class PendingDeleteService: return roots async def _find_batch_dir(self, batch_id: str) -> Optional[str]: - """Locate a batch directory across every staging parent.""" + """Locate a batch directory. + + Registry lookup first (fast path; stale entries are forgotten when + their dir vanished); then a targeted scan of the model roots for a + batch dir named exactly ``batch_id`` under a ``.lm-pending-delete`` + parent (restart / externally created batches; manifest verification + applies so random uuid-named user dirs are never registered); finally + the fixed recipe staging parent. Returns ``None`` (404 semantics) + when not found. + """ if not batch_id: return None - for parent in await self._get_all_staging_parents(): - candidate = os.path.join(parent, batch_id) - if os.path.isdir(candidate): + # 1) Registry fast path. + async with self._registry_lock: + known = self._known_batch_dirs.get(batch_id) + if known is not None: + if os.path.isdir(known): + return known + await self._forget_batch(batch_id) # stale entry - dir is gone + # 2) Targeted scan fallback across the model roots. + for root in await self._get_all_model_roots(): + if not os.path.isdir(root): + continue + candidate = await self._scan_root_for_batch(root, batch_id) + if candidate is not None: + await self._remember_batch(batch_id, candidate) return candidate + # 3) Recipe batches: fixed settings-dir parent, shallow check. + candidate = os.path.join(self._recipe_staging_parent(), batch_id) + if os.path.isdir(candidate): + return candidate + return None + + async def _scan_root_for_batch(self, root: str, batch_id: str) -> Optional[str]: + """Search one model root for a batch dir named exactly ``batch_id``. + + Walks the root (``followlinks=True``) with a realpath cycle guard, + looking for ``.lm-pending-delete`` parents whose subdir matches + ``batch_id`` AND has a parseable manifest. The manifest check prevents + random uuid-named user dirs from being treated as batches (a batch + with no parseable manifest cannot be undone anyway). + """ + from .model_scanner import _is_excluded_dir + + visited: Set[str] = set() + for dirpath, dirnames, _files in os.walk(root, followlinks=True, topdown=True): + real_dir = os.path.realpath(dirpath) + if real_dir in visited: + # Symlink cycle: prune descent and move on. + dirnames[:] = [] + continue + visited.add(real_dir) + if os.path.basename(dirpath) == PENDING_DELETE_DIR_NAME: + candidate = os.path.join(dirpath, batch_id) + if ( + os.path.isdir(candidate) + and self._read_manifest(candidate) is not None + ): + return candidate + dirnames[:] = [] + continue + next_dirs: List[str] = [] + for name in dirnames: + if name == PENDING_DELETE_DIR_NAME: + candidate = os.path.join(dirpath, name, batch_id) + if ( + os.path.isdir(candidate) + and self._read_manifest(candidate) is not None + ): + return candidate + continue + if _is_excluded_dir(name): + continue + next_dirs.append(name) + dirnames[:] = next_dirs return None def _list_dir_names(self, parent: str) -> List[str]: diff --git a/tests/services/test_pending_delete_service.py b/tests/services/test_pending_delete_service.py index 8cacca3d..f45d86ce 100644 --- a/tests/services/test_pending_delete_service.py +++ b/tests/services/test_pending_delete_service.py @@ -451,7 +451,9 @@ async def test_g_manifestless_dir_quarantined(tmp_path: Path, monkeypatch) -> No await _register_model_root(monkeypatch, lora_roots=[root]) service = await PendingDeleteService.get_instance() - await service.purge_expired() + # The batch is hand-created (unregistered): the reconciliation pass is + # required for the default registry-only purge to discover it. + await service.purge_expired(scan_roots=True) orphaned = staging / "batch1.orphaned" assert orphaned.is_dir() @@ -474,7 +476,8 @@ async def test_h_corrupted_manifest_quarantined(tmp_path: Path, monkeypatch) -> await _register_model_root(monkeypatch, lora_roots=[root]) service = await PendingDeleteService.get_instance() - await service.purge_expired() # must not crash + # Hand-created (unregistered) batch: reconciliation discovers it. + await service.purge_expired(scan_roots=True) # must not crash orphaned = staging / "batch2.orphaned" assert orphaned.is_dir() @@ -1055,7 +1058,8 @@ async def test_r_purge_expired_enumerates_all_scanner_types_and_recipe_dir( ) service = await PendingDeleteService.get_instance() - purged = await service.purge_expired() + # Hand-created (unregistered) batches: reconciliation pass discovers them. + purged = await service.purge_expired(scan_roots=True) assert purged >= 4 for root in (lora_root, ckpt_root, emb_root): @@ -1117,12 +1121,13 @@ async def test_t_quarantine_is_terminal(tmp_path: Path, monkeypatch) -> None: await _register_model_root(monkeypatch, lora_roots=[root]) service = await PendingDeleteService.get_instance() - await service.purge_expired() + # Hand-created (unregistered) batch: reconciliation discovers it. + await service.purge_expired(scan_roots=True) orphaned = staging / "qbatch.orphaned" assert orphaned.is_dir() # Second sweep must NOT re-rename or delete the quarantined dir. - await service.purge_expired() + await service.purge_expired(scan_roots=True) assert orphaned.is_dir() assert (orphaned / "model.safetensors").read_bytes() == b"data" assert not batch_dir.exists() @@ -1169,7 +1174,9 @@ async def test_u_lock_no_deadlock_with_concurrent_purge(tmp_path: Path, monkeypa cached_entry=None, ) - purge_task = asyncio.create_task(service.purge_expired()) + # Hand-created (unregistered) "expired" batch: the purge task must run the + # reconciliation pass to discover it alongside the staged "new" batch. + purge_task = asyncio.create_task(service.purge_expired(scan_roots=True)) stage_task = asyncio.create_task(do_stage()) results = await asyncio.gather(purge_task, stage_task, return_exceptions=True) @@ -1533,3 +1540,327 @@ async def test_snap2_merge_keeps_both_snapshots( snap_entries = [e for e in manifest["entries"] if e.get("snapshot")] assert len(snap_entries) == 2 assert {e["snapshot"]["file_path"] for e in snap_entries} == {str(a1), str(b1)} + + +# --------------------------------------------------------------------------- +# Batch-registry lifecycle (todo 1: in-process _known_batch_dirs) +# --------------------------------------------------------------------------- + +# (a) stage_model_delete registers in _known_batch_dirs +async def test_reg_a_stage_model_registers_batch(tmp_path: Path) -> None: + root = tmp_path / "loras" + root.mkdir() + service = await PendingDeleteService.get_instance() + batch_id = await _stage_simple(service, root, "model") + + assert service._known_batch_dirs.get(batch_id) == str( + root / PENDING_DELETE_DIR_NAME / batch_id + ) + + +# (b) undo success removes the entry +async def test_reg_b_undo_success_removes_registry_entry(tmp_path: Path) -> None: + root = tmp_path / "loras" + root.mkdir() + service = await PendingDeleteService.get_instance() + batch_id = await _stage_simple(service, root, "model") + assert batch_id in service._known_batch_dirs + + await service.undo(batch_id) + + assert batch_id not in service._known_batch_dirs + + +# (c) purge_batch removes the entry after a real purge +async def test_reg_c_purge_batch_removes_registry_entry(tmp_path: Path) -> None: + root = tmp_path / "loras" + root.mkdir() + service = await PendingDeleteService.get_instance() + batch_id = await _stage_simple(service, root, "model") + batch_dir = root / PENDING_DELETE_DIR_NAME / batch_id + manifest_path = batch_dir / "manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["expires_at"] = int(time.time()) - 10 + manifest_path.write_text(json.dumps(manifest)) + + await service.purge_batch(batch_id) + + assert batch_id not in service._known_batch_dirs + assert not batch_dir.exists() + + +# (d) quarantine (corrupted manifest) removes the entry +async def test_reg_d_quarantine_removes_registry_entry(tmp_path: Path) -> None: + root = tmp_path / "loras" + root.mkdir() + service = await PendingDeleteService.get_instance() + batch_id = await _stage_simple(service, root, "model") + batch_dir = root / PENDING_DELETE_DIR_NAME / batch_id + assert batch_id in service._known_batch_dirs + + # Corrupt the manifest: purge_batch quarantines the dir (returns True). + (batch_dir / "manifest.json").write_text("{ not valid json !!!") + + await service.purge_batch(batch_id) + + assert batch_id not in service._known_batch_dirs + assert not batch_dir.exists() + assert (batch_dir.with_name(f"{batch_id}.orphaned")).is_dir() + + +# (e) merge success: winner present + losers removed; EXDEV-abort: unchanged +async def test_reg_e_merge_registry_lifecycle( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "loras" + root.mkdir() + _spy_purge_timers(monkeypatch) + service = await PendingDeleteService.get_instance() + bid_a = await _stage_simple(service, root, "alpha") + bid_b = await _stage_simple(service, root, "beta") + bid_c = await _stage_simple(service, root, "gamma") + assert set(service._known_batch_dirs) == {bid_a, bid_b, bid_c} + + # Merge success: winner stays, processed loser forgotten, untouched batch stays. + assert await service.merge_batches([bid_a, bid_b]) == bid_a + assert bid_a in service._known_batch_dirs + assert bid_b not in service._known_batch_dirs + assert bid_c in service._known_batch_dirs + + # EXDEV-abort: registry untouched. + def exdev_rename(src: str, dst: str) -> None: + raise OSError(errno.EXDEV, "Invalid cross-device link", src, dst) + + monkeypatch.setattr("py.services.pending_delete_service.os.rename", exdev_rename) + before = dict(service._known_batch_dirs) + assert await service.merge_batches([bid_a, bid_c]) is None + assert dict(service._known_batch_dirs) == before + + +# (f) _reset_pending_delete_service clears the registry +async def test_reg_f_reset_clears_registry(tmp_path: Path) -> None: + root = tmp_path / "loras" + root.mkdir() + service = await PendingDeleteService.get_instance() + batch_id = await _stage_simple(service, root, "model") + assert service._known_batch_dirs + + _reset_pending_delete_service() + + fresh = await PendingDeleteService.get_instance() + assert fresh is not service + assert fresh._known_batch_dirs == {} + + +# (g) scan_roots=True reconciles externally created batches (expired purged, +# non-expired registered); the registry-only default does NOT find them +async def test_reg_g_reconciliation_finds_external_batches( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "loras" + root.mkdir() + await _register_model_root(monkeypatch, lora_roots=[root]) + + expired_dir = root / PENDING_DELETE_DIR_NAME / "ext-expired" + expired_dir.mkdir(parents=True) + (expired_dir / "old.safetensors").write_bytes(b"old") + _write_batch_manifest( + expired_dir, + batch_id="ext-expired", + kind="model", + model_type="loras", + expires_at=int(time.time()) - 10, + entries=[ + { + "staged": str(expired_dir / "old.safetensors"), + "original": str(root / "old.safetensors"), + "restored": False, + } + ], + ) + fresh_dir = root / PENDING_DELETE_DIR_NAME / "ext-fresh" + fresh_dir.mkdir(parents=True) + (fresh_dir / "new.safetensors").write_bytes(b"new") + _write_batch_manifest( + fresh_dir, + batch_id="ext-fresh", + kind="model", + model_type="loras", + expires_at=int(time.time()) + 100, + entries=[ + { + "staged": str(fresh_dir / "new.safetensors"), + "original": str(root / "new.safetensors"), + "restored": False, + } + ], + ) + + service = await PendingDeleteService.get_instance() + + # Registry-only default: the externally created batches are invisible. + await service.purge_expired() + assert expired_dir.is_dir() + assert fresh_dir.is_dir() + assert "ext-expired" not in service._known_batch_dirs + assert "ext-fresh" not in service._known_batch_dirs + + # Reconciliation pass: expired one purged, non-expired one registered. + await service.purge_expired(scan_roots=True) + + assert not expired_dir.exists() + assert not (root / "old.safetensors").exists() + assert fresh_dir.is_dir() + assert (fresh_dir / "new.safetensors").exists() + assert "ext-expired" not in service._known_batch_dirs + assert service._known_batch_dirs.get("ext-fresh") == str(fresh_dir) + + +# (h) _find_batch_dir with cleared registry locates + registers (restart sim) +async def test_reg_h_find_batch_dir_restart_simulation( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "loras" + root.mkdir() + await _register_model_root(monkeypatch, lora_roots=[root]) + service = await PendingDeleteService.get_instance() + batch_id = await _stage_simple(service, root, "model") + assert batch_id in service._known_batch_dirs + + # Simulate a restart: the in-process registry is empty but the batch dir + # is still on disk. + service._known_batch_dirs.clear() + + found = await service._find_batch_dir(batch_id) + + assert found == str(root / PENDING_DELETE_DIR_NAME / batch_id) + assert service._known_batch_dirs.get(batch_id) == found + + # Undo works after the restart simulation. + await service.undo(batch_id) + assert (root / "model.safetensors").read_bytes() == b"model-data" + assert batch_id not in service._known_batch_dirs + + +# (i) purge iteration uses a snapshot: no dict-changed-size when entries are +# removed mid-iteration +async def test_reg_i_purge_iteration_uses_snapshot( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "loras" + root.mkdir() + await _register_model_root(monkeypatch, lora_roots=[root]) + service = await PendingDeleteService.get_instance() + ids = [await _stage_simple(service, root, f"m{i}") for i in range(5)] + + for batch_id in ids: + manifest_path = root / PENDING_DELETE_DIR_NAME / batch_id / "manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["expires_at"] = int(time.time()) - 10 + manifest_path.write_text(json.dumps(manifest)) + + # Every purge removes its registry entry mid-loop; the snapshot makes this + # safe (iterating the dict directly would raise RuntimeError). + await service.purge_expired() + + assert service._known_batch_dirs == {} + for batch_id in ids: + assert not (root / PENDING_DELETE_DIR_NAME / batch_id).exists() + + +# (j) STARTUP SWEEP PIN: the startup sweep task passes scan_roots=True +async def test_reg_j_startup_sweep_passes_scan_roots_true( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from py import lora_manager + + sweep_calls: List[Dict[str, Any]] = [] + + class _SpySweepService: + async def purge_expired(self, scan_roots: bool = False) -> int: + sweep_calls.append({"scan_roots": scan_roots}) + return 0 + + async def _fake_get_service() -> _SpySweepService: + return _SpySweepService() + + monkeypatch.setattr(lora_manager, "get_pending_delete_service", _fake_get_service) + + async def _stub(*args: Any, **_kwargs: Any) -> Any: + return args[0] if args else None + + class _DummyScanner: + async def initialize_in_background(self) -> None: + return None + + dummy = _DummyScanner() + monkeypatch.setattr(lora_manager.ServiceRegistry, "get_civitai_client", lambda: _stub()) + monkeypatch.setattr(lora_manager.ServiceRegistry, "get_download_manager", lambda: _stub()) + monkeypatch.setattr( + lora_manager.ServiceRegistry, "get_download_queue_service", lambda: _stub() + ) + monkeypatch.setattr(lora_manager.ServiceRegistry, "get_backup_service", lambda: _stub()) + monkeypatch.setattr(lora_manager.ServiceRegistry, "get_websocket_manager", lambda: _stub()) + monkeypatch.setattr(lora_manager.ServiceRegistry, "get_lora_scanner", lambda: _stub(dummy)) + monkeypatch.setattr( + lora_manager.ServiceRegistry, "get_checkpoint_scanner", lambda: _stub(dummy) + ) + monkeypatch.setattr( + lora_manager.ServiceRegistry, "get_embedding_scanner", lambda: _stub(dummy) + ) + monkeypatch.setattr(lora_manager.ServiceRegistry, "get_recipe_scanner", lambda: _stub(dummy)) + + from py.services import metadata_service as metadata_service_module + + monkeypatch.setattr( + metadata_service_module, + "initialize_metadata_providers", + _stub, + ) + + from py.services.llm_service import LLMService + + monkeypatch.setattr(LLMService, "get_instance", _stub) + + async def _fake_migration() -> None: + return None + + monkeypatch.setattr( + lora_manager.ExampleImagesMigration, + "check_and_run_migrations", + staticmethod(_fake_migration), + ) + + captured: List[Any] = [] + + class _DummyTask: + def add_done_callback(self, _cb: Any) -> None: # pragma: no cover - stub + pass + + def done(self) -> bool: # pragma: no cover - stub + return False + + def _capture_task(coro: Any, *args: Any, **kwargs: Any) -> _DummyTask: + captured.append(coro) + return _DummyTask() + + monkeypatch.setattr(asyncio, "create_task", _capture_task) + + try: + await lora_manager.LoraManager._initialize_services() + finally: + sweep_coro: Any = None + for coro in captured: + qualname = getattr(coro.cr_code, "co_qualname", "") + if "_SpySweepService.purge_expired" in qualname: + sweep_coro = coro + else: + coro.close() + if sweep_coro is not None: + # The sweep task body only runs when awaited; execute just the + # spy's purge_expired so it records its invocation arguments. + await sweep_coro + + # The startup sweep must invoke purge_expired with scan_roots=True (the + # reconciliation flag) - forgetting it would break restart cleanup. + assert sweep_calls == [{"scan_roots": True}]