Guard against None base_dir in _collect_output_absolute_paths

Return early with an empty list when folder_paths.get_directory_by_type('output')
returns None to avoid os.path.join(None, ...) producing invalid paths.

Amp-Thread-ID: https://ampcode.com/threads/T-019cfbe5-dffc-760a-9a37-9b041dd71e73
Co-authored-by: Amp <amp@ampcode.com>

app/assets/scanner.py

@@ -338,6 +338,7 @@ def build_asset_specs(
                 "metadata": metadata,
                 "hash": asset_hash,
                 "mime_type": mime_type,
+                "job_id": None,
             }
         )
         tag_pool.update(tags)

app/assets/seeder.py

@@ -92,6 +92,7 @@ class _AssetSeeder:
         self._prune_first: bool = False
         self._progress_callback: ProgressCallback | None = None
         self._disabled: bool = False
+        self._pending_enrich: dict | None = None
 
     def disable(self) -> None:
         """Disable the asset seeder, preventing any scans from starting."""
@@ -196,6 +197,42 @@ class _AssetSeeder:
             compute_hashes=compute_hashes,
         )
 
+    def enqueue_enrich(
+        self,
+        roots: tuple[RootType, ...] = ("models", "input", "output"),
+        compute_hashes: bool = False,
+    ) -> bool:
+        """Start an enrichment scan now, or queue it for after the current scan.
+
+        If the seeder is idle, starts immediately. Otherwise, the enrich
+        request is stored and will run automatically when the current scan
+        finishes.
+
+        Args:
+            roots: Tuple of root types to scan
+            compute_hashes: If True, compute blake3 hashes
+
+        Returns:
+            True if started immediately, False if queued for later
+        """
+        if self.start_enrich(roots=roots, compute_hashes=compute_hashes):
+            return True
+        with self._lock:
+            if self._pending_enrich is not None:
+                existing_roots = set(self._pending_enrich["roots"])
+                existing_roots.update(roots)
+                self._pending_enrich["roots"] = tuple(existing_roots)
+                self._pending_enrich["compute_hashes"] = (
+                    self._pending_enrich["compute_hashes"] or compute_hashes
+                )
+            else:
+                self._pending_enrich = {
+                    "roots": roots,
+                    "compute_hashes": compute_hashes,
+                }
+            logging.info("Enrich scan queued (roots=%s)", self._pending_enrich["roots"])
+        return False
+
     def cancel(self) -> bool:
         """Request cancellation of the current scan.
 
@@ -381,9 +418,13 @@ class _AssetSeeder:
             return marked
         finally:
             with self._lock:
-                self._last_progress = self._progress
-                self._state = State.IDLE
-                self._progress = None
+                self._reset_to_idle()
+
+    def _reset_to_idle(self) -> None:
+        """Reset state to IDLE, preserving last progress. Caller must hold _lock."""
+        self._last_progress = self._progress
+        self._state = State.IDLE
+        self._progress = None
 
     def _is_cancelled(self) -> bool:
         """Check if cancellation has been requested."""
@@ -594,9 +635,14 @@ class _AssetSeeder:
                     },
                 )
             with self._lock:
-                self._last_progress = self._progress
-                self._state = State.IDLE
-                self._progress = None
+                self._reset_to_idle()
+                pending = self._pending_enrich
+                self._pending_enrich = None
+            if pending is not None:
+                self.start_enrich(
+                    roots=pending["roots"],
+                    compute_hashes=pending["compute_hashes"],
+                )
 
     def _run_fast_phase(self, roots: tuple[RootType, ...]) -> tuple[int, int, int]:
         """Run phase 1: fast scan to create stub records.

app/assets/services/__init__.py

