diff --git a/app/assets/event_log.py b/app/assets/event_log.py
new file mode 100644
index 000000000..e8962a3ad
--- /dev/null
+++ b/app/assets/event_log.py
@@ -0,0 +1,163 @@
+"""Structured event log lines for the assets system.
+
+Every line is ``[assets-event] <event> key=value ...`` on the standard logging
+INFO channel, with fields sorted by name and omitted when an event has none. This
+mirrors the ``assets.seed.*`` events the seeder already puts on the PromptServer
+bus. A log-tailing launcher can pick assets health signals out of core's output
+without parsing prose, and the existing human-readable lines stay exactly as
+they are.
+
+The field vocabulary is closed. Only the names in :data:`ALLOWED_FIELDS` may be
+carried, each has a validator, and no string value may contain a path separator
+or logfmt delimiter — so file names, paths, asset ids and content hashes cannot
+ride along.
+"""
+
+import logging
+import os
+import traceback
+from collections.abc import Callable
+from typing import Any
+
+TAG = "[assets-event]"
+
+MAX_STRING_LENGTH = 64
+FORBIDDEN_STRING_CHARS = ("/", "\\", ":", " ", "=", '"')
+
+ROOTS = frozenset({"models", "input", "output", "user", "temp"})
+PHASES = frozenset({"fast", "enrich", "full"})
+STAGES = frozenset({"mark_missing", "pruning", "fast_scan", "enrich", "finalize"})
+STAT_SITES = frozenset({"discovery", "enrich"})
+ALLOWED_EVENTS = frozenset({
+    "assets.enabled",
+    "seeder.scan_started",
+    "seeder.scan_completed",
+    "seeder.scan_failed",
+    "seeder.scan_cancelled",
+    "seeder.marked_missing",
+    "seeder.batch_insert_failed",
+    "scanner.hash_failed",
+    "scanner.enrich_failed",
+    "scanner.hash_discarded_modified",
+    "scanner.fast_scan_failed",
+    "scanner.temp_sync_failed",
+    "scanner.mark_missing_failed",
+    "scanner.stat_failed",
+})
+
+
+class EventLogError(ValueError):
+    """An emit() call that would break the closed event vocabulary."""
+
+
+def _is_safe_string(value: Any) -> bool:
+    return (
+        isinstance(value, str)
+        and 0 < len(value) <= MAX_STRING_LENGTH
+        and not any(char in value for char in FORBIDDEN_STRING_CHARS)
+    )
+
+
+def _one_of(allowed: frozenset[str]) -> Callable[[Any], bool]:
+    def validate(value: Any) -> bool:
+        return _is_safe_string(value) and value in allowed
+
+    return validate
+
+
+def _is_count(value: Any) -> bool:
+    # bool subclasses int, so it has to be excluded before the int check.
+    return isinstance(value, int) and not isinstance(value, bool)
+
+
+def _is_flag(value: Any) -> bool:
+    return isinstance(value, bool)
+
+
+ALLOWED_FIELDS: dict[str, Callable[[Any], bool]] = {
+    "root": _one_of(ROOTS),
+    "phase": _one_of(PHASES),
+    "stage": _one_of(STAGES),
+    "elapsed_ms": _is_count,
+    "created": _is_count,
+    "enriched": _is_count,
+    "skipped": _is_count,
+    "hash_failed": _is_count,
+    "enrich_failed": _is_count,
+    "permission_denied": _is_count,
+    "count": _is_count,
+    "error_type": _is_safe_string,
+    "hashing_enabled": _is_flag,
+    "site": _one_of(STAT_SITES),
+}
+
+_warned_call_sites: set[tuple[str, int]] = set()
+
+
+def _find_problem(event: Any, fields: dict[str, Any]) -> str | None:
+    if not isinstance(event, str) or event not in ALLOWED_EVENTS:
+        return "invalid event name"
+    for name, value in fields.items():
+        validate = ALLOWED_FIELDS.get(name)
+        if validate is None:
+            return f"field {name!r} is not in the allowed vocabulary"
+        if not validate(value):
+            return f"field {name!r} has a value its validator rejected"
+    return None
+
+
+def _strict_mode() -> bool:
+    return (
+        "PYTEST_CURRENT_TEST" in os.environ
+        or os.environ.get("COMFYUI_ASSETS_EVENT_LOG_STRICT") == "1"
+    )
+
+
+def _caller_call_site() -> tuple[str, int]:
+    """Identify emit()'s caller so a bad call site warns at most once."""
+    caller = traceback.extract_stack(limit=3)[0]
+    return (caller.filename, caller.lineno or 0)
+
+
+def emit(event: str, *, root: str | None = None, **fields: Any) -> None:
+    """Log one tagged event line.
+
+    An invalid call raises in strict mode (under pytest, or with
+    COMFYUI_ASSETS_EVENT_LOG_STRICT=1) so a bad call site fails the test suite.
+    In production it warns once per call site and drops the event, so a
+    vocabulary mistake can never break a running server.
+    """
+    if root is not None:
+        fields["root"] = root
+
+    problem = _find_problem(event, fields)
+    if problem is None:
+        pairs = " ".join(
+            f"{name}={str(value).lower() if isinstance(value, bool) else value}"
+            for name, value in sorted(fields.items())
+        )
+        line = f"{TAG} {event}" + (f" {pairs}" if pairs else "")
+        logging.info("%s", line)
+        return
+
+    if _strict_mode():
+        raise EventLogError(problem)
+
+    call_site = _caller_call_site()
+    if call_site not in _warned_call_sites:
+        _warned_call_sites.add(call_site)
+        logging.warning(
+            "Dropped an invalid assets event at %s:%d: %s",
+            call_site[0],
+            call_site[1],
+            problem,
+        )
+
+
+def error_type(exc: BaseException) -> str:
+    """The only sanctioned description of an exception: its class name.
+
+    Stringifying the exception itself is banned here, because FileNotFoundError
+    and friends embed the path that triggered them.
+    """
+    return type(exc).__name__
diff --git a/app/assets/scanner.py b/app/assets/scanner.py
index 7f8ce561b..76821e4c3 100644
--- a/app/assets/scanner.py
+++ b/app/assets/scanner.py
@@ -11,13 +11,14 @@ import logging
 import os
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Callable, Literal, TypedDict
+from typing import Callable, Literal, Protocol, TypedDict
 
 import folder_paths
 import sqlalchemy as sa
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 from app.assets import mode
+from app.assets.event_log import emit, error_type
 from app.assets.database.queries import (
     create_content_reporting_insert,
     mark_content_missing,
@@ -66,6 +67,14 @@ __all__ = [
 RootType = Literal["models", "input", "output"]
 
 
+class _ScanProgress(Protocol):
+    hash_failed: int
+    enrich_failed: int
+    permission_denied: int
+
+    def mark_emitted(self, key: str) -> bool: ...
+
+
 class SeedAssetSpec(TypedDict):
 
     abs_path: str
@@ -147,11 +156,13 @@ def sync_references_with_filesystem(
     session,
     root: RootType,
     collect_existing_paths: bool = False,
+    progress: _ScanProgress | None = None,
 ) -> set[str] | None:
     return sync_prefixes_with_filesystem(
         session,
         get_scan_prefixes_for_root(root),
         collect_existing_paths=collect_existing_paths,
+        progress=progress,
     )
 
 
@@ -159,6 +170,7 @@ def sync_prefixes_with_filesystem(
     session: Session,
     prefixes: list[str],
     collect_existing_paths: bool = False,
+    progress: _ScanProgress | None = None,
 ) -> set[str] | None:
     if not prefixes:
         return set() if collect_existing_paths else None
@@ -171,6 +183,8 @@ def sync_prefixes_with_filesystem(
             mark_content_missing(session, content.id)
         except PermissionError as e:
             _log_scan_error("reference_stat", e)
+            if progress is not None:
+                progress.permission_denied += 1
             logging.debug("Permission denied accessing %s", content.path)
         except OSError as e:
             _log_scan_error("reference_stat", e)
@@ -192,7 +206,9 @@ def _is_under_prefixes(path: str, prefixes: list[str]) -> bool:
     return is_path_under_prefixes(path, prefixes)
 
 
-def sync_root_safely(root: RootType) -> set[str]:
+def sync_root_safely(
+    root: RootType, progress: _ScanProgress | None = None
+) -> set[str]:
     """Sync a single root's references with the filesystem.
 
     Returns survivors (existing paths) or empty set on failure.
@@ -203,22 +219,39 @@ def sync_root_safely(root: RootType) -> set[str]:
                 sess,
                 root,
                 collect_existing_paths=True,
+                progress=progress,
             )
             sess.commit()
             return survivors or set()
-    except Exception as e:
-        logging.exception("fast DB scan failed for %s: %s", root, e)
+    except Exception as exc:
+        logging.exception("fast DB scan failed for %s: %s", root, exc)
+        emit(
+            "scanner.fast_scan_failed",
+            root=root,
+            error_type=error_type(exc),
+        )
         return set()
 
 
-def sync_temp_references_safely() -> None:
+def sync_temp_references_safely(
+    progress: _ScanProgress | None = None,
+) -> None:
     """Retire temp references whose file is gone; temp is never scanned, so nothing else stats them."""
     try:
         with create_session() as sess:
-            sync_prefixes_with_filesystem(sess, get_temp_prefixes())
+            sync_prefixes_with_filesystem(
+                sess,
+                get_temp_prefixes(),
+                progress=progress,
+            )
             sess.commit()
-    except Exception as e:
-        logging.exception("temp reference sync failed: %s", e)
+    except Exception as exc:
+        logging.exception("temp reference sync failed: %s", exc)
+        emit(
+            "scanner.temp_sync_failed",
+            root="temp",
+            error_type=error_type(exc),
+        )
 
 
 def mark_missing_outside_prefixes_safely(prefixes: list[str]) -> int:
@@ -231,8 +264,12 @@ def mark_missing_outside_prefixes_safely(prefixes: list[str]) -> int:
             count = mark_contents_missing_outside_prefixes(sess, prefixes)
             sess.commit()
             return count
-    except Exception as e:
-        logging.exception("marking missing assets failed: %s", e)
+    except Exception as exc:
+        logging.exception("marking missing assets failed: %s", exc)
+        emit(
+            "scanner.mark_missing_failed",
+            error_type=error_type(exc),
+        )
         return 0
 
 
@@ -264,6 +301,7 @@ def build_asset_specs(
     paths: list[str],
     existing_paths: set[str],
     enable_metadata_extraction: bool = True,
+    progress: _ScanProgress | None = None,
 ) -> tuple[list[SeedAssetSpec], set[str], int]:
     """Build asset specs from paths, returning (specs, tag_pool, skipped_count).
 
@@ -271,6 +309,7 @@ def build_asset_specs(
         paths: List of file paths to process
         existing_paths: Set of paths that already exist in the database
         enable_metadata_extraction: If True, extract tier 1 & 2 metadata
+        progress: Optional per-scan state for emit-once bookkeeping
     """
     specs: list[SeedAssetSpec] = []
     tag_pool: set[str] = set()
@@ -291,6 +330,11 @@ def build_asset_specs(
             continue
         except OSError as e:
             _log_scan_error("discovery_stat", e)
+            if progress is not None:
+                if isinstance(e, PermissionError):
+                    progress.permission_denied += 1
+                if progress.mark_emitted("stat_failed:discovery"):
+                    emit("scanner.stat_failed", site="discovery", error_type=error_type(e))
             continue
         if not stat_p.st_size:
             continue
@@ -447,6 +491,7 @@ def enrich_asset(
     record_id: str,
     extract_metadata: bool = True,
     compute_hash: bool = False,
+    progress: _ScanProgress | None = None,
 ) -> bool:
     """Enrich a single asset with metadata and/or hash.
 
@@ -467,6 +512,11 @@ def enrich_asset(
         return False
     except OSError as e:
         _log_scan_error("enrichment_stat", e)
+        if progress is not None:
+            if isinstance(e, PermissionError):
+                progress.permission_denied += 1
+            if progress.mark_emitted("stat_failed:enrich"):
+                emit("scanner.stat_failed", site="enrich", error_type=error_type(e))
         return False
 
     initial_mtime_ns = get_mtime_ns(stat_p)
@@ -493,6 +543,8 @@ def enrich_asset(
         try:
             snapshot = snapshot_hash(file_path)
             if snapshot is None:
+                if progress is None or progress.mark_emitted("hash_discarded_modified"):
+                    emit("scanner.hash_discarded_modified")
                 logging.warning(
                     "File modified during hashing (snapshot unstable), discarding hash: %s",
                     file_path,
@@ -500,11 +552,17 @@ def enrich_asset(
                 return False
             digest, verified_stat = snapshot
             stored_hash = to_stored_hash(digest)
-        except Exception as e:
-            if isinstance(e, OSError):
-                _log_scan_error("hashing", e)
+        except Exception as exc:
+            emit_failure = progress is None
+            if progress is not None:
+                progress.hash_failed += 1
+                emit_failure = progress.mark_emitted("hash_failed")
+            if emit_failure:
+                emit("scanner.hash_failed", error_type=error_type(exc))
+            if isinstance(exc, OSError):
+                _log_scan_error("hashing", exc)
             else:
-                logging.warning("Failed to hash %s: %s", file_path, e)
+                logging.warning("Failed to hash %s: %s", file_path, exc)
 
     record = session.get(Asset, record_id)
     if content is None or record is None or content.mtime_ns != initial_mtime_ns:
@@ -554,6 +612,7 @@ def enrich_assets_batch(
     extract_metadata: bool = True,
     compute_hash: bool = False,
     interrupt_check: Callable[[], bool] | None = None,
+    progress: _ScanProgress | None = None,
 ) -> tuple[int, list[str]]:
     """Enrich a batch of assets.
 
@@ -587,13 +646,18 @@ def enrich_assets_batch(
                     record_id=row.record_id,
                     extract_metadata=extract_metadata,
                     compute_hash=compute_hash,
+                    progress=progress,
                 )
                 if updated:
                     enriched += 1
                 else:
                     failed_ids.append(row.record_id)
-            except Exception as e:
-                logging.warning("Failed to enrich %s: %s", row.file_path, e)
+            except Exception as exc:
+                if progress is not None:
+                    progress.enrich_failed += 1
+                if progress is None or progress.mark_emitted("enrich_failed"):
+                    emit("scanner.enrich_failed", error_type=error_type(exc))
+                logging.warning("Failed to enrich %s: %s", row.file_path, exc)
                 sess.rollback()
                 failed_ids.append(row.record_id)
 
diff --git a/app/assets/seeder.py b/app/assets/seeder.py
index 82cc110a8..a78f1abf4 100644
--- a/app/assets/seeder.py
+++ b/app/assets/seeder.py
@@ -14,6 +14,7 @@ from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any, Callable, TypedDict
 
+from app.assets.event_log import emit, error_type
 from app.assets.scanner import (
     RootType,
     build_asset_specs,
@@ -60,14 +61,25 @@ class PendingScan(TypedDict):
     compute_hashes: bool
 
 
+class _ScanStage(Enum):
+    MARK_MISSING = "mark_missing"
+    PRUNING = "pruning"
+    FAST_SCAN = "fast_scan"
+    ENRICH = "enrich"
+    FINALIZE = "finalize"
+
+
 @dataclass
 class Progress:
-    """Progress information for a scan operation."""
+    """Public snapshot of a scan's progress. Carries counters only."""
 
     scanned: int = 0
     total: int = 0
     created: int = 0
     skipped: int = 0
+    hash_failed: int = 0
+    enrich_failed: int = 0
+    permission_denied: int = 0
 
 
 @dataclass
@@ -82,6 +94,46 @@ class ScanStatus:
 ProgressCallback = Callable[[Progress], None]
 
 
+@dataclass
+class _ScanState:
+    """Mutable in-flight state for one scan; never exposed outside this module.
+
+    Satisfies scanner.py's `_ScanProgress` Protocol. `cancel_stage` stores the
+    stage's string value (not the `_ScanStage` enum) so scanner.py never needs
+    to import `_ScanStage`. Take a `Progress` snapshot before exposing state.
+    """
+
+    scanned: int = 0
+    total: int = 0
+    created: int = 0
+    skipped: int = 0
+    hash_failed: int = 0
+    enrich_failed: int = 0
+    permission_denied: int = 0
+    cancel_stage: str | None = None
+    _emitted_keys: set[str] = field(default_factory=set)
+
+    def mark_emitted(self, key: str) -> bool:
+        """Return True the first call with `key` this scan, False every call after."""
+        if key in self._emitted_keys:
+            return False
+        self._emitted_keys.add(key)
+        return True
+
+
+def _snapshot_progress(state: _ScanState) -> Progress:
+    """Build the public counters-only `Progress` snapshot from live scan state."""
+    return Progress(
+        scanned=state.scanned,
+        total=state.total,
+        created=state.created,
+        skipped=state.skipped,
+        hash_failed=state.hash_failed,
+        enrich_failed=state.enrich_failed,
+        permission_denied=state.permission_denied,
+    )
+
+
 class _AssetSeeder:
     """Background asset scanning manager.
 
@@ -95,7 +147,7 @@ class _AssetSeeder:
         # holding _lock and re-enters start() which also acquires _lock.
         self._lock = threading.RLock()
         self._state = State.IDLE
-        self._progress: Progress | None = None
+        self._scan_state: _ScanState | None = None
         self._last_progress: Progress | None = None
         self._errors: list[str] = []
         self._thread: threading.Thread | None = None
@@ -155,7 +207,7 @@ class _AssetSeeder:
                 logging.info("Asset seeder already running, skipping start")
                 return False
             self._state = State.PAUSED if _start_paused else State.RUNNING
-            self._progress = Progress()
+            self._scan_state = _ScanState()
             self._errors = []
             self._roots = roots
             self._phase = phase
@@ -355,17 +407,14 @@ class _AssetSeeder:
     def get_status(self) -> ScanStatus:
         """Get the current status and progress of the seeder."""
         with self._lock:
-            src = self._progress or self._last_progress
+            progress = (
+                _snapshot_progress(self._scan_state)
+                if self._scan_state is not None
+                else self._last_progress
+            )
             return ScanStatus(
                 state=self._state,
-                progress=Progress(
-                    scanned=src.scanned,
-                    total=src.total,
-                    created=src.created,
-                    skipped=src.skipped,
-                )
-                if src
-                else None,
+                progress=progress,
                 errors=list(self._errors),
             )
 
@@ -425,6 +474,11 @@ class _AssetSeeder:
 
             all_prefixes = get_owned_prefixes()
             marked = mark_missing_outside_prefixes_safely(all_prefixes)
+            emit(
+                "seeder.marked_missing",
+                count=marked,
+                stage=_ScanStage.MARK_MISSING.value,
+            )
             if marked > 0:
                 logging.info("Marked %d references as missing", marked)
             return marked
@@ -434,9 +488,10 @@ class _AssetSeeder:
 
     def _reset_to_idle(self) -> None:
         """Reset state to IDLE, preserving last progress. Caller must hold _lock."""
-        self._last_progress = self._progress
+        if self._scan_state is not None:
+            self._last_progress = _snapshot_progress(self._scan_state)
         self._state = State.IDLE
-        self._progress = None
+        self._scan_state = None
 
     def _is_cancelled(self) -> bool:
         """Check if cancellation has been requested."""
@@ -450,9 +505,17 @@ class _AssetSeeder:
         open while blocked. The caller is responsible for blocking on
         _check_pause_and_cancel() afterward.
         """
-        return not self._run_gate.is_set() or self._cancel_event.is_set()
+        cancelled = self._cancel_event.is_set()
+        if cancelled:
+            self._record_cancel_stage(_ScanStage.ENRICH)
+        return not self._run_gate.is_set() or cancelled
 
-    def _check_pause_and_cancel(self) -> bool:
+    def _record_cancel_stage(self, stage: _ScanStage) -> None:
+        with self._lock:
+            if self._scan_state is not None and self._scan_state.cancel_stage is None:
+                self._scan_state.cancel_stage = stage.value
+
+    def _check_pause_and_cancel(self, stage: _ScanStage) -> bool:
         """Block while paused, then check if cancelled.
 
         Call this at checkpoint locations in scan loops. It will:
@@ -465,7 +528,10 @@ class _AssetSeeder:
         if not self._run_gate.is_set():
             self._emit_event("assets.seed.paused", {})
         self._run_gate.wait()  # Blocks if paused
-        return self._is_cancelled()
+        cancelled = self._is_cancelled()
+        if cancelled:
+            self._record_cancel_stage(stage)
+        return cancelled
 
     def _emit_event(self, event_type: str, data: dict[str, Any]) -> None:
         """Emit a WebSocket event if server is available."""
@@ -487,24 +553,19 @@ class _AssetSeeder:
         progress: Progress | None = None
 
         with self._lock:
-            if self._progress is None:
+            if self._scan_state is None:
                 return
             if scanned is not None:
-                self._progress.scanned = scanned
+                self._scan_state.scanned = scanned
             if total is not None:
-                self._progress.total = total
+                self._scan_state.total = total
             if created is not None:
-                self._progress.created = created
+                self._scan_state.created = created
             if skipped is not None:
-                self._progress.skipped = skipped
+                self._scan_state.skipped = skipped
             if self._progress_callback:
                 callback = self._progress_callback
-                progress = Progress(
-                    scanned=self._progress.scanned,
-                    total=self._progress.total,
-                    created=self._progress.created,
-                    skipped=self._progress.skipped,
-                )
+                progress = _snapshot_progress(self._scan_state)
 
         if callback and progress:
             try:
@@ -540,6 +601,7 @@ class _AssetSeeder:
         t_start = time.perf_counter()
         roots = self._roots
         phase = self._phase
+        root = roots[0] if len(roots) == 1 else None
         cancelled = False
         total_created = 0
         total_enriched = 0
@@ -555,14 +617,23 @@ class _AssetSeeder:
                 )
                 return
 
+            emit("seeder.scan_started", phase=phase.value, root=root)
+            assert self._scan_state is not None
+            scan_state = self._scan_state
+
             if self._prune_first:
                 all_prefixes = get_owned_prefixes()
                 marked = mark_missing_outside_prefixes_safely(all_prefixes)
+                emit(
+                    "seeder.marked_missing",
+                    count=marked,
+                    stage=_ScanStage.PRUNING.value,
+                )
                 if marked > 0:
                     logging.info("Marked %d refs as missing before scan", marked)
-                sync_temp_references_safely()
+                sync_temp_references_safely(scan_state)
 
-            if self._check_pause_and_cancel():
+            if self._check_pause_and_cancel(_ScanStage.PRUNING):
                 logging.info("Asset scan cancelled after pruning phase")
                 cancelled = True
                 return
@@ -574,7 +645,7 @@ class _AssetSeeder:
                 created, skipped, paths = self._run_fast_phase(roots)
                 total_created, skipped_existing, total_paths = created, skipped, paths
 
-                if self._check_pause_and_cancel():
+                if self._check_pause_and_cancel(_ScanStage.FAST_SCAN):
                     cancelled = True
                     return
 
@@ -590,7 +661,7 @@ class _AssetSeeder:
 
             # Phase 2: Enrichment scan (metadata + hashes)
             if phase in (ScanPhase.ENRICH, ScanPhase.FULL):
-                if self._check_pause_and_cancel():
+                if self._check_pause_and_cancel(_ScanStage.ENRICH):
                     cancelled = True
                     return
 
@@ -608,6 +679,14 @@ class _AssetSeeder:
                     },
                 )
 
+            # Deliberately non-blocking, unlike every other checkpoint: no work
+            # remains, so pausing here would hold the scan open with nothing to
+            # do until main.py's next resume (a whole gc_collect_interval away).
+            if self._is_cancelled():
+                self._record_cancel_stage(_ScanStage.FINALIZE)
+                cancelled = True
+                return
+
             elapsed = time.perf_counter() - t_start
             logging.info(
                 "Scan(%s, %s) done %.3fs: created=%d enriched=%d skipped=%d",
@@ -618,6 +697,18 @@ class _AssetSeeder:
                 total_enriched,
                 skipped_existing,
             )
+            emit(
+                "seeder.scan_completed",
+                phase=phase.value,
+                elapsed_ms=round(elapsed * 1000),
+                created=total_created,
+                enriched=total_enriched,
+                skipped=skipped_existing,
+                hash_failed=scan_state.hash_failed,
+                enrich_failed=scan_state.enrich_failed,
+                permission_denied=scan_state.permission_denied,
+                root=root,
+            )
 
             self._emit_event(
                 "assets.seed.completed",
@@ -634,35 +725,51 @@ class _AssetSeeder:
         except Exception as e:
             self._add_error(f"Scan failed: {e}")
             logging.exception("Asset scan failed")
+            emit(
+                "seeder.scan_failed",
+                phase=phase.value,
+                error_type=error_type(e),
+                root=root,
+            )
             self._emit_event("assets.seed.error", {"message": str(e)})
         finally:
-            if cancelled:
-                self._emit_event(
-                    "assets.seed.cancelled",
-                    {
-                        "scanned": self._progress.scanned if self._progress else 0,
-                        "total": total_paths,
-                        "created": total_created,
-                    },
-                )
-            with self._lock:
-                start_paused = self._state is State.PAUSED
-                self._reset_to_idle()
-                pending = self._pending_scan
-                if pending is not None:
-                    self._pending_scan = None
-                    if not self.start(
-                        roots=pending["roots"],
-                        phase=pending["phase"],
-                        prune_first=False,
-                        compute_hashes=pending["compute_hashes"],
-                        _start_paused=start_paused,
-                    ):
-                        logging.warning(
-                            "Pending scan could not start (roots=%s, phase=%s)",
-                            pending["roots"],
-                            pending["phase"].value,
+            try:
+                if cancelled:
+                    stage = self._scan_state.cancel_stage if self._scan_state else None
+                    if stage is not None:
+                        emit(
+                            "seeder.scan_cancelled",
+                            phase=phase.value,
+                            stage=stage,
+                            root=root,
                         )
+                        self._emit_event(
+                            "assets.seed.cancelled",
+                            {
+                                "scanned": self._scan_state.scanned if self._scan_state else 0,
+                                "total": total_paths,
+                                "created": total_created,
+                            },
+                        )
+            finally:
+                with self._lock:
+                    start_paused = self._state is State.PAUSED
+                    self._reset_to_idle()
+                    pending = self._pending_scan
+                    if pending is not None:
+                        self._pending_scan = None
+                        if not self.start(
+                            roots=pending["roots"],
+                            phase=pending["phase"],
+                            prune_first=False,
+                            compute_hashes=pending["compute_hashes"],
+                            _start_paused=start_paused,
+                        ):
+                            logging.warning(
+                                "Pending scan could not start (roots=%s, phase=%s)",
+                                pending["roots"],
+                                pending["phase"].value,
+                            )
 
     def _run_fast_phase(self, roots: tuple[RootType, ...]) -> tuple[int, int, int]:
         """Run phase 1: fast scan to create stub records.
@@ -676,17 +783,19 @@ class _AssetSeeder:
 
         existing_paths: set[str] = set()
         t_sync = time.perf_counter()
+        assert self._scan_state is not None
+        scan_state = self._scan_state
         for r in roots:
-            if self._check_pause_and_cancel():
+            if self._check_pause_and_cancel(_ScanStage.FAST_SCAN):
                 return total_created, skipped_existing, 0
-            existing_paths.update(sync_root_safely(r))
+            existing_paths.update(sync_root_safely(r, scan_state))
         logging.debug(
             "Fast scan: sync_root phase took %.3fs (%d existing paths)",
             time.perf_counter() - t_sync,
             len(existing_paths),
         )
 
-        if self._check_pause_and_cancel():
+        if self._check_pause_and_cancel(_ScanStage.FAST_SCAN):
             return total_created, skipped_existing, 0
 
         t_collect = time.perf_counter()
@@ -710,6 +819,7 @@ class _AssetSeeder:
             paths,
             existing_paths,
             enable_metadata_extraction=False,
+            progress=scan_state,
         )
         logging.debug(
             "Fast scan: build_asset_specs took %.3fs (%d specs, %d skipped)",
@@ -719,7 +829,7 @@ class _AssetSeeder:
         )
         self._update_progress(skipped=skipped_existing)
 
-        if self._check_pause_and_cancel():
+        if self._check_pause_and_cancel(_ScanStage.FAST_SCAN):
             return total_created, skipped_existing, total_paths
 
         batch_size = 500
@@ -727,7 +837,7 @@ class _AssetSeeder:
         progress_interval = 1.0
 
         for i in range(0, len(specs), batch_size):
-            if self._check_pause_and_cancel():
+            if self._check_pause_and_cancel(_ScanStage.FAST_SCAN):
                 logging.info(
                     "Fast scan cancelled after %d/%d files (created=%d)",
                     i,
@@ -744,6 +854,7 @@ class _AssetSeeder:
             except Exception as e:
                 self._add_error(f"Batch insert failed at offset {i}: {e}")
                 logging.exception("Batch insert failed at offset %d", i)
+                emit("seeder.batch_insert_failed", error_type=error_type(e))
 
             scanned = i + len(batch)
             now = time.perf_counter()
@@ -781,6 +892,7 @@ class _AssetSeeder:
             Tuple of (cancelled, total_enriched)
         """
         total_enriched = 0
+        scan_state = self._scan_state
         with create_session() as session:
             drain_pending_verifications(session)
             tick_watch_list(session)
@@ -803,7 +915,7 @@ class _AssetSeeder:
         max_consecutive_empty = 3
 
         while True:
-            if self._check_pause_and_cancel():
+            if self._check_pause_and_cancel(_ScanStage.ENRICH):
                 logging.info("Enrich scan cancelled after %d assets", total_enriched)
                 return True, total_enriched
 
@@ -826,6 +938,7 @@ class _AssetSeeder:
                 extract_metadata=True,
                 compute_hash=self._compute_hashes,
                 interrupt_check=self._is_paused_or_cancelled,
+                progress=scan_state,
             )
             total_enriched += enriched
             skip_ids.update(failed_ids)
diff --git a/server.py b/server.py
index aedfa4ec7..ac8719110 100644
--- a/server.py
+++ b/server.py
@@ -45,6 +45,7 @@ from comfyui_version import __version__
 from app.frontend_management import FrontendManager, parse_version
 from comfy_api.internal import _ComfyNodeInternal
 from app.assets.services.asset_management import resolve_hash_to_path
+from app.assets.event_log import emit
 
 from app.user_manager import UserManager
 from app.model_manager import ModelFileManager
@@ -256,6 +257,8 @@ class PromptServer():
         logging.info(f"[Prompt Server] web root: {self.web_root}")
         self.asset_manager.register_routes(self.app, self.user_manager)
         self.asset_manager.set_event_sink(self.send_sync)
+        if self.asset_manager.enabled:
+            emit("assets.enabled", hashing_enabled=args.enable_asset_hashing)
         routes = web.RouteTableDef()
         self.routes = routes
         self.last_node_id = None
diff --git a/tests-unit/assets_test/fixtures/assets_event_lines.txt b/tests-unit/assets_test/fixtures/assets_event_lines.txt
new file mode 100644
index 000000000..2aa36ce15
--- /dev/null
+++ b/tests-unit/assets_test/fixtures/assets_event_lines.txt
@@ -0,0 +1,3 @@
+[assets-event] seeder.scan_completed created=12 elapsed_ms=8123 enrich_failed=0 enriched=4 hash_failed=2 permission_denied=0 phase=fast root=models skipped=3
+[assets-event] seeder.scan_started phase=enrich
+[assets-event] scanner.stat_failed error_type=PermissionError site=discovery
diff --git a/tests-unit/assets_test/test_event_log.py b/tests-unit/assets_test/test_event_log.py
new file mode 100644
index 000000000..27c9bba13
--- /dev/null
+++ b/tests-unit/assets_test/test_event_log.py
@@ -0,0 +1,280 @@
+"""Tests for the structured assets event log lines (``app/assets/event_log.py``)."""
+
+import logging
+import re
+from pathlib import Path
+
+import pytest
+
+from app.assets import event_log
+from app.assets.event_log import ALLOWED_FIELDS, TAG, EventLogError, emit, error_type
+
+# The line grammar below is the CONTRACT shared with the desktop launcher's log
+# tap: Comfy-Org/Comfy-Desktop `src/main/lib/assetsTap.ts` holds the equivalent
+# regex, and `tests-unit/assets_test/fixtures/assets_event_lines.txt` is a
+# byte-identical copy of that repo's `src/main/lib/__fixtures__/assets-event-lines.txt`.
+# Neither side may change without the other.
+EVENT_LINE_PATTERN = re.compile(
+    r"^\[assets-event\] (?P<event>[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)*)"
+    r"(?P<fields>(?: [a-z_]+=[^ =]+)*)$"
+)
+
+FIXTURE_PATH = Path(__file__).parent / "fixtures" / "assets_event_lines.txt"
+
+# One valid value per allowed field, covering every enum member so the desktop
+# tap's mirrored validator matrix has a counterpart on this side.
+VALID_VALUES: dict[str, list[object]] = {
+    "root": ["models", "input", "output", "user", "temp"],
+    "phase": ["fast", "enrich", "full"],
+    "stage": ["mark_missing", "pruning", "fast_scan", "enrich", "finalize"],
+    "elapsed_ms": [0, 8123],
+    "created": [0, 12],
+    "enriched": [4],
+    "skipped": [3],
+    "hash_failed": [2],
+    "enrich_failed": [0],
+    "permission_denied": [0],
+    "count": [1],
+    "error_type": ["ValueError", "FileNotFoundError"],
+    "hashing_enabled": [True, False],
+    "site": ["discovery", "enrich"],
+}
+
+
+@pytest.fixture(autouse=True)
+def autoclean_unit_test_assets():
+    """Shadow the conftest fixture of the same name.
+
+    The conftest version reaches a running server to delete test-tagged assets,
+    which transitively boots ComfyUI for every test in this directory. Nothing
+    here touches a server or creates an asset, so the boot is pure cost.
+    """
+    yield
+
+
+def fixture_lines() -> list[str]:
+    return FIXTURE_PATH.read_text(encoding="utf-8").splitlines()
+
+
+def parse_fields(raw: str) -> dict[str, bool | int | str]:
+    fields: dict[str, bool | int | str] = {}
+    for pair in raw.split():
+        name, value = pair.split("=", maxsplit=1)
+        if value == "true":
+            fields[name] = True
+        elif value == "false":
+            fields[name] = False
+        elif value.removeprefix("-").isdigit():
+            fields[name] = int(value)
+        else:
+            fields[name] = value
+    return fields
+
+
+def emit_line(caplog: pytest.LogCaptureFixture, event: str, **fields: object) -> str:
+    """Emit one event and return the single tagged line it produced."""
+    caplog.clear()
+    with caplog.at_level(logging.INFO):
+        emit(event, **fields)
+    tagged = [r.getMessage() for r in caplog.records if r.getMessage().startswith(TAG)]
+    assert len(tagged) == 1, tagged
+    return tagged[0]
+
+
+def go_to_production_mode(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Leave strict mode so invalid calls warn-and-drop instead of raising."""
+    monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False)
+    monkeypatch.delenv("COMFYUI_ASSETS_EVENT_LOG_STRICT", raising=False)
+    event_log._warned_call_sites.clear()
+
+
+# --- the shared cross-repo fixture -------------------------------------------------
+
+
+def test_shared_fixture_file_holds_three_newline_terminated_lines():
+    raw = FIXTURE_PATH.read_text(encoding="utf-8")
+
+    assert raw.endswith("\n")
+    assert len(raw.splitlines()) == 3
+
+
+@pytest.mark.parametrize("line", fixture_lines())
+def test_emit_reproduces_each_shared_fixture_line_byte_for_byte(caplog, line):
+    """Given a canonical line, When its fields are re-emitted, Then the bytes match."""
+    match = EVENT_LINE_PATTERN.match(line)
+    assert match is not None, line
+    fields = parse_fields(match.group("fields"))
+
+    assert emit_line(caplog, match.group("event"), **fields) == line
+
+
+# --- line shape ---------------------------------------------------------------------
+
+
+def test_fields_are_serialized_as_sorted_logfmt(caplog):
+    line = emit_line(caplog, "seeder.scan_started", root="models", phase="fast")
+
+    assert line == "[assets-event] seeder.scan_started phase=fast root=models"
+
+
+def test_a_fieldless_event_still_matches_the_shared_pattern(caplog):
+    line = emit_line(caplog, "scanner.hash_discarded_modified")
+
+    assert line == "[assets-event] scanner.hash_discarded_modified"
+    assert EVENT_LINE_PATTERN.match(line) is not None
+
+
+def test_the_emitted_record_is_a_single_line(caplog):
+    line = emit_line(caplog, "seeder.scan_failed", error_type="ValueError")
+
+    assert "\n" not in line
+    assert "\r" not in line
+
+
+# --- error_type ---------------------------------------------------------------------
+
+
+def test_error_type_is_the_class_name_and_the_path_never_reaches_the_line(caplog):
+    exc = FileNotFoundError("/home/x/model.safetensors")
+
+    assert error_type(exc) == "FileNotFoundError"
+
+    line = emit_line(caplog, "seeder.scan_failed", error_type=error_type(exc))
+    assert "/home/x/model.safetensors" not in line
+    assert "model.safetensors" not in line
+
+
+# --- the closed vocabulary ----------------------------------------------------------
+
+
+def test_the_valid_value_matrix_covers_every_allowed_field():
+    assert set(VALID_VALUES) == set(ALLOWED_FIELDS)
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [(field, value) for field, values in VALID_VALUES.items() for value in values],
+)
+def test_every_allowed_field_value_round_trips(caplog, field, value):
+    line = emit_line(caplog, "seeder.scan_completed", **{field: value})
+
+    match = EVENT_LINE_PATTERN.match(line)
+    assert match is not None, line
+    assert parse_fields(match.group("fields")) == {field: value}
+
+
+def test_unknown_field_raises_under_pytest():
+    with pytest.raises(EventLogError):
+        emit("seeder.scan_started", path="/home/x/models")
+
+
+@pytest.mark.parametrize("value", ["a/b", "a\\b", "a:b", "a b", "a=b", 'a"b'])
+def test_a_string_value_carrying_a_forbidden_character_raises(value):
+    with pytest.raises(EventLogError):
+        emit("seeder.scan_failed", error_type=value)
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("root", "checkpoints"),
+        ("root", 1),
+        ("phase", "quick"),
+        ("phase", None),
+        ("stage", "scanning"),
+        ("site", "reference"),
+        ("error_type", "x" * 65),
+        ("error_type", ""),
+        ("error_type", 7),
+        ("elapsed_ms", "8123"),
+        ("count", 1.5),
+        ("created", True),
+        ("hashing_enabled", 1),
+        ("hashing_enabled", "true"),
+    ],
+    ids=[
+        "bad-root",
+        "non-string-root",
+        "bad-phase",
+        "none-phase",
+        "bad-stage",
+        "bad-site",
+        "oversized-string",
+        "empty-string",
+        "non-string-error-type",
+        "string-into-int-field",
+        "float-into-int-field",
+        "bool-into-int-field",
+        "int-into-bool-field",
+        "string-into-bool-field",
+    ],
+)
+def test_every_validator_rejects_its_bad_value(field, value):
+    with pytest.raises(EventLogError):
+        emit("seeder.scan_completed", **{field: value})
+
+
+@pytest.mark.parametrize(
+    "event",
+    ["", "Seeder.scan_started", "seeder..scan", "9seeder.scan", "seeder.scan-started", "seeder scan", "seeder.", "scanner.made_up"],
+)
+def test_an_invalid_event_name_raises(event):
+    with pytest.raises(EventLogError):
+        emit(event)
+
+
+# --- strict mode vs production mode -------------------------------------------------
+
+
+def test_the_env_var_enables_strict_mode_without_pytest(monkeypatch):
+    go_to_production_mode(monkeypatch)
+    monkeypatch.setenv("COMFYUI_ASSETS_EVENT_LOG_STRICT", "1")
+
+    with pytest.raises(EventLogError):
+        emit("seeder.scan_started", path="/home/x")
+
+
+def test_an_env_var_value_other_than_1_is_not_strict(caplog, monkeypatch):
+    go_to_production_mode(monkeypatch)
+    monkeypatch.setenv("COMFYUI_ASSETS_EVENT_LOG_STRICT", "true")
+
+    with caplog.at_level(logging.WARNING):
+        emit("seeder.scan_started", path="/home/x")
+
+    assert [r for r in caplog.records if r.levelno == logging.WARNING]
+
+
+def test_production_mode_warns_once_for_repeated_calls_from_one_call_site(caplog, monkeypatch):
+    go_to_production_mode(monkeypatch)
+    caplog.clear()
+
+    with caplog.at_level(logging.WARNING):
+        for _ in range(3):
+            emit("seeder.scan_started", path="/home/x/models")
+
+    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+    assert len(warnings) == 1
+    assert not [r for r in caplog.records if r.getMessage().startswith(TAG)]
+    assert "/home/x/models" not in caplog.text
+
+
+def test_production_mode_warns_once_per_distinct_call_site(caplog, monkeypatch):
+    go_to_production_mode(monkeypatch)
+    caplog.clear()
+
+    with caplog.at_level(logging.WARNING):
+        emit("seeder.scan_started", path="/home/x/models")
+        emit("seeder.scan_started", path="/home/x/models")
+
+    assert len([r for r in caplog.records if r.levelno == logging.WARNING]) == 2
+
+
+def test_production_mode_still_emits_valid_events_after_a_dropped_one(caplog, monkeypatch):
+    go_to_production_mode(monkeypatch)
+
+    with caplog.at_level(logging.INFO):
+        emit("seeder.scan_started", path="/home/x/models")
+        emit("seeder.scan_started", phase="fast")
+
+    tagged = [r.getMessage() for r in caplog.records if r.getMessage().startswith(TAG)]
+    assert tagged == ["[assets-event] seeder.scan_started phase=fast"]
diff --git a/tests-unit/assets_test/test_event_log_sites.py b/tests-unit/assets_test/test_event_log_sites.py
new file mode 100644
index 000000000..499713856
--- /dev/null
+++ b/tests-unit/assets_test/test_event_log_sites.py
@@ -0,0 +1,66 @@
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from app.assets.event_log import TAG
+
+
+STARTUP_SCRIPT = (
+    "import runpy, comfy_kitchen; "
+    "comfy_kitchen.int8_attention_is_available=lambda: False; "
+    'runpy.run_path("main.py", run_name="__main__")'
+)
+
+
+@pytest.fixture(autouse=True)
+def autoclean_unit_test_assets():
+    yield
+
+
+def run_quick_startup(tmp_path: Path, *flags: str) -> str:
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            STARTUP_SCRIPT,
+            "--cpu",
+            "--quick-test-for-ci",
+            "--disable-all-custom-nodes",
+            "--disable-api-nodes",
+            f"--base-directory={tmp_path}",
+            f"--front-end-root={tmp_path}",
+            f"--database-url=sqlite:///{tmp_path / 'assets.sqlite3'}",
+            *flags,
+        ],
+        cwd=Path(__file__).resolve().parents[2],
+        capture_output=True,
+        text=True,
+        timeout=120,
+        check=True,
+    )
+    return result.stdout + result.stderr
+
+
+@pytest.mark.parametrize(
+    ("hashing_flag", "expected"),
+    [
+        pytest.param((), False, id="hashing-disabled"),
+        pytest.param(("--enable-asset-hashing",), True, id="hashing-enabled"),
+    ],
+)
+def test_enabled_assets_emits_once_with_the_hashing_flag(
+    tmp_path: Path, hashing_flag: tuple[str, ...], expected: bool
+) -> None:
+    output = run_quick_startup(tmp_path, "--enable-assets", *hashing_flag)
+    lines = [line for line in output.splitlines() if f"{TAG} assets.enabled " in line]
+
+    assert len(lines) == 1
+    assert f"hashing_enabled={str(expected).lower()}" in lines[0]
+
+
+def test_noassets_emits_no_enabled_event(tmp_path: Path) -> None:
+    output = run_quick_startup(tmp_path)
+
+    assert f"{TAG} assets.enabled " not in output
diff --git a/tests-unit/assets_test/test_scanner_ingest_event_log.py b/tests-unit/assets_test/test_scanner_ingest_event_log.py
new file mode 100644
index 000000000..f81ebee0f
--- /dev/null
+++ b/tests-unit/assets_test/test_scanner_ingest_event_log.py
@@ -0,0 +1,436 @@
+import logging
+import re
+from contextlib import nullcontext
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from app.assets import scanner
+from app.assets.event_log import TAG
+from app.assets.scanner import UnenrichedContent
+from app.assets.seeder import _ScanState
+
+
+EVENT_LINE_PATTERN = re.compile(
+    rf"^{re.escape(TAG)} (?P<event>[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)*)"
+    r"(?P<fields>(?: [a-z_]+=[^ =]+)*)$"
+)
+EventFields = dict[str, bool | int | str]
+
+
+@pytest.fixture(autouse=True)
+def autoclean_unit_test_assets():
+    yield
+
+
+def parse_fields(raw: str) -> EventFields:
+    fields: EventFields = {}
+    for pair in raw.split():
+        name, value = pair.split("=", maxsplit=1)
+        if value == "true":
+            fields[name] = True
+        elif value == "false":
+            fields[name] = False
+        elif value.removeprefix("-").isdigit():
+            fields[name] = int(value)
+        else:
+            fields[name] = value
+    return fields
+
+
+def tagged_events(caplog: pytest.LogCaptureFixture) -> list[tuple[str, EventFields]]:
+    events: list[tuple[str, EventFields]] = []
+    for record in caplog.records:
+        match = EVENT_LINE_PATTERN.match(record.getMessage())
+        if match is not None:
+            events.append((match.group("event"), parse_fields(match.group("fields"))))
+    return events
+
+
+def events_named(
+    caplog: pytest.LogCaptureFixture, event_name: str
+) -> list[EventFields]:
+    return [fields for event, fields in tagged_events(caplog) if event == event_name]
+
+
+def tagged_lines(caplog: pytest.LogCaptureFixture) -> list[str]:
+    return [record.getMessage() for record in caplog.records if record.getMessage().startswith(TAG)]
+
+
+def hash_session(path: Path) -> Mock:
+    stat_result = path.stat()
+    content = SimpleNamespace(hash=None, mtime_ns=stat_result.st_mtime_ns)
+    record = SimpleNamespace(system_metadata=None, mime_type=None)
+    session = Mock()
+    session.get.side_effect = lambda _model, row_id: content if row_id == "content" else record
+    return session
+
+
+def run_hash_failure(session: Mock, path: Path, progress: _ScanState) -> bool:
+    return scanner.enrich_asset(
+        session,
+        file_path=str(path),
+        content_id="content",
+        record_id="record",
+        extract_metadata=False,
+        compute_hash=True,
+        progress=progress,
+    )
+
+
+@pytest.mark.parametrize(
+    ("operation", "event_name", "expected_fields", "expected_result"),
+    [
+        pytest.param(
+            lambda: scanner.sync_root_safely("models"),
+            "scanner.fast_scan_failed",
+            {"error_type": "FileNotFoundError", "root": "models"},
+            set(),
+            id="fast-scan",
+        ),
+    ],
+)
+def test_scanner_safe_failures_emit_exception_type_without_path(
+    operation,
+    event_name: str,
+    expected_fields: EventFields,
+    expected_result,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    secret_path = "/private/assets/secret.safetensors"
+
+    def fail_session():
+        raise FileNotFoundError(secret_path)
+
+    monkeypatch.setattr(scanner, "create_session", fail_session)
+
+    with caplog.at_level(logging.INFO):
+        result = operation()
+
+    assert result == expected_result
+    assert events_named(caplog, event_name) == [expected_fields]
+    assert all(secret_path not in line for line in tagged_lines(caplog))
+
+
+def test_permission_error_in_reference_sync_increments_scan_counter(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    secret_path = "/private/assets/unreadable.safetensors"
+    content = SimpleNamespace(id="content", path=secret_path)
+    progress = _ScanState()
+
+    def deny_stat(*_args, **_kwargs):
+        raise PermissionError(secret_path)
+
+    monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=deny_stat, path=scanner.os.path))
+    monkeypatch.setattr(scanner, "live_contents_under_prefixes", lambda _session, _prefixes: [content])
+
+    scanner.sync_prefixes_with_filesystem(Mock(), ["/private/assets"], progress=progress)
+
+    assert progress.permission_denied == 1
+
+
+def test_hash_failures_emit_once_per_scan_and_count_every_failure(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    path = tmp_path / "model.safetensors"
+    path.write_bytes(b"model")
+    progress = _ScanState()
+
+    def fail_hash(_path: str):
+        raise FileNotFoundError(str(path))
+
+    monkeypatch.setattr(scanner, "snapshot_hash", fail_hash)
+    session = hash_session(path)
+
+    with caplog.at_level(logging.INFO):
+        assert run_hash_failure(session, path, progress) is False
+        assert run_hash_failure(session, path, progress) is False
+
+    assert events_named(caplog, "scanner.hash_failed") == [
+        {"error_type": "FileNotFoundError"}
+    ]
+    assert progress.hash_failed == 2
+
+
+def test_hash_failure_first_occurrence_resets_with_new_scan(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    path = tmp_path / "model.safetensors"
+    path.write_bytes(b"model")
+
+    def fail_hash(_path: str):
+        raise OSError("hash unavailable")
+
+    monkeypatch.setattr(scanner, "snapshot_hash", fail_hash)
+    session = hash_session(path)
+
+    with caplog.at_level(logging.INFO):
+        run_hash_failure(session, path, _ScanState())
+        run_hash_failure(session, path, _ScanState())
+
+    assert events_named(caplog, "scanner.hash_failed") == [
+        {"error_type": "OSError"},
+        {"error_type": "OSError"},
+    ]
+
+
+def test_hash_failure_no_log_line_leaks_exception_path(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # Both the tagged [assets-event] line and the privacy-safe
+    # "Asset scan error" line must omit the file path.
+    path = tmp_path / "private-model.safetensors"
+    path.write_bytes(b"model")
+
+    def fail_hash(_path: str):
+        raise FileNotFoundError(str(path))
+
+    monkeypatch.setattr(scanner, "snapshot_hash", fail_hash)
+
+    with caplog.at_level(logging.INFO):
+        run_hash_failure(hash_session(path), path, _ScanState())
+
+    assert events_named(caplog, "scanner.hash_failed") == [
+        {"error_type": "FileNotFoundError"}
+    ]
+    assert any("Asset scan error" in record.getMessage() for record in caplog.records)
+    assert all(str(path) not in record.getMessage() for record in caplog.records)
+    assert all(str(path) not in line for line in tagged_lines(caplog))
+
+
+def test_modified_during_hash_emits_fieldless_discard_event(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    path = tmp_path / "changing.safetensors"
+    path.write_bytes(b"model")
+    monkeypatch.setattr(scanner, "snapshot_hash", lambda _path: None)
+
+    with caplog.at_level(logging.INFO):
+        updated = scanner.enrich_asset(
+            hash_session(path),
+            file_path=str(path),
+            content_id="content",
+            record_id="record",
+            extract_metadata=False,
+            compute_hash=True,
+            progress=_ScanState(),
+        )
+
+    assert updated is False
+    assert events_named(caplog, "scanner.hash_discarded_modified") == [{}]
+
+
+def test_modified_hash_event_emits_once_per_scan_and_resets(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    path = tmp_path / "changing.safetensors"
+    path.write_bytes(b"model")
+    monkeypatch.setattr(scanner, "snapshot_hash", lambda _path: None)
+    progress = _ScanState()
+    session = hash_session(path)
+
+    with caplog.at_level(logging.INFO):
+        assert run_hash_failure(session, path, progress) is False
+        assert run_hash_failure(session, path, progress) is False
+        assert events_named(caplog, "scanner.hash_discarded_modified") == [{}]
+        assert run_hash_failure(session, path, _ScanState()) is False
+
+    assert events_named(caplog, "scanner.hash_discarded_modified") == [{}, {}]
+    assert progress.hash_failed == 0
+
+
+def test_locked_files_during_discovery_emit_stat_failed_exactly_once(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    def deny_stat(*_args, **_kwargs):
+        raise PermissionError("/private/assets/secret.bin")
+
+    monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=deny_stat, path=scanner.os.path))
+    progress = _ScanState()
+    paths = [f"/private/assets/locked-{i}.bin" for i in range(3)]
+
+    with caplog.at_level(logging.INFO):
+        specs, _tag_pool, _skipped = scanner.build_asset_specs(paths, set(), progress=progress)
+
+    assert specs == []
+    assert events_named(caplog, "scanner.stat_failed") == [
+        {"error_type": "PermissionError", "site": "discovery"}
+    ]
+    assert progress.permission_denied == 3
+
+
+def test_missing_files_during_discovery_emit_nothing(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    def vanish_stat(*_args, **_kwargs):
+        raise FileNotFoundError("/private/assets/gone.bin")
+
+    monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=vanish_stat, path=scanner.os.path))
+    progress = _ScanState()
+
+    with caplog.at_level(logging.INFO):
+        specs, _tag_pool, _skipped = scanner.build_asset_specs(
+            ["/private/assets/gone.bin"], set(), progress=progress
+        )
+
+    assert specs == []
+    assert events_named(caplog, "scanner.stat_failed") == []
+
+
+def test_discovery_stat_failure_emits_nothing_when_progress_is_none(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    def deny_stat(*_args, **_kwargs):
+        raise PermissionError("/private/assets/secret.bin")
+
+    monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=deny_stat, path=scanner.os.path))
+
+    with caplog.at_level(logging.INFO):
+        scanner.build_asset_specs(["/private/assets/locked.bin"], set(), progress=None)
+
+    assert events_named(caplog, "scanner.stat_failed") == []
+
+
+def test_locked_files_during_enrichment_emit_stat_failed_exactly_once(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    def deny_stat(*_args, **_kwargs):
+        raise PermissionError("/private/assets/secret.bin")
+
+    monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=deny_stat, path=scanner.os.path))
+    progress = _ScanState()
+
+    with caplog.at_level(logging.INFO):
+        first = scanner.enrich_asset(
+            Mock(),
+            file_path="/private/assets/locked-1.bin",
+            content_id="content-1",
+            record_id="record-1",
+            extract_metadata=False,
+            compute_hash=False,
+            progress=progress,
+        )
+        second = scanner.enrich_asset(
+            Mock(),
+            file_path="/private/assets/locked-2.bin",
+            content_id="content-2",
+            record_id="record-2",
+            extract_metadata=False,
+            compute_hash=False,
+            progress=progress,
+        )
+
+    assert first is False
+    assert second is False
+    assert events_named(caplog, "scanner.stat_failed") == [
+        {"error_type": "PermissionError", "site": "enrich"}
+    ]
+    assert progress.permission_denied == 2
+
+
+def test_missing_file_during_enrichment_returns_false_without_emitting(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    def vanish_stat(*_args, **_kwargs):
+        raise FileNotFoundError("/private/assets/gone.bin")
+
+    monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=vanish_stat, path=scanner.os.path))
+    progress = _ScanState()
+
+    with caplog.at_level(logging.INFO):
+        updated = scanner.enrich_asset(
+            Mock(),
+            file_path="/private/assets/gone.bin",
+            content_id="content",
+            record_id="record",
+            extract_metadata=False,
+            compute_hash=False,
+            progress=progress,
+        )
+
+    assert updated is False
+    assert events_named(caplog, "scanner.stat_failed") == []
+
+
+def test_enrich_failures_emit_once_per_scan_and_reset_with_new_scan(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    rows = [
+        UnenrichedContent("content-1", "record-1", "/private/assets/one.bin"),
+        UnenrichedContent("content-2", "record-2", "/private/assets/two.bin"),
+    ]
+    monkeypatch.setattr(scanner, "create_session", lambda: nullcontext(Mock()))
+
+    def fail_enrich(*_args, **_kwargs):
+        raise FileNotFoundError("/private/assets/secret.bin")
+
+    monkeypatch.setattr(scanner, "enrich_asset", fail_enrich)
+
+    with caplog.at_level(logging.INFO):
+        first_result = scanner.enrich_assets_batch(rows, progress=_ScanState())
+        second_result = scanner.enrich_assets_batch(rows[:1], progress=_ScanState())
+
+    assert first_result == (0, ["record-1", "record-2"])
+    assert second_result == (0, ["record-1"])
+    assert events_named(caplog, "scanner.enrich_failed") == [
+        {"error_type": "FileNotFoundError"},
+        {"error_type": "FileNotFoundError"},
+    ]
+
+
+def test_enrich_exception_counts_one_failure_per_raising_row(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    rows = [
+        UnenrichedContent("content-1", "record-1", "/private/assets/one.bin"),
+        UnenrichedContent("content-2", "record-2", "/private/assets/two.bin"),
+    ]
+    monkeypatch.setattr(scanner, "create_session", lambda: nullcontext(Mock()))
+
+    def fail_enrich(*_args, **_kwargs):
+        raise FileNotFoundError("/private/assets/secret.bin")
+
+    monkeypatch.setattr(scanner, "enrich_asset", fail_enrich)
+    progress = _ScanState()
+
+    enriched, failed_ids = scanner.enrich_assets_batch(rows, progress=progress)
+
+    assert enriched == 0
+    assert failed_ids == ["record-1", "record-2"]
+    assert progress.enrich_failed == 2
+
+
+def test_benign_enrich_no_op_is_skipped_without_counting_a_failure(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    deleted = tmp_path / "gone.safetensors"
+    rows = [UnenrichedContent("content-1", "record-1", str(deleted))]
+    monkeypatch.setattr(scanner, "create_session", lambda: nullcontext(Mock()))
+    progress = _ScanState()
+
+    enriched, failed_ids = scanner.enrich_assets_batch(rows, progress=progress)
+
+    assert enriched == 0
+    assert failed_ids == ["record-1"]
+    assert progress.enrich_failed == 0
diff --git a/tests-unit/seeder_test/test_seeder.py b/tests-unit/seeder_test/test_seeder.py
index c8544b29e..2637c74d9 100644
--- a/tests-unit/seeder_test/test_seeder.py
+++ b/tests-unit/seeder_test/test_seeder.py
@@ -1,8 +1,70 @@
+import logging
+import re
+import threading
+from contextlib import nullcontext
+from unittest.mock import Mock
+
+import pytest
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
 
+from app.assets import seeder as seeder_module
 from app.assets.database.models import Base
 from app.assets.database.queries import create_content, create_record, mark_content_missing
+from app.assets.event_log import TAG
+from app.assets.seeder import ScanPhase, State, _AssetSeeder, _ScanStage, _ScanState
+
+
+EVENT_LINE_PATTERN = re.compile(
+    rf"^{re.escape(TAG)} (?P<event>[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)*)"
+    r"(?P<fields>(?: [a-z_]+=[^ =]+)*)$"
+)
+EventFields = dict[str, bool | int | str]
+
+# Hang detector: a checkpoint that parks the scan fails the test, not the suite.
+SCAN_JOIN_TIMEOUT = 5.0
+
+
+@pytest.fixture
+def scan_seeder(monkeypatch: pytest.MonkeyPatch) -> _AssetSeeder:
+    instance = _AssetSeeder()
+    instance._state = State.RUNNING
+    instance._scan_state = _ScanState()
+    instance._roots = ("models", "input")
+    instance._phase = ScanPhase.FULL
+    monkeypatch.setattr(seeder_module, "dependencies_available", lambda: True)
+    monkeypatch.setattr(instance, "_log_scan_config", lambda roots: None)
+    return instance
+
+
+def parse_fields(raw: str) -> EventFields:
+    fields: EventFields = {}
+    for pair in raw.split():
+        name, value = pair.split("=", maxsplit=1)
+        if value == "true":
+            fields[name] = True
+        elif value == "false":
+            fields[name] = False
+        elif value.removeprefix("-").isdigit():
+            fields[name] = int(value)
+        else:
+            fields[name] = value
+    return fields
+
+
+def tagged_events(caplog: pytest.LogCaptureFixture) -> list[tuple[str, EventFields]]:
+    events: list[tuple[str, EventFields]] = []
+    for record in caplog.records:
+        match = EVENT_LINE_PATTERN.match(record.getMessage())
+        if match is not None:
+            events.append((match.group("event"), parse_fields(match.group("fields"))))
+    return events
+
+
+def events_named(
+    caplog: pytest.LogCaptureFixture, event_name: str
+) -> list[EventFields]:
+    return [fields for event, fields in tagged_events(caplog) if event == event_name]
 
 
 def test_seeder_models_missing_as_content_state():
@@ -22,3 +84,376 @@ def test_seeder_models_missing_as_content_state():
 
         assert content.is_missing is True
         assert record.content_id == content.id
+
+
+def test_multi_root_scan_emits_one_started_and_completed_without_root(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    clock = iter((10.0, 10.8126))
+    monkeypatch.setattr(seeder_module.time, "perf_counter", lambda: next(clock))
+    monkeypatch.setattr(scan_seeder, "_run_fast_phase", lambda roots: (3, 2, 5))
+    monkeypatch.setattr(scan_seeder, "_run_enrich_phase", lambda roots: (False, 4))
+
+    with caplog.at_level(logging.INFO):
+        scan_seeder._run_scan()
+
+    assert events_named(caplog, "seeder.scan_started") == [{"phase": "full"}]
+    completed = events_named(caplog, "seeder.scan_completed")
+    assert len(completed) == 1
+    assert completed[0] == {
+        "created": 3,
+        "elapsed_ms": 813,
+        "enrich_failed": 0,
+        "enriched": 4,
+        "hash_failed": 0,
+        "permission_denied": 0,
+        "phase": "full",
+        "skipped": 2,
+    }
+
+
+def test_scan_completed_reports_per_scan_failure_counts(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    scan_seeder._scan_state = _ScanState(
+        hash_failed=2,
+        enrich_failed=3,
+        permission_denied=1,
+    )
+    clock = iter((10.0, 10.5))
+    monkeypatch.setattr(seeder_module.time, "perf_counter", lambda: next(clock))
+    monkeypatch.setattr(scan_seeder, "_run_fast_phase", lambda roots: (0, 0, 0))
+    monkeypatch.setattr(scan_seeder, "_run_enrich_phase", lambda roots: (False, 0))
+
+    with caplog.at_level(logging.INFO):
+        scan_seeder._run_scan()
+
+    completed = events_named(caplog, "seeder.scan_completed")
+    assert len(completed) == 1
+    assert completed[0]["hash_failed"] == 2
+    assert completed[0]["enrich_failed"] == 3
+    assert completed[0]["permission_denied"] == 1
+
+
+def test_enrich_phase_does_not_count_returned_ids_as_failures(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    session = Mock()
+    batches = iter(
+        (
+            [
+                Mock(record_id="record-1"),
+                Mock(record_id="record-2"),
+            ],
+            [],
+        )
+    )
+    monkeypatch.setattr(seeder_module, "create_session", lambda: nullcontext(session))
+    monkeypatch.setattr(seeder_module, "drain_pending_verifications", lambda _session: None)
+    monkeypatch.setattr(seeder_module, "tick_watch_list", lambda _session: None)
+    monkeypatch.setattr(seeder_module, "drain_transition_queue", lambda _session: None)
+    monkeypatch.setattr(
+        seeder_module,
+        "get_unenriched_assets_for_roots",
+        lambda *_args, **_kwargs: next(batches),
+    )
+    monkeypatch.setattr(
+        seeder_module,
+        "enrich_assets_batch",
+        lambda *_args, **_kwargs: (0, ["record-1", "record-2"]),
+    )
+    monkeypatch.setattr(scan_seeder, "_check_pause_and_cancel", lambda _stage: False)
+
+    cancelled, enriched = scan_seeder._run_enrich_phase(("models",))
+
+    assert cancelled is False
+    assert enriched == 0
+    assert scan_seeder._scan_state is not None
+    assert scan_seeder._scan_state.enrich_failed == 0
+
+
+def test_starting_a_scan_installs_fresh_per_scan_failure_state(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    instance = _AssetSeeder()
+    instance._scan_state = _ScanState(
+        hash_failed=7,
+        enrich_failed=4,
+        permission_denied=2,
+    )
+    instance._scan_state.mark_emitted("enrich_failed")
+    monkeypatch.setattr(instance, "_run_scan", lambda: None)
+
+    started = instance.start(roots=("models",), phase=ScanPhase.FAST)
+
+    assert started is True
+    assert instance._thread is not None
+    instance._thread.join(timeout=5)
+    assert instance._scan_state == _ScanState()
+
+
+def test_single_root_scan_emits_root_and_phase(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    scan_seeder._roots = ("output",)
+    scan_seeder._phase = ScanPhase.FAST
+    monkeypatch.setattr(scan_seeder, "_run_fast_phase", lambda roots: (0, 0, 0))
+
+    with caplog.at_level(logging.INFO):
+        scan_seeder._run_scan()
+
+    assert events_named(caplog, "seeder.scan_started") == [
+        {"phase": "fast", "root": "output"}
+    ]
+    completed = events_named(caplog, "seeder.scan_completed")
+    assert len(completed) == 1
+    assert completed[0]["phase"] == "fast"
+    assert completed[0]["root"] == "output"
+
+
+def test_dependency_failure_emits_no_tagged_scan_lifecycle_lines(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    monkeypatch.setattr(seeder_module, "dependencies_available", lambda: False)
+
+    with caplog.at_level(logging.INFO):
+        scan_seeder._run_scan()
+
+    assert [
+        event for event, _fields in tagged_events(caplog) if event.startswith("seeder.scan_")
+    ] == []
+
+
+def test_scan_failure_emits_exception_type_without_message(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    scan_seeder._roots = ("models",)
+    scan_seeder._phase = ScanPhase.ENRICH
+
+    def fail_scan(_roots: tuple[str, ...]) -> None:
+        raise FileNotFoundError("/private/models/secret.safetensors")
+
+    monkeypatch.setattr(scan_seeder, "_log_scan_config", fail_scan)
+
+    with caplog.at_level(logging.INFO):
+        scan_seeder._run_scan()
+
+    assert events_named(caplog, "seeder.scan_failed") == [
+        {"error_type": "FileNotFoundError", "phase": "enrich", "root": "models"}
+    ]
+    tagged = "\n".join(record.getMessage() for record in caplog.records if TAG in record.getMessage())
+    assert "/private/models/secret.safetensors" not in tagged
+
+
+@pytest.mark.parametrize(
+    ("stage", "phase"),
+    [
+        pytest.param("pruning", ScanPhase.FAST, id="pruning"),
+        pytest.param("fast_scan", ScanPhase.FAST, id="fast-scan"),
+        pytest.param("enrich", ScanPhase.ENRICH, id="enrich"),
+        pytest.param("finalize", ScanPhase.ENRICH, id="finalize"),
+    ],
+)
+def test_scan_cancellation_emits_the_checkpoint_stage(
+    stage: str,
+    phase: ScanPhase,
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    scan_seeder._phase = phase
+    original_check = scan_seeder._check_pause_and_cancel
+
+    def cancel_at_stage(checkpoint_stage) -> bool:
+        if checkpoint_stage.value == stage:
+            scan_seeder._cancel_event.set()
+        return original_check(checkpoint_stage)
+
+    def run_enrich(roots) -> tuple[bool, int]:
+        # The finalize checkpoint is non-blocking and never routes through
+        # _check_pause_and_cancel, so its cancel has to land before it.
+        if stage == "finalize":
+            scan_seeder._cancel_event.set()
+        return (False, 0)
+
+    monkeypatch.setattr(scan_seeder, "_check_pause_and_cancel", cancel_at_stage)
+    monkeypatch.setattr(scan_seeder, "_run_fast_phase", lambda roots: (0, 0, 0))
+    monkeypatch.setattr(scan_seeder, "_run_enrich_phase", run_enrich)
+
+    with caplog.at_level(logging.INFO):
+        scan_seeder._run_scan()
+
+    assert events_named(caplog, "seeder.scan_cancelled") == [
+        {"phase": phase.value, "stage": stage}
+    ]
+
+
+def test_idle_reset_survives_a_raising_cancellation_emit(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    original_check = scan_seeder._check_pause_and_cancel
+
+    def cancel_at_pruning(stage) -> bool:
+        if stage == _ScanStage.PRUNING:
+            scan_seeder._cancel_event.set()
+        return original_check(stage)
+
+    monkeypatch.setattr(scan_seeder, "_check_pause_and_cancel", cancel_at_pruning)
+    monkeypatch.setattr(seeder_module, "get_owned_prefixes", lambda: ())
+    monkeypatch.setattr(
+        seeder_module, "mark_missing_outside_prefixes_safely", lambda prefixes: 0
+    )
+
+    original_emit = seeder_module.emit
+
+    def raise_on_scan_cancelled(event, **kwargs):
+        if event == "seeder.scan_cancelled":
+            raise RuntimeError("event bus down")
+        return original_emit(event, **kwargs)
+
+    monkeypatch.setattr(seeder_module, "emit", raise_on_scan_cancelled)
+
+    with pytest.raises(RuntimeError, match="event bus down"):
+        scan_seeder._run_scan()
+
+    assert scan_seeder._state is State.IDLE
+    assert scan_seeder._scan_state is None
+    assert scan_seeder.mark_missing_outside_prefixes() == 0
+
+
+def test_scan_paused_after_its_last_phase_still_completes(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    scan_seeder._phase = ScanPhase.ENRICH
+    monkeypatch.setattr(seeder_module, "get_owned_prefixes", lambda: ())
+    monkeypatch.setattr(
+        seeder_module, "mark_missing_outside_prefixes_safely", lambda prefixes: 0
+    )
+
+    def pause_while_finishing(roots) -> tuple[bool, int]:
+        scan_seeder.pause()
+        return (False, 0)
+
+    monkeypatch.setattr(scan_seeder, "_run_enrich_phase", pause_while_finishing)
+    events: list[str] = []
+    scan_seeder.set_event_sink(lambda event_type, data: events.append(event_type))
+
+    scan = threading.Thread(target=scan_seeder._run_scan, daemon=True)
+    scan.start()
+    try:
+        scan.join(timeout=SCAN_JOIN_TIMEOUT)
+
+        assert scan.is_alive() is False, "paused scan parked at the finalize checkpoint"
+        assert "assets.seed.completed" in events
+        assert "assets.seed.paused" not in events
+        assert scan_seeder.mark_missing_outside_prefixes() == 0
+    finally:
+        scan_seeder._run_gate.set()
+        scan.join(timeout=SCAN_JOIN_TIMEOUT)
+
+
+def test_enrich_interrupt_records_the_enrich_cancellation_stage(
+    scan_seeder: _AssetSeeder,
+) -> None:
+    scan_seeder._cancel_event.set()
+
+    assert scan_seeder._is_paused_or_cancelled() is True
+    assert scan_seeder._scan_state is not None
+    assert scan_seeder._scan_state.cancel_stage is not None
+    assert scan_seeder._scan_state.cancel_stage == "enrich"
+
+
+def test_prune_before_scan_emits_marked_missing_with_pruning_stage(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    scan_seeder._prune_first = True
+    scan_seeder._phase = ScanPhase.FAST
+    monkeypatch.setattr(seeder_module, "get_owned_prefixes", lambda: ())
+    monkeypatch.setattr(
+        seeder_module, "mark_missing_outside_prefixes_safely", lambda prefixes: 5
+    )
+    monkeypatch.setattr(
+        seeder_module, "sync_temp_references_safely", lambda _progress: None
+    )
+    monkeypatch.setattr(scan_seeder, "_run_fast_phase", lambda roots: (0, 0, 0))
+
+    with caplog.at_level(logging.INFO):
+        scan_seeder._run_scan()
+
+    assert events_named(caplog, "seeder.marked_missing") == [
+        {"count": 5, "stage": "pruning"}
+    ]
+
+
+def test_standalone_mark_missing_emits_count_with_mark_missing_stage(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    scan_seeder._state = State.IDLE
+    monkeypatch.setattr(seeder_module, "get_owned_prefixes", lambda: ())
+    monkeypatch.setattr(
+        seeder_module, "mark_missing_outside_prefixes_safely", lambda prefixes: 7
+    )
+
+    with caplog.at_level(logging.INFO):
+        result = scan_seeder.mark_missing_outside_prefixes()
+
+    assert result == 7
+    assert events_named(caplog, "seeder.marked_missing") == [
+        {"count": 7, "stage": "mark_missing"}
+    ]
+
+
+def test_batch_insert_failure_emits_only_the_exception_type(
+    scan_seeder: _AssetSeeder,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    session = Mock()
+    monkeypatch.setattr(
+        seeder_module, "sync_root_safely", lambda _root, _progress: set()
+    )
+    monkeypatch.setattr(
+        seeder_module, "collect_paths_for_roots", lambda roots: ["asset.safetensors"]
+    )
+    monkeypatch.setattr(
+        seeder_module,
+        "build_asset_specs",
+        lambda paths, existing_paths, enable_metadata_extraction, progress=None: (
+            [{"tags": []}],
+            {},
+            0,
+        ),
+    )
+
+    def fail_insert(batch, batch_tags) -> int:
+        raise PermissionError("/private/models/asset.safetensors")
+
+    monkeypatch.setattr(seeder_module, "insert_asset_specs", fail_insert)
+    monkeypatch.setattr(seeder_module, "create_session", lambda: nullcontext(session))
+    monkeypatch.setattr(seeder_module, "tick_watch_list", lambda current_session: None)
+
+    with caplog.at_level(logging.INFO):
+        scan_seeder._run_fast_phase(("models",))
+
+    assert events_named(caplog, "seeder.batch_insert_failed") == [
+        {"error_type": "PermissionError"}
+    ]
+    tagged = "\n".join(record.getMessage() for record in caplog.records if TAG in record.getMessage())
+    assert "/private/models/asset.safetensors" not in tagged
diff --git a/tests-unit/test_assets_event_log_static.py b/tests-unit/test_assets_event_log_static.py
new file mode 100644
index 000000000..cbab691de
--- /dev/null
+++ b/tests-unit/test_assets_event_log_static.py
@@ -0,0 +1,260 @@
+"""Static discipline check for the ``[assets-event]`` log lines.
+
+Pure :mod:`ast` analysis: no module under ``app/assets`` is imported or
+executed, so this check never needs a running ComfyUI. It deliberately lives at
+the ``tests-unit`` root rather than under ``tests-unit/assets_test/``, whose
+autouse fixture boots a ComfyUI subprocess for every test in that subtree.
+
+Four rules are enforced over every emit call site:
+
+a. keyword fields come from the closed vocabulary, the event is a string literal
+b. no other log line anywhere carries the tag, so the tap only ever sees emits
+c. the call sites present in the tree match an explicit manifest
+d. ``error_type=`` values come from ``event_log.error_type()``, never a string
+"""
+
+from __future__ import annotations
+
+import ast
+from collections import Counter
+from collections.abc import Iterator
+from pathlib import Path
+from typing import NamedTuple
+
+from app.assets.event_log import ALLOWED_EVENTS, ALLOWED_FIELDS, TAG
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+MODULE_SCOPE = "<module>"
+EVENT_LOG_NAME = "event_log"
+LOG_METHODS = frozenset({"debug", "info", "warning", "error", "exception", "critical", "log"})
+FUNCTION_NODES = (ast.FunctionDef, ast.AsyncFunctionDef)
+
+
+class CallSite(NamedTuple):
+    """The identity of one emit call: file, enclosing function, event."""
+
+    path: str
+    function: str
+    event: str
+
+
+# The manifest of every tagged event this branch emits: (file, enclosing
+# function, event) triples that must be present in the tree exactly as written.
+EXPECTED_CALL_SITES: frozenset[CallSite] = frozenset(
+    {
+        # todo 10 - seeder lifecycle + the single assets.enabled site
+        CallSite("server.py", "__init__", "assets.enabled"),
+        CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_started"),
+        CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_completed"),
+        CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_failed"),
+        CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_cancelled"),
+        CallSite("app/assets/seeder.py", "_run_scan", "seeder.marked_missing"),
+        CallSite("app/assets/seeder.py", "mark_missing_outside_prefixes", "seeder.marked_missing"),
+        CallSite("app/assets/seeder.py", "_run_fast_phase", "seeder.batch_insert_failed"),
+        # todo 11 - scanner failure paths
+        CallSite("app/assets/scanner.py", "sync_root_safely", "scanner.fast_scan_failed"),
+        CallSite("app/assets/scanner.py", "sync_temp_references_safely", "scanner.temp_sync_failed"),
+        CallSite(
+            "app/assets/scanner.py", "mark_missing_outside_prefixes_safely", "scanner.mark_missing_failed"
+        ),
+        CallSite("app/assets/scanner.py", "enrich_asset", "scanner.hash_failed"),
+        CallSite("app/assets/scanner.py", "enrich_asset", "scanner.hash_discarded_modified"),
+        CallSite("app/assets/scanner.py", "enrich_assets_batch", "scanner.enrich_failed"),
+        # todo 16 - discovery/enrich stat failures, emit-once per scan per site
+        CallSite("app/assets/scanner.py", "build_asset_specs", "scanner.stat_failed"),
+        CallSite("app/assets/scanner.py", "enrich_asset", "scanner.stat_failed"),
+    }
+)
+
+class Aliases(NamedTuple):
+    """The names one module binds to the event_log module and its functions."""
+
+    module: frozenset[str]
+    emit: frozenset[str]
+    error_type: frozenset[str]
+
+
+class Scan(NamedTuple):
+    """Everything the AST walk learned about the tree."""
+
+    files: tuple[str, ...]
+    call_sites: Counter[CallSite]
+    vocabulary: tuple[str, ...]
+    event_names: tuple[str, ...]
+    error_types: tuple[str, ...]
+    tagged_logs: tuple[str, ...]
+
+
+def _scanned_files(root: Path) -> tuple[str, ...]:
+    """Every assets module, plus server.py for its single assets.enabled emit."""
+    assets = sorted(p.relative_to(root).as_posix() for p in root.glob("app/assets/**/*.py"))
+    return (*assets, "server.py")
+
+
+def _scoped_nodes(tree: ast.Module) -> Iterator[tuple[ast.AST, str]]:
+    """Yield every node paired with the name of its innermost enclosing function."""
+
+    def walk(node: ast.AST, scope: str) -> Iterator[tuple[ast.AST, str]]:
+        for child in ast.iter_child_nodes(node):
+            child_scope = child.name if isinstance(child, FUNCTION_NODES) else scope
+            yield child, child_scope
+            yield from walk(child, child_scope)
+
+    yield from walk(tree, MODULE_SCOPE)
+
+
+def _resolve_aliases(tree: ast.Module) -> Aliases:
+    module: set[str] = set()
+    emit: set[str] = set()
+    error_type: set[str] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                if alias.name.split(".")[-1] == EVENT_LOG_NAME:
+                    module.add(alias.asname or alias.name)
+        elif isinstance(node, ast.ImportFrom):
+            from_event_log = (node.module or "").rsplit(".", 1)[-1] == EVENT_LOG_NAME
+            for alias in node.names:
+                if from_event_log and alias.name == "emit":
+                    emit.add(alias.asname or alias.name)
+                elif from_event_log and alias.name == "error_type":
+                    error_type.add(alias.asname or alias.name)
+                elif not from_event_log and alias.name == EVENT_LOG_NAME:
+                    module.add(alias.asname or alias.name)
+    return Aliases(frozenset(module), frozenset(emit), frozenset(error_type))
+
+
+def _is_emit_call(func: ast.expr, aliases: Aliases) -> bool:
+    if isinstance(func, ast.Attribute) and func.attr == "emit" and isinstance(func.value, ast.Name):
+        return func.value.id in aliases.module
+    return isinstance(func, ast.Name) and func.id in aliases.emit
+
+
+def _is_error_type_call(value: ast.expr, aliases: Aliases) -> bool:
+    """True only for a call to the sanctioned event_log.error_type()."""
+    if not isinstance(value, ast.Call):
+        return False
+    func = value.func
+    if isinstance(func, ast.Attribute) and func.attr == "error_type" and isinstance(func.value, ast.Name):
+        return func.value.id in aliases.module
+    return isinstance(func, ast.Name) and func.id in aliases.error_type
+
+
+def _carries_tag(node: ast.Call) -> bool:
+    return any(
+        isinstance(child, ast.Constant) and isinstance(child.value, str) and TAG in child.value
+        for child in ast.walk(node)
+    )
+
+
+def _is_log_call(func: ast.expr) -> bool:
+    return isinstance(func, ast.Attribute) and func.attr in LOG_METHODS
+
+
+def _event_of(call: ast.Call) -> str | None:
+    """The literal event name, or None when it is not a plain string literal."""
+    if len(call.args) != 1:
+        return None
+    first = call.args[0]
+    if not isinstance(first, ast.Constant) or not isinstance(first.value, str):
+        return None
+    return first.value
+
+
+def _field_faults(call: ast.Call, aliases: Aliases) -> Iterator[tuple[str, str]]:
+    """(category, reason) for every keyword that breaks rule (a) or rule (d)."""
+    for keyword in call.keywords:
+        if keyword.arg is None:
+            yield "vocabulary", "**splat fields cannot be checked statically"
+        elif keyword.arg not in ALLOWED_FIELDS:
+            yield "vocabulary", f"field {keyword.arg!r} is not in ALLOWED_FIELDS"
+        elif keyword.arg == "error_type" and not _is_error_type_call(keyword.value, aliases):
+            yield (
+                "error_types",
+                "error_type= must be a call to event_log.error_type(), got "
+                f"{ast.unparse(keyword.value)!r}",
+            )
+
+
+def _file_faults(call: ast.Call, aliases: Aliases) -> Iterator[tuple[str, str]]:
+    if _is_emit_call(call.func, aliases):
+        event = _event_of(call)
+        if event not in ALLOWED_EVENTS:
+            yield "event_names", "the event must be one string literal in the allowed vocabulary"
+        yield from _field_faults(call, aliases)
+    elif _is_log_call(call.func) and _carries_tag(call):
+        yield "tagged_logs", f"log line carries {TAG} outside event_log.emit()"
+
+
+def _scan_file(root: Path, relative: str) -> tuple[Counter[CallSite], list[tuple[str, str]]]:
+    tree = ast.parse((root / relative).read_text(encoding="utf-8"), filename=relative)
+    aliases = _resolve_aliases(tree)
+    sites: Counter[CallSite] = Counter()
+    faults: list[tuple[str, str]] = []
+    for node, scope in _scoped_nodes(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        if _is_emit_call(node.func, aliases):
+            event = _event_of(node)
+            if event is not None and event in ALLOWED_EVENTS:
+                sites[CallSite(relative, scope, event)] += 1
+        for category, reason in _file_faults(node, aliases):
+            faults.append((category, f"{relative}:{node.lineno}: {reason}"))
+    return sites, faults
+
+
+def scan_repository(root: Path = REPO_ROOT) -> Scan:
+    files = _scanned_files(root)
+    sites: Counter[CallSite] = Counter()
+    found: dict[str, list[str]] = {"vocabulary": [], "event_names": [], "error_types": [], "tagged_logs": []}
+    for relative in files:
+        file_sites, faults = _scan_file(root, relative)
+        sites += file_sites
+        for category, message in faults:
+            found[category].append(message)
+    return Scan(
+        files=files,
+        call_sites=sites,
+        vocabulary=tuple(found["vocabulary"]),
+        event_names=tuple(found["event_names"]),
+        error_types=tuple(found["error_types"]),
+        tagged_logs=tuple(found["tagged_logs"]),
+    )
+
+
+SCAN = scan_repository()
+
+
+def test_the_walk_actually_covers_the_assets_tree() -> None:
+    """Guards every other check: a broken glob would make them all vacuous."""
+    assert "app/assets/event_log.py" in SCAN.files
+    assert "app/assets/seeder.py" in SCAN.files
+    assert "server.py" in SCAN.files
+    assert len(SCAN.files) > 20
+
+
+def test_emit_fields_stay_inside_the_closed_vocabulary() -> None:
+    assert SCAN.vocabulary == ()
+
+
+def test_emit_events_are_literals_in_the_allowed_vocabulary() -> None:
+    assert SCAN.event_names == ()
+
+
+def test_error_type_values_come_from_event_log_error_type() -> None:
+    assert SCAN.error_types == ()
+
+
+def test_no_other_log_line_carries_the_event_tag() -> None:
+    assert SCAN.tagged_logs == ()
+
+
+def test_call_sites_match_the_manifest() -> None:
+    manifest = Counter(EXPECTED_CALL_SITES)
+    unexpected = SCAN.call_sites - manifest
+    missing = manifest - SCAN.call_sites
+    assert not unexpected, (
+        f"emit call sites not in the manifest: {sorted(unexpected)} — add them to "
+        "EXPECTED_CALL_SITES"
+    )
+    assert not missing, f"manifest call sites absent from the tree: {sorted(missing)}"