@@ -23,6 +23,8 @@ from app.assets.services.ingest import (
     DependencyMissingError,
     HashMismatchError,
     create_from_hash,
+    ingest_existing_file,
+    register_output_files,
     upload_from_temp_path,
 )
 from app.assets.database.queries import (
@@ -72,6 +74,8 @@ __all__ = [
     "delete_asset_reference",
     "get_asset_by_hash",
     "get_asset_detail",
+    "ingest_existing_file",
+    "register_output_files",
     "get_mtime_ns",
     "get_size_and_mtime_ns",
     "list_assets_page",

app/assets/services/bulk_ingest.py

@@ -34,9 +34,10 @@ class SeedAssetSpec(TypedDict):
     info_name: str
     tags: list[str]
     fname: str
-    metadata: ExtractedMetadata | None
+    metadata: ExtractedMetadata | dict[str, Any] | None
     hash: str | None
     mime_type: str | None
+    job_id: str | None
 
 
 class AssetRow(TypedDict):
@@ -60,6 +61,7 @@ class ReferenceRow(TypedDict):
     name: str
     preview_id: str | None
     user_metadata: dict[str, Any] | None
+    job_id: str | None
     created_at: datetime
     updated_at: datetime
     last_access_time: datetime
@@ -150,8 +152,10 @@ def batch_insert_seed_assets(
 
         # Build user_metadata from extracted metadata or fallback to filename
         extracted_metadata = spec.get("metadata")
-        if extracted_metadata:
-            user_metadata: dict[str, Any] | None = extracted_metadata.to_user_metadata()
+        if isinstance(extracted_metadata, dict):
+            user_metadata: dict[str, Any] | None = extracted_metadata
+        elif extracted_metadata:
+            user_metadata = extracted_metadata.to_user_metadata()
         elif spec["fname"]:
             user_metadata = {"filename": spec["fname"]}
         else:
@@ -167,6 +171,7 @@ def batch_insert_seed_assets(
                 "name": spec["info_name"],
                 "preview_id": None,
                 "user_metadata": user_metadata,
+                "job_id": spec.get("job_id"),
                 "created_at": current_time,
                 "updated_at": current_time,
                 "last_access_time": current_time,

app/assets/services/ingest.py

@@ -24,6 +24,7 @@ from app.assets.database.queries import (
     validate_tags_exist,
 )
 from app.assets.helpers import normalize_tags
+from app.assets.services.bulk_ingest import batch_insert_seed_assets
 from app.assets.services.file_utils import get_size_and_mtime_ns
 from app.assets.services.path_utils import (
     compute_relative_filename,
@@ -130,6 +131,62 @@ def _ingest_file_from_path(
     )
 
 
+def register_output_files(
+    file_paths: Sequence[str],
+    user_metadata: UserMetadata = None,
+    job_id: str | None = None,
+) -> int:
+    """Register a batch of output file paths as assets.
+
+    Returns the number of files successfully registered.
+    """
+    registered = 0
+    for abs_path in file_paths:
+        if not os.path.isfile(abs_path):
+            continue
+        try:
+            ingest_existing_file(abs_path, user_metadata=user_metadata, job_id=job_id)
+            registered += 1
+        except Exception:
+            logging.exception("Failed to register output: %s", abs_path)
+    return registered
+
+
+def ingest_existing_file(
+    abs_path: str,
+    user_metadata: UserMetadata = None,
+    extra_tags: Sequence[str] = (),
+    owner_id: str = "",
+    job_id: str | None = None,
+) -> None:
+    """Register an existing on-disk file as an asset stub.
+
+    Inserts a stub record (hash=NULL) for immediate UX visibility.
+    The caller is responsible for triggering background enrichment
+    (hash computation, metadata extraction) via the asset seeder.
+    """
+    size_bytes, mtime_ns = get_size_and_mtime_ns(abs_path)
+    mime_type = mimetypes.guess_type(abs_path, strict=False)[0]
+    name, path_tags = get_name_and_tags_from_asset_path(abs_path)
+    tags = list(dict.fromkeys(path_tags + list(extra_tags)))
+
+    spec = {
+        "abs_path": abs_path,
+        "size_bytes": size_bytes,
+        "mtime_ns": mtime_ns,
+        "info_name": name,
+        "tags": tags,
+        "fname": os.path.basename(abs_path),
+        "metadata": user_metadata,
+        "hash": None,
+        "mime_type": mime_type,
+        "job_id": job_id,
+    }
+    with create_session() as session:
+        batch_insert_seed_assets(session, [spec], owner_id=owner_id)
+        session.commit()
+
+
 def _register_existing_asset(
     asset_hash: str,
     name: str,

app/user_manager.py

@@ -6,6 +6,7 @@ import uuid
 import glob
 import shutil
 import logging
+import tempfile
 from aiohttp import web
 from urllib import parse
 from comfy.cli_args import args
@@ -377,8 +378,15 @@ class UserManager():
             try:
                 body = await request.read()
 
-                with open(path, "wb") as f:
-                    f.write(body)
+                dir_name = os.path.dirname(path)
+                fd, tmp_path = tempfile.mkstemp(dir=dir_name)
+                try:
+                    with os.fdopen(fd, "wb") as f:
+                        f.write(body)
+                    os.replace(tmp_path, path)
+                except:
+                    os.unlink(tmp_path)
+                    raise
             except OSError as e:
                 logging.warning(f"Error saving file '{path}': {e}")
                 return web.Response(

comfy/ldm/hunyuan3dv2_1/hunyuandit.py

@@ -343,6 +343,7 @@ class CrossAttention(nn.Module):
             k.reshape(b, s2, self.num_heads * self.head_dim),
             v,
             heads=self.num_heads,
+            low_precision_attention=False,
         )
 
         out = self.out_proj(x)
@@ -412,6 +413,7 @@ class Attention(nn.Module):
             key.reshape(B, N, self.num_heads * self.head_dim),
             value,
             heads=self.num_heads,
+            low_precision_attention=False,
         )
 
         x = self.out_proj(x)

comfy/ops.py

@@ -776,6 +776,71 @@ from .quant_ops import (
 )
 
 
+class QuantLinearFunc(torch.autograd.Function):
+    """Custom autograd function for quantized linear: quantized forward, compute_dtype backward.
+    Handles any input rank by flattening to 2D for matmul and restoring shape after.
+    """
+
+    @staticmethod
+    def forward(ctx, input_float, weight, bias, layout_type, input_scale, compute_dtype):
+        input_shape = input_float.shape
+        inp = input_float.detach().flatten(0, -2)  # zero-cost view to 2D
+
+        # Quantize input (same as inference path)
+        if layout_type is not None:
+            q_input = QuantizedTensor.from_float(inp, layout_type, scale=input_scale)
+        else:
+            q_input = inp
+
+        w = weight.detach() if weight.requires_grad else weight
+        b = bias.detach() if bias is not None and bias.requires_grad else bias
+
+        output = torch.nn.functional.linear(q_input, w, b)
+
+        # Restore original input shape
+        if len(input_shape) > 2:
+            output = output.unflatten(0, input_shape[:-1])
+
+        ctx.save_for_backward(input_float, weight)
+        ctx.input_shape = input_shape
+        ctx.has_bias = bias is not None
+        ctx.compute_dtype = compute_dtype
+        ctx.weight_requires_grad = weight.requires_grad
+
+        return output
+
+    @staticmethod
+    @torch.autograd.function.once_differentiable
+    def backward(ctx, grad_output):
+        input_float, weight = ctx.saved_tensors
+        compute_dtype = ctx.compute_dtype
+        grad_2d = grad_output.flatten(0, -2).to(compute_dtype)
+
+        # Dequantize weight to compute dtype for backward matmul
+        if isinstance(weight, QuantizedTensor):
+            weight_f = weight.dequantize().to(compute_dtype)
+        else:
+            weight_f = weight.to(compute_dtype)
+
+        # grad_input = grad_output @ weight
+        grad_input = torch.mm(grad_2d, weight_f)
+        if len(ctx.input_shape) > 2:
+            grad_input = grad_input.unflatten(0, ctx.input_shape[:-1])
+
+        # grad_weight (only if weight requires grad, typically frozen for quantized training)
+        grad_weight = None
+        if ctx.weight_requires_grad:
+            input_f = input_float.flatten(0, -2).to(compute_dtype)
+            grad_weight = torch.mm(grad_2d.t(), input_f)
+
+        # grad_bias
+        grad_bias = None
+        if ctx.has_bias:
+            grad_bias = grad_2d.sum(dim=0)
+
+        return grad_input, grad_weight, grad_bias, None, None, None
+
+
 def mixed_precision_ops(quant_config={}, compute_dtype=torch.bfloat16, full_precision_mm=False, disabled=[]):
     class MixedPrecisionOps(manual_cast):
         _quant_config = quant_config
@@ -970,10 +1035,37 @@ def mixed_precision_ops(quant_config={}, compute_dtype=torch.bfloat16, full_prec
                 #If cast needs to apply lora, it should be done in the compute dtype
                 compute_dtype = input.dtype
 
-                if (getattr(self, 'layout_type', None) is not None and
+                _use_quantized = (
+                    getattr(self, 'layout_type', None) is not None and
                     not isinstance(input, QuantizedTensor) and not self._full_precision_mm and
                     not getattr(self, 'comfy_force_cast_weights', False) and
-                    len(self.weight_function) == 0 and len(self.bias_function) == 0):
+                    len(self.weight_function) == 0 and len(self.bias_function) == 0
+                )
+
+                # Training path: quantized forward with compute_dtype backward via autograd function
+                if (input.requires_grad and _use_quantized):
+
+                    weight, bias, offload_stream = cast_bias_weight(
+                        self,
+                        input,
+                        offloadable=True,
+                        compute_dtype=compute_dtype,
+                        want_requant=True
+                    )
+
+                    scale = getattr(self, 'input_scale', None)
+                    if scale is not None:
+                        scale = comfy.model_management.cast_to_device(scale, input.device, None)
+
+                    output = QuantLinearFunc.apply(
+                        input, weight, bias, self.layout_type, scale, compute_dtype
+                    )
+
+                    uncast_bias_weight(self, weight, bias, offload_stream)
+                    return output
+
+                # Inference path (unchanged)
+                if _use_quantized:
 
                     # Reshape 3D tensors to 2D for quantization (needed for NVFP4 and others)
                     input_reshaped = input.reshape(-1, input_shape[2]) if input.ndim == 3 else input
@@ -1021,7 +1113,10 @@ def mixed_precision_ops(quant_config={}, compute_dtype=torch.bfloat16, full_prec
                 for key, param in self._parameters.items():
                     if param is None:
                         continue
-                    self.register_parameter(key, torch.nn.Parameter(fn(param), requires_grad=False))
+                    p = fn(param)
+                    if p.is_inference():
+                        p = p.clone()
+                    self.register_parameter(key, torch.nn.Parameter(p, requires_grad=False))
                 for key, buf in self._buffers.items():
                     if buf is not None:
                         self._buffers[key] = fn(buf)

comfy/utils.py

@@ -897,6 +897,10 @@ def set_attr(obj, attr, value):
     return prev
 
 def set_attr_param(obj, attr, value):
+    # Clone inference tensors (created under torch.inference_mode) since
+    # their version counter is frozen and nn.Parameter() cannot wrap them.
+    if (not torch.is_inference_mode_enabled()) and value.is_inference():
+        value = value.clone()
     return set_attr(obj, attr, torch.nn.Parameter(value, requires_grad=False))
 
 def set_attr_buffer(obj, attr, value):

comfy_extras/nodes_train.py

@@ -15,6 +15,7 @@ import comfy.sampler_helpers
 import comfy.sd
 import comfy.utils
 import comfy.model_management
+from comfy.cli_args import args, PerformanceFeature
 import comfy_extras.nodes_custom_sampler
 import folder_paths
 import node_helpers
@@ -138,6 +139,7 @@ class TrainSampler(comfy.samplers.Sampler):
         training_dtype=torch.bfloat16,
         real_dataset=None,
         bucket_latents=None,
+        use_grad_scaler=False,
     ):
         self.loss_fn = loss_fn
         self.optimizer = optimizer
@@ -152,6 +154,8 @@ class TrainSampler(comfy.samplers.Sampler):
         self.bucket_latents: list[torch.Tensor] | None = (
             bucket_latents  # list of (Bi, C, Hi, Wi)
         )
+        # GradScaler for fp16 training
+        self.grad_scaler = torch.amp.GradScaler() if use_grad_scaler else None
         # Precompute bucket offsets and weights for sampling
         if bucket_latents is not None:
             self._init_bucket_data(bucket_latents)
@@ -204,10 +208,13 @@ class TrainSampler(comfy.samplers.Sampler):
                 batch_sigmas.requires_grad_(True),
                 **batch_extra_args,
             )
-            loss = self.loss_fn(x0_pred, x0)
+            loss = self.loss_fn(x0_pred.float(), x0.float())
         if bwd:
             bwd_loss = loss / self.grad_acc
-            bwd_loss.backward()
+            if self.grad_scaler is not None:
+                self.grad_scaler.scale(bwd_loss).backward()
+            else:
+                bwd_loss.backward()
         return loss
 
     def _generate_batch_sigmas(self, model_wrap, batch_size, device):
@@ -307,7 +314,10 @@ class TrainSampler(comfy.samplers.Sampler):
             )
             total_loss += loss
         total_loss = total_loss / self.grad_acc / len(indicies)
-        total_loss.backward()
+        if self.grad_scaler is not None:
+            self.grad_scaler.scale(total_loss).backward()
+        else:
+            total_loss.backward()
         if self.loss_callback:
             self.loss_callback(total_loss.item())
         pbar.set_postfix({"loss": f"{total_loss.item():.4f}"})
@@ -348,12 +358,18 @@ class TrainSampler(comfy.samplers.Sampler):
                 self._train_step_multires_mode(model_wrap, cond, extra_args, noisegen, latent_image, dataset_size, pbar)
 
             if (i + 1) % self.grad_acc == 0:
+                if self.grad_scaler is not None:
+                    self.grad_scaler.unscale_(self.optimizer)
                 for param_groups in self.optimizer.param_groups:
                     for param in param_groups["params"]:
                         if param.grad is None:
                             continue
                         param.grad.data = param.grad.data.to(param.data.dtype)
-                self.optimizer.step()
+                if self.grad_scaler is not None:
+                    self.grad_scaler.step(self.optimizer)
+                    self.grad_scaler.update()
+                else:
+                    self.optimizer.step()
                 self.optimizer.zero_grad()
             ui_pbar.update(1)
         torch.cuda.empty_cache()
@@ -1004,9 +1020,9 @@ class TrainLoraNode(io.ComfyNode):
                 ),
                 io.Combo.Input(
                     "training_dtype",
-                    options=["bf16", "fp32"],
+                    options=["bf16", "fp32", "none"],
                     default="bf16",
-                    tooltip="The dtype to use for training.",
+                    tooltip="The dtype to use for training. 'none' preserves the model's native compute dtype instead of overriding it. For fp16 models, GradScaler is automatically enabled.",
                 ),
                 io.Combo.Input(
                     "lora_dtype",
@@ -1035,7 +1051,7 @@ class TrainLoraNode(io.ComfyNode):
                 io.Boolean.Input(
                     "offloading",
                     default=False,
-                    tooltip="Offload the Model to RAM. Requires Bypass Mode.",
+                    tooltip="Offload model weights to CPU during training to save GPU memory.",
                 ),
                 io.Combo.Input(
                     "existing_lora",
@@ -1120,22 +1136,32 @@ class TrainLoraNode(io.ComfyNode):
 
         # Setup model and dtype
         mp = model.clone()
-        dtype = node_helpers.string_to_torch_dtype(training_dtype)
+        use_grad_scaler = False
+        if training_dtype != "none":
+            dtype = node_helpers.string_to_torch_dtype(training_dtype)
+            mp.set_model_compute_dtype(dtype)
+        else:
+            # Detect model's native dtype for autocast
+            model_dtype = mp.model.get_dtype()
+            if model_dtype == torch.float16:
+                dtype = torch.float16
+                use_grad_scaler = True
+                # Warn about fp16 accumulation instability during training
+                if PerformanceFeature.Fp16Accumulation in args.fast:
+                    logging.warning(
+                        "WARNING: FP16 model detected with fp16_accumulation enabled. "
+                        "This combination can be numerically unstable during training and may cause NaN values. "
+                        "Suggested fixes: 1) Set training_dtype to 'bf16', or 2) Disable fp16_accumulation (remove from --fast flags)."
+                    )
+            else:
+                # For fp8, bf16, or other dtypes, use bf16 autocast
+                dtype = torch.bfloat16
         lora_dtype = node_helpers.string_to_torch_dtype(lora_dtype)
-        mp.set_model_compute_dtype(dtype)
-
-        if mp.is_dynamic():
-            if not bypass_mode:
-                logging.info("Training MP is Dynamic - forcing bypass mode. Start comfy with --highvram to force weight diff mode")
-                bypass_mode = True
-            offloading = True
-        elif offloading:
-            if not bypass_mode:
-                logging.info("Training Offload selected - forcing bypass mode. Set bypass = True to remove this message")
 
         # Prepare latents and compute counts
+        latents_dtype = dtype if dtype not in (None,) else torch.bfloat16
         latents, num_images, multi_res = _prepare_latents_and_count(
-            latents, dtype, bucket_mode
+            latents, latents_dtype, bucket_mode
         )
 
         # Validate and expand conditioning
@@ -1201,6 +1227,7 @@ class TrainLoraNode(io.ComfyNode):
                     seed=seed,
                     training_dtype=dtype,
                     bucket_latents=latents,
+                    use_grad_scaler=use_grad_scaler,
                 )
             else:
                 train_sampler = TrainSampler(
@@ -1213,6 +1240,7 @@ class TrainLoraNode(io.ComfyNode):
                     seed=seed,
                     training_dtype=dtype,
                     real_dataset=latents if multi_res else None,
+                    use_grad_scaler=use_grad_scaler,
                 )
 
             # Setup guider
@@ -1337,7 +1365,7 @@ class SaveLoRA(io.ComfyNode):
                 io.Int.Input(
                     "steps",
                     optional=True,
-                    tooltip="Optional: The number of steps to LoRA has been trained for, used to name the saved file.",
+                    tooltip="Optional: The number of steps the LoRA has been trained for, used to name the saved file.",
                 ),
             ],
             outputs=[],

main.py

@@ -9,6 +9,8 @@ import folder_paths
 import time
 from comfy.cli_args import args, enables_dynamic_vram
 from app.logger import setup_logger
+from app.assets.seeder import asset_seeder
+from app.assets.services import register_output_files
 import itertools
 import utils.extra_config
 from utils.mime_types import init_mime_types
@@ -192,7 +194,6 @@ if 'torch' in sys.modules:
 
 
 import comfy.utils
-from app.assets.seeder import asset_seeder
 
 import execution
 import server
@@ -240,6 +241,26 @@ def cuda_malloc_warning():
             logging.warning("\nWARNING: this card most likely does not support cuda-malloc, if you get \"CUDA error\" please run ComfyUI with: --disable-cuda-malloc\n")
 
 
+def _collect_output_absolute_paths(history_result: dict) -> list[str]:
+    """Extract absolute file paths for output items from a history result."""
+    paths = []
+    base_dir = folder_paths.get_directory_by_type("output")
+    if base_dir is None:
+        return paths
+    for node_output in history_result.get("outputs", {}).values():
+        for items in node_output.values():
+            if not isinstance(items, list):
+                continue
+            for item in items:
+                if not isinstance(item, dict) or item.get("type") != "output":
+                    continue
+                filename = item.get("filename")
+                if not filename:
+                    continue
+                paths.append(os.path.join(base_dir, item.get("subfolder", ""), filename))
+    return paths
+
+
 def prompt_worker(q, server_instance):
     current_time: float = 0.0
     cache_type = execution.CacheType.CLASSIC
@@ -274,6 +295,7 @@ def prompt_worker(q, server_instance):
 
             asset_seeder.pause()
             e.execute(item[2], prompt_id, extra_data, item[4])
+
             need_gc = True
 
             remove_sensitive = lambda prompt: prompt[:5] + prompt[6:]
@@ -317,6 +339,11 @@ def prompt_worker(q, server_instance):
                 last_gc_collect = current_time
                 need_gc = False
                 hook_breaker_ac10a0.restore_functions()
+
+                if not asset_seeder.is_disabled():
+                    paths = _collect_output_absolute_paths(e.history_result)
+                    if register_output_files(paths, job_id=prompt_id) > 0:
+                        asset_seeder.enqueue_enrich(roots=("output",), compute_hashes=True)
                 asset_seeder.resume()
 
 

requirements.txt

@@ -1,5 +1,5 @@
 comfyui-frontend-package==1.41.20
-comfyui-workflow-templates==0.9.21
+comfyui-workflow-templates==0.9.26
 comfyui-embedded-docs==0.4.3
 torch
 torchsde

tests/test_asset_seeder.py

@@ -0,0 +1,250 @@
+"""Tests for app.assets.seeder – enqueue_enrich and pending-queue behaviour."""
+
+import threading
+from unittest.mock import patch
+
+import pytest
+
+from app.assets.seeder import Progress, _AssetSeeder, State
+
+
+@pytest.fixture()
+def seeder():
+    """Fresh seeder instance for each test."""
+    return _AssetSeeder()
+
+
+# ---------------------------------------------------------------------------
+# _reset_to_idle
+# ---------------------------------------------------------------------------
+
+
+class TestResetToIdle:
+    def test_sets_idle_and_clears_progress(self, seeder):
+        """_reset_to_idle should move state to IDLE and snapshot progress."""
+        progress = Progress(scanned=10, total=20, created=5, skipped=3)
+        seeder._state = State.RUNNING
+        seeder._progress = progress
+
+        with seeder._lock:
+            seeder._reset_to_idle()
+
+        assert seeder._state is State.IDLE
+        assert seeder._progress is None
+        assert seeder._last_progress is progress
+
+    def test_noop_when_progress_already_none(self, seeder):
+        """_reset_to_idle should handle None progress gracefully."""
+        seeder._state = State.CANCELLING
+        seeder._progress = None
+
+        with seeder._lock:
+            seeder._reset_to_idle()
+
+        assert seeder._state is State.IDLE
+        assert seeder._progress is None
+        assert seeder._last_progress is None
+
+
+# ---------------------------------------------------------------------------
+# enqueue_enrich – immediate start when idle
+# ---------------------------------------------------------------------------
+
+
+class TestEnqueueEnrichStartsImmediately:
+    def test_starts_when_idle(self, seeder):
+        """enqueue_enrich should delegate to start_enrich and return True when idle."""
+        with patch.object(seeder, "start_enrich", return_value=True) as mock:
+            assert seeder.enqueue_enrich(roots=("output",), compute_hashes=True) is True
+            mock.assert_called_once_with(roots=("output",), compute_hashes=True)
+
+    def test_no_pending_when_started_immediately(self, seeder):
+        """No pending request should be stored when start_enrich succeeds."""
+        with patch.object(seeder, "start_enrich", return_value=True):
+            seeder.enqueue_enrich(roots=("output",))
+        assert seeder._pending_enrich is None
+
+
+# ---------------------------------------------------------------------------
+# enqueue_enrich – queuing when busy
+# ---------------------------------------------------------------------------
+
+
+class TestEnqueueEnrichQueuesWhenBusy:
+    def test_queues_when_busy(self, seeder):
+        """enqueue_enrich should store a pending request when seeder is busy."""
+        with patch.object(seeder, "start_enrich", return_value=False):
+            result = seeder.enqueue_enrich(roots=("models",), compute_hashes=False)
+
+        assert result is False
+        assert seeder._pending_enrich == {
+            "roots": ("models",),
+            "compute_hashes": False,
+        }
+
+    def test_queues_preserves_compute_hashes_true(self, seeder):
+        with patch.object(seeder, "start_enrich", return_value=False):
+            seeder.enqueue_enrich(roots=("input",), compute_hashes=True)
+
+        assert seeder._pending_enrich["compute_hashes"] is True
+
+
+# ---------------------------------------------------------------------------
+# enqueue_enrich – merging when a pending request already exists
+# ---------------------------------------------------------------------------
+
+
+class TestEnqueueEnrichMergesPending:
+    def _make_busy(self, seeder):
+        """Patch start_enrich to always return False (seeder busy)."""
+        return patch.object(seeder, "start_enrich", return_value=False)
+
+    def test_merges_roots(self, seeder):
+        """A second enqueue should merge roots with the existing pending request."""
+        with self._make_busy(seeder):
+            seeder.enqueue_enrich(roots=("models",))
+            seeder.enqueue_enrich(roots=("output",))
+
+        merged = set(seeder._pending_enrich["roots"])
+        assert merged == {"models", "output"}
+
+    def test_merges_overlapping_roots(self, seeder):
+        """Duplicate roots should be deduplicated."""
+        with self._make_busy(seeder):
+            seeder.enqueue_enrich(roots=("models", "input"))
+            seeder.enqueue_enrich(roots=("input", "output"))
+
+        merged = set(seeder._pending_enrich["roots"])
+        assert merged == {"models", "input", "output"}
+
+    def test_compute_hashes_sticky_true(self, seeder):
+        """Once compute_hashes is True it should stay True after merging."""
+        with self._make_busy(seeder):
+            seeder.enqueue_enrich(roots=("models",), compute_hashes=True)
+            seeder.enqueue_enrich(roots=("output",), compute_hashes=False)
+
+        assert seeder._pending_enrich["compute_hashes"] is True
+
+    def test_compute_hashes_upgrades_to_true(self, seeder):
+        """A later enqueue with compute_hashes=True should upgrade the pending request."""
+        with self._make_busy(seeder):
+            seeder.enqueue_enrich(roots=("models",), compute_hashes=False)
+            seeder.enqueue_enrich(roots=("output",), compute_hashes=True)
+
+        assert seeder._pending_enrich["compute_hashes"] is True
+
+    def test_compute_hashes_stays_false(self, seeder):
+        """If both enqueues have compute_hashes=False it stays False."""
+        with self._make_busy(seeder):
+            seeder.enqueue_enrich(roots=("models",), compute_hashes=False)
+            seeder.enqueue_enrich(roots=("output",), compute_hashes=False)
+
+        assert seeder._pending_enrich["compute_hashes"] is False
+
+    def test_triple_merge(self, seeder):
+        """Three successive enqueues should all merge correctly."""
+        with self._make_busy(seeder):
+            seeder.enqueue_enrich(roots=("models",), compute_hashes=False)
+            seeder.enqueue_enrich(roots=("input",), compute_hashes=False)
+            seeder.enqueue_enrich(roots=("output",), compute_hashes=True)
+
+        merged = set(seeder._pending_enrich["roots"])
+        assert merged == {"models", "input", "output"}
+        assert seeder._pending_enrich["compute_hashes"] is True
+
+
+# ---------------------------------------------------------------------------
+# Pending enrich drains after scan completes
+# ---------------------------------------------------------------------------
+
+
+class TestPendingEnrichDrain:
+    """Verify that _run_scan drains _pending_enrich via start_enrich."""
+
+    @patch("app.assets.seeder.dependencies_available", return_value=True)
+    @patch("app.assets.seeder.get_all_known_prefixes", return_value=[])
+    @patch("app.assets.seeder.sync_root_safely", return_value=set())
+    @patch("app.assets.seeder.collect_paths_for_roots", return_value=[])
+    @patch("app.assets.seeder.build_asset_specs", return_value=([], {}, 0))
+    def test_pending_enrich_starts_after_scan(self, *_mocks):
+        """After a fast scan finishes, the pending enrich should be started."""
+        seeder = _AssetSeeder()
+
+        seeder._pending_enrich = {
+            "roots": ("output",),
+            "compute_hashes": True,
+        }
+
+        with patch.object(seeder, "start_enrich", return_value=True) as mock_start:
+            seeder.start_fast(roots=("models",))
+            seeder.wait(timeout=5)
+
+            mock_start.assert_called_once_with(
+                roots=("output",),
+                compute_hashes=True,
+            )
+
+        assert seeder._pending_enrich is None
+
+    @patch("app.assets.seeder.dependencies_available", return_value=True)
+    @patch("app.assets.seeder.get_all_known_prefixes", return_value=[])
+    @patch("app.assets.seeder.sync_root_safely", return_value=set())
+    @patch("app.assets.seeder.collect_paths_for_roots", return_value=[])
+    @patch("app.assets.seeder.build_asset_specs", return_value=([], {}, 0))
+    def test_pending_cleared_even_when_start_fails(self, *_mocks):
+        """_pending_enrich should be cleared even if start_enrich returns False."""
+        seeder = _AssetSeeder()
+        seeder._pending_enrich = {
+            "roots": ("output",),
+            "compute_hashes": False,
+        }
+
+        with patch.object(seeder, "start_enrich", return_value=False):
+            seeder.start_fast(roots=("models",))
+            seeder.wait(timeout=5)
+
+        assert seeder._pending_enrich is None
+
+    @patch("app.assets.seeder.dependencies_available", return_value=True)
+    @patch("app.assets.seeder.get_all_known_prefixes", return_value=[])
+    @patch("app.assets.seeder.sync_root_safely", return_value=set())
+    @patch("app.assets.seeder.collect_paths_for_roots", return_value=[])
+    @patch("app.assets.seeder.build_asset_specs", return_value=([], {}, 0))
+    def test_no_drain_when_no_pending(self, *_mocks):
+        """start_enrich should not be called when there is no pending request."""
+        seeder = _AssetSeeder()
+        assert seeder._pending_enrich is None
+
+        with patch.object(seeder, "start_enrich", return_value=True) as mock_start:
+            seeder.start_fast(roots=("models",))
+            seeder.wait(timeout=5)
+
+            mock_start.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Thread-safety of enqueue_enrich
+# ---------------------------------------------------------------------------
+
+
+class TestEnqueueEnrichThreadSafety:
+    def test_concurrent_enqueues(self, seeder):
+        """Multiple threads enqueuing should not lose roots."""
+        with patch.object(seeder, "start_enrich", return_value=False):
+            barrier = threading.Barrier(3)
+
+            def enqueue(root):
+                barrier.wait()
+                seeder.enqueue_enrich(roots=(root,), compute_hashes=False)
+
+            threads = [
+                threading.Thread(target=enqueue, args=(r,))
+                for r in ("models", "input", "output")
+            ]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join(timeout=5)
+
+        merged = set(seeder._pending_enrich["roots"])
+        assert merged == {"models", "input", "output"}