Rebuild stale asset records when the assets system is enabled

Alembic migrates the shape of the assets tables. Nothing migrated their
meaning, so a row could be structurally current and still hold values a
superseded rule computed, with no maintenance path that ever repaired it.

loader_path is the clearest case: the column was added to existing databases
with no backfill, and it is only ever written when a reference is first
created. The scan computes it for paths it has not seen before, so every
reference older than the column serves a null loader path forever -- while the
API tells clients to prefer loader_path over name.

A semantics version now records which generation of the derivation logic
produced a database's rows, tracked separately from the Alembic schema version
because the two move independently. Reset steps are numbered and applied in
order from the stored version, each stamped only once it finishes, so an
interrupted run resumes instead of half-applying.

The first step re-derives what a file's location implies -- loader_path, the
backend tags a path carries, and whether the file is there -- and leaves
everything else alone. It reads no file contents: verify_file_unchanged says
whether a row's recorded hash and size still describe the file, and a file that
has moved on is handed to the existing needs_verify path rather than re-read,
so an untouched model library costs one stat per file and no hashing. Manual
tags, user metadata, previews, deletions and job ids are never touched, nor are
references whose file is gone or whose path falls outside every root this
install currently knows about.

It runs at the start of a scan, before anything reads or extends those rows. A
database already at the current version costs one indexed row read.

alembic_db/versions/0007_add_asset_semantics_version.py

@@ -0,0 +1,34 @@
+"""
+Add asset_semantics_version table.
+
+Alembic records the *shape* of the assets tables. This table records the
+*meaning* of their contents: which generation of the derivation logic produced
+the values currently stored in them. The two move independently, so they are
+tracked independently.
+
+Revision ID: 0007_add_asset_semantics_version
+Revises: 0006_add_loader_path
+Create Date: 2026-08-18
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0007_add_asset_semantics_version"
+down_revision = "0006_add_loader_path"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "asset_semantics_version",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("version", sa.Integer(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=False), nullable=False),
+        sa.PrimaryKeyConstraint("id", name="pk_asset_semantics_version"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("asset_semantics_version")

app/assets/database/models.py

@@ -23,6 +23,28 @@ from app.assets.helpers import get_utc_now
 from app.database.models import Base
 
 
+class AssetSemanticsVersion(Base):
+    """Which generation of the derivation logic produced this database's rows.
+
+    Alembic tracks the *shape* of the assets tables; this tracks the *meaning*
+    of what is stored in them. A row can be structurally current and still hold
+    values a superseded rule computed, so the two versions move independently.
+    Reset steps in ``app.assets.semantics`` bring such rows forward and then
+    advance this stamp. Always a single row, keyed on id 1.
+    """
+
+    __tablename__ = "asset_semantics_version"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, default=1)
+    version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=False), nullable=False, default=get_utc_now
+    )
+
+    def __repr__(self) -> str:
+        return f"<AssetSemanticsVersion version={self.version}>"
+
+
 class Asset(Base):
     __tablename__ = "assets"
 

app/assets/database/queries/__init__.py

@@ -52,6 +52,17 @@ from app.assets.database.queries.asset_reference import (
     update_reference_updated_at,
     upsert_reference,
 )
+from app.assets.database.queries.semantics import (
+    AUTOMATIC_TAG_ORIGIN,
+    DerivedStateRow,
+    bulk_add_automatic_tags,
+    bulk_remove_automatic_tags,
+    bulk_set_loader_paths,
+    get_file_backed_references_page,
+    get_semantics_version,
+    get_tags_by_reference,
+    set_semantics_version,
+)
 from app.assets.database.queries.tags import (
     AddTagsResult,
     RemoveTagsResult,
@@ -71,7 +82,9 @@ from app.assets.database.queries.tags import (
 
 __all__ = [
     "AddTagsResult",
+    "AUTOMATIC_TAG_ORIGIN",
     "CacheStateRow",
+    "DerivedStateRow",
     "RemoveTagsResult",
     "SetTagsResult",
     "UnenrichedReferenceRow",
@@ -81,6 +94,9 @@ __all__ = [
     "bulk_insert_assets",
     "bulk_insert_references_ignore_conflicts",
     "bulk_insert_tags_and_meta",
+    "bulk_add_automatic_tags",
+    "bulk_remove_automatic_tags",
+    "bulk_set_loader_paths",
     "bulk_update_enrichment_level",
     "count_active_siblings",
     "create_stub_asset",
@@ -96,6 +112,7 @@ __all__ = [
     "fetch_reference_asset_and_tags",
     "get_asset_by_hash",
     "get_existing_asset_ids",
+    "get_file_backed_references_page",
     "get_or_create_reference",
     "get_reference_by_file_path",
     "get_reference_by_id",
@@ -104,6 +121,8 @@ __all__ = [
     "get_reference_tags",
     "get_references_by_paths_and_asset_ids",
     "get_references_for_prefixes",
+    "get_semantics_version",
+    "get_tags_by_reference",
     "get_unenriched_references",
     "get_unreferenced_unhashed_asset_ids",
     "insert_reference",
@@ -123,6 +142,7 @@ __all__ = [
     "set_reference_metadata",
     "set_reference_preview",
     "set_reference_system_metadata",
+    "set_semantics_version",
     "soft_delete_reference_by_id",
     "set_reference_tags",
     "update_asset_hash_and_mime",

app/assets/database/queries/semantics.py

@@ -0,0 +1,195 @@
+"""Queries backing the asset semantics reset (see ``app.assets.semantics``)."""
+
+from dataclasses import dataclass
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import sqlite
+from sqlalchemy.orm import Session
+
+from app.assets.database.models import (
+    Asset,
+    AssetReference,
+    AssetReferenceTag,
+    AssetSemanticsVersion,
+)
+from app.assets.database.queries.common import (
+    MAX_BIND_PARAMS,
+    iter_chunks,
+    iter_row_chunks,
+)
+from app.assets.database.queries.tags import ensure_tags_exist
+from app.assets.helpers import get_utc_now
+
+# The version table holds exactly one row; the id is a constant, not a sequence.
+_VERSION_ROW_ID = 1
+
+# Tags this subsystem derives from the filesystem rather than being told.
+AUTOMATIC_TAG_ORIGIN = "automatic"
+
+
+def get_semantics_version(session: Session) -> int:
+    """Return the stored semantics version, or 0 if none has been stamped."""
+    row = session.get(AssetSemanticsVersion, _VERSION_ROW_ID)
+    return int(row.version) if row is not None else 0
+
+
+def set_semantics_version(session: Session, version: int) -> None:
+    """Stamp the semantics version. Caller commits."""
+    row = session.get(AssetSemanticsVersion, _VERSION_ROW_ID)
+    if row is None:
+        session.add(
+            AssetSemanticsVersion(
+                id=_VERSION_ROW_ID, version=int(version), updated_at=get_utc_now()
+            )
+        )
+    else:
+        row.version = int(version)
+        row.updated_at = get_utc_now()
+    session.flush()
+
+
+@dataclass(frozen=True)
+class DerivedStateRow:
+    """A file-backed reference and the state a reprojection may rewrite."""
+
+    reference_id: str
+    file_path: str
+    loader_path: str | None
+    mtime_ns: int | None
+    is_missing: bool
+    needs_verify: bool
+    size_bytes: int | None
+
+
+def get_file_backed_references_page(
+    session: Session,
+    after_id: str | None,
+    limit: int,
+) -> list[DerivedStateRow]:
+    """Return up to ``limit`` file-backed references ordered by id, after ``after_id``.
+
+    Keyset pagination on the primary key so a reprojection can walk the whole
+    table in bounded transactions without holding one open across the walk.
+    Soft-deleted references are included: their derived columns are as stale as
+    anyone else's, and an undelete would serve them.
+    """
+    query = (
+        sa.select(
+            AssetReference.id,
+            AssetReference.file_path,
+            AssetReference.loader_path,
+            AssetReference.mtime_ns,
+            AssetReference.is_missing,
+            AssetReference.needs_verify,
+            Asset.size_bytes,
+        )
+        .join(Asset, Asset.id == AssetReference.asset_id)
+        .where(AssetReference.file_path.isnot(None))
+        .order_by(AssetReference.id.asc())
+        .limit(limit)
+    )
+    if after_id is not None:
+        query = query.where(AssetReference.id > after_id)
+
+    return [
+        DerivedStateRow(
+            reference_id=row[0],
+            file_path=row[1],
+            loader_path=row[2],
+            mtime_ns=row[3],
+            is_missing=bool(row[4]),
+            needs_verify=bool(row[5]),
+            size_bytes=int(row[6]) if row[6] is not None else None,
+        )
+        for row in session.execute(query).all()
+    ]
+
+
+def get_tags_by_reference(
+    session: Session, reference_ids: list[str]
+) -> dict[str, dict[str, str]]:
+    """Return {reference_id: {tag_name: origin}} for the given references."""
+    if not reference_ids:
+        return {}
+
+    by_reference: dict[str, dict[str, str]] = {}
+    for chunk in iter_chunks(reference_ids, MAX_BIND_PARAMS):
+        rows = session.execute(
+            sa.select(
+                AssetReferenceTag.asset_reference_id,
+                AssetReferenceTag.tag_name,
+                AssetReferenceTag.origin,
+            ).where(AssetReferenceTag.asset_reference_id.in_(chunk))
+        ).all()
+        for reference_id, tag_name, origin in rows:
+            by_reference.setdefault(reference_id, {})[tag_name] = origin
+    return by_reference
+
+
+def bulk_set_loader_paths(
+    session: Session, loader_paths: dict[str, str | None]
+) -> None:
+    """Set loader_path on the given references. Caller commits."""
+    if not loader_paths:
+        return
+    session.execute(
+        sa.update(AssetReference),
+        [
+            {"id": reference_id, "loader_path": loader_path}
+            for reference_id, loader_path in loader_paths.items()
+        ],
+    )
+
+
+def bulk_add_automatic_tags(session: Session, links: list[tuple[str, str]]) -> None:
+    """Attach (reference_id, tag_name) links with automatic origin. Caller commits."""
+    if not links:
+        return
+
+    now = get_utc_now()
+    ensure_tags_exist(session, {tag_name for _, tag_name in links})
+    rows = [
+        {
+            "asset_reference_id": reference_id,
+            "tag_name": tag_name,
+            "origin": AUTOMATIC_TAG_ORIGIN,
+            "added_at": now,
+        }
+        for reference_id, tag_name in links
+    ]
+    for chunk in iter_row_chunks(rows, cols_per_row=4):
+        session.execute(
+            sqlite.insert(AssetReferenceTag)
+            .values(chunk)
+            .on_conflict_do_nothing(
+                index_elements=[
+                    AssetReferenceTag.asset_reference_id,
+                    AssetReferenceTag.tag_name,
+                ]
+            )
+        )
+
+
+def bulk_remove_automatic_tags(session: Session, links: list[tuple[str, str]]) -> None:
+    """Detach (reference_id, tag_name) links, but only automatic ones.
+
+    The origin is re-asserted in the statement rather than trusted from the
+    caller's snapshot, so a manual tag can never be removed by this path.
+    Caller commits.
+    """
+    if not links:
+        return
+
+    by_reference: dict[str, list[str]] = {}
+    for reference_id, tag_name in links:
+        by_reference.setdefault(reference_id, []).append(tag_name)
+
+    for reference_id, tag_names in by_reference.items():
+        for chunk in iter_chunks(tag_names, MAX_BIND_PARAMS):
+            session.execute(
+                sa.delete(AssetReferenceTag).where(
+                    AssetReferenceTag.asset_reference_id == reference_id,
+                    AssetReferenceTag.tag_name.in_(chunk),
+                    AssetReferenceTag.origin == AUTOMATIC_TAG_ORIGIN,
+                )
+            )

app/assets/seeder.py

@@ -23,6 +23,7 @@ from app.assets.scanner import (
     sync_root_safely,
     sync_temp_references_safely,
 )
+from app.assets.semantics import run_pending_semantics_steps
 from app.database.db import dependencies_available
 
 
@@ -548,6 +549,11 @@ class _AssetSeeder:
                 )
                 return
 
+            # Stale rows are brought forward before anything reads or extends
+            # them. A database already at the current semantics version costs a
+            # single row read here.
+            run_pending_semantics_steps(interrupt_check=self._is_cancelled)
+
             if self._prune_first:
                 all_prefixes = get_owned_prefixes()
                 marked = mark_missing_outside_prefixes_safely(all_prefixes)

app/assets/semantics/__init__.py

@@ -0,0 +1,119 @@
+"""Reset steps that bring stale asset rows forward to the current semantics.
+
+Alembic migrates the *shape* of the assets tables. Nothing migrates their
+*meaning*: a row whose columns are structurally current can still hold values
+that a superseded rule computed, and the steady-state scan will not repair one,
+because it only ever looks at paths it has not seen before.
+
+A reset step closes that gap. Each step is numbered, applied in order from the
+version stamped in the database up to ``CURRENT_SEMANTICS_VERSION``, and stamped
+only once it finishes, so an interrupted run resumes rather than half-applying.
+Steps must be idempotent: re-running one on rows it has already fixed changes
+nothing.
+
+The registry, not any individual step, is the durable part. A step is free to be
+as broad or as surgical as its particular drift calls for.
+"""
+
+import logging
+import time
+
+from app.assets.database.queries.semantics import (
+    get_semantics_version,
+    set_semantics_version,
+)
+from app.assets.semantics.reproject_derived import reproject_derived_state
+from app.assets.semantics.step import (
+    InterruptCheck,
+    SemanticsStep,
+    SemanticsStepInterrupted,
+)
+from app.database.db import can_create_session, create_session
+
+__all__ = [
+    "CURRENT_SEMANTICS_VERSION",
+    "SEMANTICS_STEPS",
+    "InterruptCheck",
+    "SemanticsStep",
+    "SemanticsStepInterrupted",
+    "run_pending_semantics_steps",
+]
+
+SEMANTICS_STEPS: tuple[SemanticsStep, ...] = (
+    SemanticsStep(
+        version=1,
+        description="reproject path-derived reference state",
+        apply=reproject_derived_state,
+    ),
+)
+
+CURRENT_SEMANTICS_VERSION = SEMANTICS_STEPS[-1].version
+
+
+def run_pending_semantics_steps(interrupt_check: InterruptCheck | None = None) -> int:
+    """Apply every reset step this database has not been stamped for.
+
+    A database already at ``CURRENT_SEMANTICS_VERSION`` costs one indexed row
+    read and nothing else -- no filesystem is touched to discover there is
+    nothing to do, which is the overwhelmingly common case.
+
+    Returns the number of steps applied.
+    """
+    if not can_create_session():
+        return 0
+
+    try:
+        with create_session() as session:
+            stored_version = get_semantics_version(session)
+    except Exception:
+        logging.exception(
+            "Could not read the asset semantics version; skipping semantics reset"
+        )
+        return 0
+
+    pending = [step for step in SEMANTICS_STEPS if step.version > stored_version]
+    if not pending:
+        return 0
+
+    logging.info(
+        "Asset semantics at version %d, current is %d: applying %d step(s)",
+        stored_version,
+        CURRENT_SEMANTICS_VERSION,
+        len(pending),
+    )
+
+    applied = 0
+    for step in pending:
+        started = time.perf_counter()
+        try:
+            summary = step.apply(interrupt_check)
+        except SemanticsStepInterrupted:
+            logging.info(
+                "Asset semantics step %d (%s) interrupted; resuming on next start",
+                step.version,
+                step.description,
+            )
+            return applied
+        except Exception:
+            logging.exception(
+                "Asset semantics step %d (%s) failed; database stays at version %d",
+                step.version,
+                step.description,
+                stored_version + applied,
+            )
+            return applied
+
+        with create_session() as session:
+            set_semantics_version(session, step.version)
+            session.commit()
+
+        applied += 1
+        logging.info(
+            "Asset semantics step %d (%s) applied in %.3fs: %s",
+            step.version,
+            step.description,
+            time.perf_counter() - started,
+            summary,
+        )
+
+    return applied

app/assets/semantics/reproject_derived.py

@@ -0,0 +1,215 @@
+"""Semantics step 1: re-derive the reference state that comes from the path.
+
+Three columns are pure functions of a file's location and whether it is there:
+``loader_path``, the backend tags a path implies, and ``is_missing``. Every one
+of them was computed once, when the reference was first written, by whatever
+rules were current that day. ``loader_path`` in particular was added to existing
+databases as a NULL column and no path has ever filled it in, so references
+older than that column serve a null loader path forever.
+
+This step recomputes those three from the filesystem as it stands now. It reads
+no file contents: ``verify_file_unchanged`` answers whether a file still matches
+what the database recorded, and the only thing that answer is used for is
+deciding *not* to touch content-derived state. Nothing here re-hashes, so a
+library of untouched 500GB models costs one ``stat`` each.
+
+What the step will not touch:
+
+- ``hash`` and ``size_bytes`` -- content facts, unrecoverable without reading
+  the file. A file that has changed underneath its row is handed to the existing
+  ``needs_verify`` path instead, exactly as the scanner would flag it.
+- Anything a person chose: manual and upload-origin tags, ``user_metadata``,
+  ``preview_id``, ``deleted_at``, ``job_id``, ``name``.
+- References whose file is gone, or whose path is not under any root this
+  install currently knows about. Retiring the former is the scanner's job under
+  its own rules; the latter cannot be classified at all, and a misconfigured
+  ``extra_model_paths.yaml`` must not be able to strip tags off assets that are
+  merely out of view.
+"""
+
+import logging
+import os
+from dataclasses import dataclass
+
+from app.assets.database.queries import (
+    bulk_update_is_missing,
+    bulk_update_needs_verify,
+)
+from app.assets.database.queries.semantics import (
+    AUTOMATIC_TAG_ORIGIN,
+    DerivedStateRow,
+    bulk_add_automatic_tags,
+    bulk_remove_automatic_tags,
+    bulk_set_loader_paths,
+    get_file_backed_references_page,
+    get_tags_by_reference,
+)
+from app.assets.helpers import normalize_tags
+from app.assets.semantics.step import InterruptCheck, SemanticsStepInterrupted
+from app.assets.services.file_utils import verify_file_unchanged
+from app.assets.services.path_utils import (
+    compute_loader_path,
+    get_path_derived_tag_vocabulary,
+    get_path_derived_tags_from_path,
+)
+from app.database.db import create_session
+
+# Bounds each transaction. Small enough that an interrupt loses little work,
+# large enough that the walk is not dominated by session setup.
+_BATCH_SIZE = 500
+
+
+@dataclass
+class ReprojectionSummary:
+    """What a reprojection pass did, for the log line."""
+
+    scanned: int = 0
+    unchanged_files: int = 0
+    changed_files: int = 0
+    absent_files: int = 0
+    unclassified_paths: int = 0
+    loader_paths_rewritten: int = 0
+    tags_added: int = 0
+    tags_removed: int = 0
+    missing_flags_cleared: int = 0
+    verify_flags_set: int = 0
+
+    def __str__(self) -> str:
+        return (
+            f"scanned={self.scanned} unchanged={self.unchanged_files} "
+            f"changed={self.changed_files} absent={self.absent_files} "
+            f"unclassified={self.unclassified_paths} "
+            f"loader_paths={self.loader_paths_rewritten} "
+            f"tags+{self.tags_added}/-{self.tags_removed} "
+            f"unflagged_missing={self.missing_flags_cleared} "
+            f"flagged_verify={self.verify_flags_set}"
+        )
+
+
+def reproject_derived_state(
+    interrupt_check: InterruptCheck | None = None,
+) -> ReprojectionSummary:
+    """Re-derive path-derived state for every file-backed reference.
+
+    Walks the reference table in keyset-paginated batches, committing each one,
+    so a kill mid-walk leaves committed batches reprojected and the rest
+    untouched -- both states the step handles on its next run.
+    """
+    summary = ReprojectionSummary()
+    vocabulary = get_path_derived_tag_vocabulary()
+    after_id: str | None = None
+
+    while True:
+        if interrupt_check is not None and interrupt_check():
+            raise SemanticsStepInterrupted(
+                f"interrupted after {summary.scanned} references"
+            )
+
+        with create_session() as session:
+            rows = get_file_backed_references_page(
+                session, after_id=after_id, limit=_BATCH_SIZE
+            )
+            if not rows:
+                return summary
+            after_id = rows[-1].reference_id
+            _reproject_batch(session, rows, vocabulary, summary)
+            session.commit()
+
+
+def _reproject_batch(
+    session,
+    rows: list[DerivedStateRow],
+    vocabulary: set[str],
+    summary: ReprojectionSummary,
+) -> None:
+    stored_tags = get_tags_by_reference(session, [row.reference_id for row in rows])
+
+    loader_paths: dict[str, str | None] = {}
+    tags_to_add: list[tuple[str, str]] = []
+    tags_to_remove: list[tuple[str, str]] = []
+    clear_missing: list[str] = []
+    set_needs_verify: list[str] = []
+
+    for row in rows:
+        summary.scanned += 1
+        present, unchanged = _classify_file(row)
+
+        if not present:
+            # The scanner owns retiring these, under its own missing semantics.
+            summary.absent_files += 1
+            continue
+
+        if unchanged:
+            summary.unchanged_files += 1
+        else:
+            # The file moved on from what the row records, so its hash, size and
+            # mime no longer describe it. Re-deriving those means reading the
+            # file, which this step does not do; flag it for the path that does.
+            summary.changed_files += 1
+            if not row.needs_verify:
+                set_needs_verify.append(row.reference_id)
+
+        try:
+            derived_tags = set(
+                normalize_tags(get_path_derived_tags_from_path(row.file_path))
+            )
+        except ValueError:
+            summary.unclassified_paths += 1
+            continue
+
+        loader_path = compute_loader_path(row.file_path)
+        if loader_path != row.loader_path:
+            loader_paths[row.reference_id] = loader_path
+            summary.loader_paths_rewritten += 1
+
+        current_tags = stored_tags.get(row.reference_id, {})
+        for tag_name in sorted(derived_tags - set(current_tags)):
+            tags_to_add.append((row.reference_id, tag_name))
+        for tag_name, origin in sorted(current_tags.items()):
+            if (
+                origin == AUTOMATIC_TAG_ORIGIN
+                and tag_name in vocabulary
+                and tag_name not in derived_tags
+            ):
+                tags_to_remove.append((row.reference_id, tag_name))
+
+        if row.is_missing:
+            # The file is right there. Nothing else un-flags this: the scanner
+            # excludes already-missing references from its temp reconciliation.
+            clear_missing.append(row.reference_id)
+
+    bulk_set_loader_paths(session, loader_paths)
+    bulk_add_automatic_tags(session, tags_to_add)
+    bulk_remove_automatic_tags(session, tags_to_remove)
+    bulk_update_is_missing(session, clear_missing, value=False)
+    bulk_update_needs_verify(session, set_needs_verify, value=True)
+
+    summary.tags_added += len(tags_to_add)
+    summary.tags_removed += len(tags_to_remove)
+    summary.missing_flags_cleared += len(clear_missing)
+    summary.verify_flags_set += len(set_needs_verify)
+
+
+def _classify_file(row: DerivedStateRow) -> tuple[bool, bool]:
+    """Return (file is present, file still matches what the row recorded).
+
+    Mirrors the scanner's stat handling so the two agree about what counts as a
+    present file: a permission error means the file is there but unreadable, any
+    other OS error means treat it as gone.
+    """
+    try:
+        stat_result = os.stat(row.file_path, follow_symlinks=True)
+    except FileNotFoundError:
+        return False, False
+    except PermissionError:
+        logging.debug("Permission denied accessing %s", row.file_path)
+        return True, False
+    except OSError as error:
+        logging.debug("OSError checking %s: %s", row.file_path, error)
+        return False, False
+
+    return True, verify_file_unchanged(
+        mtime_db=row.mtime_ns,
+        size_db=row.size_bytes,
+        stat_result=stat_result,
+    )

app/assets/semantics/step.py

@@ -0,0 +1,28 @@
+"""What a semantics reset step is. See ``app.assets.semantics`` for the why."""
+
+from dataclasses import dataclass
+from typing import Callable
+
+InterruptCheck = Callable[[], bool]
+
+
+class SemanticsStepInterrupted(Exception):
+    """A step stopped early on request.
+
+    Its partial work stands -- steps are idempotent, so a resumed run repeats it
+    harmlessly -- but the version is not stamped, so the step runs again.
+    """
+
+
+@dataclass(frozen=True)
+class SemanticsStep:
+    """One numbered reset step.
+
+    ``apply`` takes an optional interrupt check and returns a summary of what it
+    did, which is logged. It must be idempotent and must raise
+    ``SemanticsStepInterrupted`` rather than return if it stops early.
+    """
+
+    version: int
+    description: str
+    apply: Callable[[InterruptCheck | None], object]

app/assets/services/path_utils.py

@@ -325,6 +325,21 @@ def get_path_derived_tags_from_path(path: str) -> list[str]:
     return tags
 
 
+def get_path_derived_tag_vocabulary() -> set[str]:
+    """Every tag get_path_derived_tags_from_path can emit under the current config.
+
+    Bounds which tags a re-derivation may take away: a stored automatic tag
+    inside this vocabulary that the current rules no longer emit was produced
+    by a superseded rule, while one outside it came from somewhere else and is
+    none of the derivation's business.
+    """
+    vocabulary = {"input", "output", "temp", "models"}
+    vocabulary.update(_KNOWN_SUBFOLDER_TAGS)
+    for folder_name, _bases, _extensions in get_comfy_models_folders():
+        vocabulary.add(f"model_type:{folder_name}")
+    return vocabulary
+
+
 def get_name_and_tags_from_asset_path(file_path: str) -> tuple[str, list[str]]:
     """Return (name, tags) derived from a filesystem path.
 

tests-unit/assets_test/test_semantics_reset.py

@@ -0,0 +1,672 @@
+"""Tests for the asset semantics reset (app/assets/semantics).
+
+Runs standalone against in-memory SQLite:
+
+    pytest tests-unit/assets_test/test_semantics_reset.py --noconftest
+"""
+
+import os
+import tempfile
+from datetime import datetime
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+import app.assets.semantics as semantics
+from app.assets.database.models import (
+    Asset,
+    AssetReference,
+    AssetReferenceTag,
+    AssetSemanticsVersion,
+    Base,
+    Tag,
+)
+from app.assets.database.queries.semantics import get_semantics_version
+from app.assets.semantics import run_pending_semantics_steps
+from app.assets.semantics.reproject_derived import reproject_derived_state
+from app.assets.semantics.step import SemanticsStep, SemanticsStepInterrupted
+from app.assets.services.file_utils import get_mtime_ns
+
+
+@pytest.fixture(autouse=True)
+def autoclean_unit_test_assets():
+    """Override the package autouse fixture; these tests need no server."""
+    yield
+
+
+@pytest.fixture
+def session_factory():
+    """A session factory over one shared in-memory database.
+
+    StaticPool keeps every session on the same connection, so the batching walk
+    sees its own committed writes across sessions.
+    """
+    engine = create_engine(
+        "sqlite:///:memory:",
+        poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    Base.metadata.create_all(engine)
+    factory = sessionmaker(bind=engine)
+    with (
+        patch("app.assets.semantics.reproject_derived.create_session", factory),
+        patch("app.assets.semantics.create_session", factory),
+        patch("app.assets.semantics.can_create_session", return_value=True),
+    ):
+        yield factory
+
+
+@pytest.fixture
+def session(session_factory) -> Session:
+    """A session for the test's own reads and writes."""
+    with session_factory() as sess:
+        yield sess
+
+
+@pytest.fixture
+def comfy_dirs():
+    """Point every asset root at a throwaway tree with one model category."""
+    with tempfile.TemporaryDirectory() as base:
+        dirs = {
+            name: Path(base) / name
+            for name in ("checkpoints", "loras", "input", "output", "temp", "elsewhere")
+        }
+        for directory in dirs.values():
+            directory.mkdir()
+        with (
+            patch("folder_paths.get_input_directory", return_value=str(dirs["input"])),
+            patch(
+                "folder_paths.get_output_directory", return_value=str(dirs["output"])
+            ),
+            patch("folder_paths.get_temp_directory", return_value=str(dirs["temp"])),
+            patch(
+                "app.assets.services.path_utils.get_comfy_models_folders",
+                return_value=[
+                    ("checkpoints", [str(dirs["checkpoints"])], {".safetensors"}),
+                    ("loras", [str(dirs["loras"])], {".safetensors"}),
+                ],
+            ),
+        ):
+            yield dirs
+
+
+def _write(directory: Path, name: str, content: bytes = b"\x00" * 100) -> str:
+    path = directory / name
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(content)
+    return os.path.abspath(str(path))
+
+
+def _register(
+    session: Session,
+    file_path: str,
+    ref_id: str,
+    *,
+    loader_path: str | None = None,
+    asset_hash: str | None = "",
+    size_bytes: int = 100,
+    mtime_ns: int | None = None,
+    is_missing: bool = False,
+    needs_verify: bool = False,
+    user_metadata: dict | None = None,
+    preview_id: str | None = None,
+    deleted_at: datetime | None = None,
+    job_id: str | None = None,
+    tags: dict[str, str] | None = None,
+) -> AssetReference:
+    """Insert an Asset + AssetReference (+ tags as {name: origin}) and commit.
+
+    ``asset_hash=""`` means "any unique hash"; assets.hash is unique-indexed.
+    """
+    if asset_hash == "":
+        asset_hash = f"blake3:{ref_id}"
+    if mtime_ns is None:
+        mtime_ns = get_mtime_ns(os.stat(file_path, follow_symlinks=True))
+
+    session.add(Asset(id=f"asset-{ref_id}", hash=asset_hash, size_bytes=size_bytes))
+    session.flush()
+    ref = AssetReference(
+        id=ref_id,
+        asset_id=f"asset-{ref_id}",
+        name=os.path.basename(file_path),
+        owner_id="",
+        file_path=file_path,
+        loader_path=loader_path,
+        mtime_ns=mtime_ns,
+        is_missing=is_missing,
+        needs_verify=needs_verify,
+        user_metadata=user_metadata,
+        preview_id=preview_id,
+        deleted_at=deleted_at,
+        job_id=job_id,
+    )
+    session.add(ref)
+    session.flush()
+
+    for tag_name, origin in (tags or {}).items():
+        session.merge(Tag(name=tag_name))
+        session.add(
+            AssetReferenceTag(
+                asset_reference_id=ref_id, tag_name=tag_name, origin=origin
+            )
+        )
+    session.commit()
+    return ref
+
+
+def _tags(session: Session, ref_id: str) -> dict[str, str]:
+    rows = session.execute(
+        select(AssetReferenceTag.tag_name, AssetReferenceTag.origin).where(
+            AssetReferenceTag.asset_reference_id == ref_id
+        )
+    ).all()
+    return {name: origin for name, origin in rows}
+
+
+def _snapshot(session: Session) -> list[tuple]:
+    """Every field the reset could plausibly touch, for equality across runs."""
+    session.expire_all()
+    refs = session.execute(select(AssetReference).order_by(AssetReference.id)).scalars()
+    rows = [
+        (
+            ref.id,
+            ref.file_path,
+            ref.loader_path,
+            ref.is_missing,
+            ref.needs_verify,
+            ref.mtime_ns,
+            ref.name,
+            ref.preview_id,
+            str(ref.user_metadata),
+            ref.deleted_at,
+            ref.job_id,
+            tuple(sorted(_tags(session, ref.id).items())),
+        )
+        for ref in refs
+    ]
+    assets = session.execute(select(Asset).order_by(Asset.id)).scalars()
+    rows.extend((asset.id, asset.hash, asset.size_bytes) for asset in assets)
+    return rows
+
+
+class TestLoaderPathReprojection:
+    def test_null_loader_path_is_backfilled(self, session, comfy_dirs):
+        """The drift 0006 left behind: a column added with no backfill."""
+        path = _write(comfy_dirs["checkpoints"], "flux/model.safetensors")
+        _register(session, path, "ref-1", loader_path=None)
+
+        reproject_derived_state()
+
+        session.expire_all()
+        ref = session.get(AssetReference, "ref-1")
+        assert ref.loader_path == "flux/model.safetensors"
+
+    def test_stale_loader_path_is_rewritten(self, session, comfy_dirs):
+        path = _write(comfy_dirs["input"], "sub/photo.png")
+        _register(session, path, "ref-1", loader_path="checkpoints/sub/photo.png")
+
+        summary = reproject_derived_state()
+
+        session.expire_all()
+        assert session.get(AssetReference, "ref-1").loader_path == "sub/photo.png"
+        assert summary.loader_paths_rewritten == 1
+
+    def test_correct_loader_path_is_left_alone(self, session, comfy_dirs):
+        path = _write(comfy_dirs["loras"], "style.safetensors")
+        _register(session, path, "ref-1", loader_path="style.safetensors")
+
+        summary = reproject_derived_state()
+
+        assert summary.loader_paths_rewritten == 0
+
+    def test_unloadable_extension_loses_its_loader_path(self, session, comfy_dirs):
+        """The current rule gives no loader path to a file its category cannot load."""
+        path = _write(comfy_dirs["checkpoints"], "notes.txt")
+        _register(session, path, "ref-1", loader_path="notes.txt")
+
+        reproject_derived_state()
+
+        session.expire_all()
+        assert session.get(AssetReference, "ref-1").loader_path is None
+
+    def test_reference_without_file_path_is_skipped(self, session, comfy_dirs):
+        session.add(Asset(id="asset-api", hash="blake3:abc", size_bytes=10))
+        session.flush()
+        session.add(
+            AssetReference(
+                id="ref-api", asset_id="asset-api", name="api.png", owner_id=""
+            )
+        )
+        session.commit()
+
+        summary = reproject_derived_state()
+
+        assert summary.scanned == 0
+        session.expire_all()
+        assert session.get(AssetReference, "ref-api").loader_path is None
+
+
+class TestHashPreservation:
+    def test_unchanged_file_keeps_its_hash_and_is_not_read(self, session, comfy_dirs):
+        path = _write(comfy_dirs["checkpoints"], "big.safetensors")
+        _register(session, path, "ref-1", asset_hash="blake3:original", size_bytes=100)
+
+        with patch(
+            "app.assets.services.hashing.compute_blake3_hash",
+            side_effect=AssertionError("the reset must never re-hash"),
+        ):
+            summary = reproject_derived_state()
+
+        session.expire_all()
+        assert session.get(Asset, "asset-ref-1").hash == "blake3:original"
+        assert summary.unchanged_files == 1
+        assert summary.changed_files == 0
+
+    def test_changed_file_keeps_its_hash_and_is_flagged_for_verify(
+        self, session, comfy_dirs
+    ):
+        """A file that moved on is handed to the existing verify path, not re-read."""
+        path = _write(comfy_dirs["checkpoints"], "big.safetensors")
+        _register(
+            session,
+            path,
+            "ref-1",
+            asset_hash="blake3:original",
+            mtime_ns=1,
+            size_bytes=100,
+        )
+
+        summary = reproject_derived_state()
+
+        session.expire_all()
+        assert session.get(Asset, "asset-ref-1").hash == "blake3:original"
+        assert session.get(Asset, "asset-ref-1").size_bytes == 100
+        assert session.get(AssetReference, "ref-1").needs_verify is True
+        assert summary.changed_files == 1
+
+    def test_changed_file_still_gets_its_path_state_reprojected(
+        self, session, comfy_dirs
+    ):
+        path = _write(comfy_dirs["loras"], "style.safetensors")
+        _register(session, path, "ref-1", loader_path=None, mtime_ns=1)
+
+        reproject_derived_state()
+
+        session.expire_all()
+        assert session.get(AssetReference, "ref-1").loader_path == "style.safetensors"
+
+
+class TestIntentIsPreserved:
+    def test_manual_tags_metadata_preview_and_deletion_survive(
+        self, session, comfy_dirs
+    ):
+        path = _write(comfy_dirs["checkpoints"], "curated.safetensors")
+        other = _write(comfy_dirs["input"], "thumb.png")
+        _register(session, other, "ref-preview")
+        deleted = datetime(2026, 1, 2, 3, 4, 5)
+        _register(
+            session,
+            path,
+            "ref-1",
+            loader_path=None,
+            user_metadata={"note": "hand written", "filename": "kept.safetensors"},
+            preview_id="ref-preview",
+            deleted_at=deleted,
+            job_id="job-42",
+            tags={"favourite": "manual", "uploaded": "upload"},
+        )
+
+        reproject_derived_state()
+
+        session.expire_all()
+        ref = session.get(AssetReference, "ref-1")
+        assert ref.user_metadata == {
+            "note": "hand written",
+            "filename": "kept.safetensors",
+        }
+        assert ref.preview_id == "ref-preview"
+        assert ref.deleted_at == deleted
+        assert ref.job_id == "job-42"
+        assert ref.name == "curated.safetensors"
+        tags = _tags(session, "ref-1")
+        assert tags["favourite"] == "manual"
+        assert tags["uploaded"] == "upload"
+        # ...while the derived state around them was still brought forward.
+        assert ref.loader_path == "curated.safetensors"
+
+    def test_manual_tag_inside_the_derived_vocabulary_is_not_removed(
+        self, session, comfy_dirs
+    ):
+        """A person may tag an input file 'models'; that is their business."""
+        path = _write(comfy_dirs["input"], "photo.png")
+        _register(session, path, "ref-1", tags={"models": "manual"})
+
+        reproject_derived_state()
+
+        assert _tags(session, "ref-1")["models"] == "manual"
+
+
+class TestTagReprojection:
+    def test_missing_derived_tags_are_added(self, session, comfy_dirs):
+        path = _write(comfy_dirs["checkpoints"], "model.safetensors")
+        _register(session, path, "ref-1", tags={})
+
+        reproject_derived_state()
+
+        assert _tags(session, "ref-1") == {
+            "models": "automatic",
+            "model_type:checkpoints": "automatic",
+        }
+
+    def test_superseded_automatic_tag_is_removed(self, session, comfy_dirs):
+        """An older rule tagged by directory alone; extensions now gate the tag."""
+        path = _write(comfy_dirs["checkpoints"], "model.safetensors")
+        _register(
+            session,
+            path,
+            "ref-1",
+            tags={
+                "models": "automatic",
+                "model_type:checkpoints": "automatic",
+                "model_type:loras": "automatic",
+            },
+        )
+
+        reproject_derived_state()
+
+        assert "model_type:loras" not in _tags(session, "ref-1")
+        assert "model_type:checkpoints" in _tags(session, "ref-1")
+
+    def test_automatic_tag_outside_the_vocabulary_is_left_alone(
+        self, session, comfy_dirs
+    ):
+        """'missing' is automatic but not path-derived; the scanner owns it."""
+        path = _write(comfy_dirs["checkpoints"], "model.safetensors")
+        _register(session, path, "ref-1", tags={"missing": "automatic"})
+
+        reproject_derived_state()
+
+        assert "missing" in _tags(session, "ref-1")
+
+
+class TestFileState:
+    def test_present_file_is_unflagged_as_missing(self, session, comfy_dirs):
+        path = _write(comfy_dirs["temp"], "preview.png")
+        _register(session, path, "ref-1", is_missing=True)
+
+        summary = reproject_derived_state()
+
+        session.expire_all()
+        assert session.get(AssetReference, "ref-1").is_missing is False
+        assert summary.missing_flags_cleared == 1
+
+    def test_absent_file_is_left_to_the_scanner(self, session, comfy_dirs):
+        path = os.path.join(str(comfy_dirs["checkpoints"]), "gone.safetensors")
+        _register(
+            session,
+            path,
+            "ref-1",
+            mtime_ns=1,
+            loader_path="stale/gone.safetensors",
+            is_missing=True,
+        )
+
+        summary = reproject_derived_state()
+
+        session.expire_all()
+        ref = session.get(AssetReference, "ref-1")
+        assert ref.is_missing is True
+        assert ref.loader_path == "stale/gone.safetensors"
+        assert summary.absent_files == 1
+
+    def test_path_outside_every_known_root_is_left_alone(self, session, comfy_dirs):
+        """A root missing from the config must not strip tags off its assets."""
+        path = _write(comfy_dirs["elsewhere"], "model.safetensors")
+        _register(
+            session,
+            path,
+            "ref-1",
+            loader_path="model.safetensors",
+            tags={"models": "automatic", "model_type:checkpoints": "automatic"},
+        )
+
+        summary = reproject_derived_state()
+
+        session.expire_all()
+        assert session.get(AssetReference, "ref-1").loader_path == "model.safetensors"
+        assert _tags(session, "ref-1") == {
+            "models": "automatic",
+            "model_type:checkpoints": "automatic",
+        }
+        assert summary.unclassified_paths == 1
+
+
+class TestIdempotence:
+    def test_empty_database_is_a_no_op(self, session, comfy_dirs):
+        summary = reproject_derived_state()
+
+        assert summary.scanned == 0
+        assert summary.loader_paths_rewritten == 0
+        assert summary.tags_added == 0
+
+    def test_second_run_changes_nothing(self, session, comfy_dirs):
+        _register(
+            session,
+            _write(comfy_dirs["checkpoints"], "a/model.safetensors"),
+            "ref-1",
+            loader_path=None,
+            tags={"model_type:loras": "automatic", "favourite": "manual"},
+        )
+        _register(
+            session,
+            _write(comfy_dirs["input"], "pasted/clip.png"),
+            "ref-2",
+            loader_path="wrong.png",
+            is_missing=True,
+        )
+        _register(
+            session,
+            os.path.join(str(comfy_dirs["output"]), "gone.png"),
+            "ref-3",
+            mtime_ns=1,
+        )
+
+        reproject_derived_state()
+        after_first = _snapshot(session)
+
+        second = reproject_derived_state()
+        assert _snapshot(session) == after_first
+        assert second.loader_paths_rewritten == 0
+        assert second.tags_added == 0
+        assert second.tags_removed == 0
+        assert second.missing_flags_cleared == 0
+
+    def test_statements_chunked_by_bind_param_limit_stay_correct(
+        self, session, comfy_dirs
+    ):
+        """The bind-param chunking must not drop or duplicate a tag."""
+        for index in range(4):
+            _register(
+                session,
+                _write(comfy_dirs["checkpoints"], f"m{index}.safetensors"),
+                f"ref-{index}",
+                tags={"model_type:loras": "automatic"},
+            )
+
+        with (
+            patch("app.assets.database.queries.semantics.MAX_BIND_PARAMS", 2),
+            patch("app.assets.database.queries.common.MAX_BIND_PARAMS", 2),
+        ):
+            reproject_derived_state()
+
+        for index in range(4):
+            assert _tags(session, f"ref-{index}") == {
+                "models": "automatic",
+                "model_type:checkpoints": "automatic",
+            }
+
+    def test_walk_crosses_batch_boundaries(self, session, comfy_dirs):
+        for index in range(7):
+            _register(
+                session,
+                _write(comfy_dirs["checkpoints"], f"m{index:02d}.safetensors"),
+                f"ref-{index:02d}",
+                loader_path=None,
+            )
+
+        with patch("app.assets.semantics.reproject_derived._BATCH_SIZE", 2):
+            summary = reproject_derived_state()
+
+        assert summary.scanned == 7
+        session.expire_all()
+        assert all(
+            session.get(AssetReference, f"ref-{index:02d}").loader_path
+            == f"m{index:02d}.safetensors"
+            for index in range(7)
+        )
+
+
+class TestRunner:
+    def test_pending_step_runs_and_stamps(self, session, comfy_dirs):
+        _register(
+            session,
+            _write(comfy_dirs["checkpoints"], "model.safetensors"),
+            "ref-1",
+            loader_path=None,
+        )
+
+        assert run_pending_semantics_steps() == 1
+
+        session.expire_all()
+        assert get_semantics_version(session) == semantics.CURRENT_SEMANTICS_VERSION
+        assert session.get(AssetReference, "ref-1").loader_path == "model.safetensors"
+
+    def test_stamped_database_does_not_walk_again(self, session, comfy_dirs):
+        run_pending_semantics_steps()
+
+        with patch(
+            "app.assets.semantics.reproject_derived.get_file_backed_references_page",
+            side_effect=AssertionError("a stamped database must not be walked"),
+        ):
+            assert run_pending_semantics_steps() == 0
+
+    def test_stamp_is_not_advanced_when_a_step_raises(self, session, comfy_dirs):
+        def _explode(_interrupt_check):
+            raise RuntimeError("step failed")
+
+        with patch.object(
+            semantics,
+            "SEMANTICS_STEPS",
+            (SemanticsStep(version=1, description="explodes", apply=_explode),),
+        ):
+            assert run_pending_semantics_steps() == 0
+
+        session.expire_all()
+        assert get_semantics_version(session) == 0
+        assert session.get(AssetSemanticsVersion, 1) is None
+
+    def test_stamp_is_not_advanced_when_a_step_is_interrupted(
+        self, session, comfy_dirs
+    ):
+        def _interrupted(_interrupt_check):
+            raise SemanticsStepInterrupted("stopped")
+
+        with patch.object(
+            semantics,
+            "SEMANTICS_STEPS",
+            (SemanticsStep(version=1, description="interrupts", apply=_interrupted),),
+        ):
+            assert run_pending_semantics_steps() == 0
+
+        session.expire_all()
+        assert get_semantics_version(session) == 0
+
+    def test_earlier_steps_stay_stamped_when_a_later_one_fails(
+        self, session, comfy_dirs
+    ):
+        def _ok(_interrupt_check):
+            return "fine"
+
+        def _explode(_interrupt_check):
+            raise RuntimeError("step failed")
+
+        with patch.object(
+            semantics,
+            "SEMANTICS_STEPS",
+            (
+                SemanticsStep(version=1, description="ok", apply=_ok),
+                SemanticsStep(version=2, description="explodes", apply=_explode),
+            ),
+        ):
+            assert run_pending_semantics_steps() == 1
+
+        session.expire_all()
+        assert get_semantics_version(session) == 1
+
+    def test_steps_below_the_stored_version_are_skipped(self, session, comfy_dirs):
+        applied: list[int] = []
+
+        def _record(version):
+            def _apply(_interrupt_check):
+                applied.append(version)
+                return version
+
+            return _apply
+
+        steps = (
+            SemanticsStep(version=1, description="one", apply=_record(1)),
+            SemanticsStep(version=2, description="two", apply=_record(2)),
+        )
+        with patch.object(semantics, "SEMANTICS_STEPS", steps[:1]):
+            run_pending_semantics_steps()
+        with patch.object(semantics, "SEMANTICS_STEPS", steps):
+            run_pending_semantics_steps()
+
+        assert applied == [1, 2]
+
+    def test_interrupted_walk_leaves_committed_work_and_resumes(
+        self, session, comfy_dirs
+    ):
+        for index in range(4):
+            _register(
+                session,
+                _write(comfy_dirs["checkpoints"], f"m{index}.safetensors"),
+                f"ref-{index}",
+                loader_path=None,
+            )
+
+        calls = {"n": 0}
+
+        def _stop_after_first_batch() -> bool:
+            calls["n"] += 1
+            return calls["n"] > 1
+
+        with (
+            patch("app.assets.semantics.reproject_derived._BATCH_SIZE", 2),
+            patch.object(
+                semantics,
+                "SEMANTICS_STEPS",
+                (
+                    SemanticsStep(
+                        version=1,
+                        description="reproject",
+                        apply=reproject_derived_state,
+                    ),
+                ),
+            ),
+        ):
+            assert run_pending_semantics_steps(_stop_after_first_batch) == 0
+            session.expire_all()
+            assert get_semantics_version(session) == 0
+            assert session.get(AssetReference, "ref-0").loader_path is not None
+            assert session.get(AssetReference, "ref-3").loader_path is None
+
+            assert run_pending_semantics_steps() == 1
+
+        session.expire_all()
+        assert get_semantics_version(session) == 1
+        assert session.get(AssetReference, "ref-3").loader_path == "m3.safetensors"

tests-unit/seeder_test/test_seeder.py

@@ -23,6 +23,7 @@ def mock_dependencies():
     """Mock all external dependencies for isolated testing."""
     with (
         patch("app.assets.seeder.dependencies_available", return_value=True),
+        patch("app.assets.seeder.run_pending_semantics_steps", return_value=0),
         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=([], set(), 0)),
@@ -454,6 +455,7 @@ class TestSeederMarkMissing:
 
         with (
             patch("app.assets.seeder.dependencies_available", return_value=True),
+            patch("app.assets.seeder.run_pending_semantics_steps", return_value=0),
             patch("app.assets.seeder.get_owned_prefixes", return_value=["/models"]),
             patch("app.assets.seeder.mark_missing_outside_prefixes_safely", side_effect=track_mark),
             patch("app.assets.seeder.sync_temp_references_safely"),
@@ -475,6 +477,7 @@ class TestSeederMarkMissing:
     ):
         with (
             patch("app.assets.seeder.dependencies_available", return_value=True),
+            patch("app.assets.seeder.run_pending_semantics_steps", return_value=0),
             patch("app.assets.seeder.get_owned_prefixes", return_value=["/models"]),
             patch("app.assets.seeder.mark_missing_outside_prefixes_safely", return_value=0),
             patch("app.assets.seeder.sync_temp_references_safely") as sync_temp,
@@ -494,6 +497,73 @@ class TestSeederMarkMissing:
             )
 
 
+class TestSeederSemanticsReset:
+    """The scan brings stale rows forward before it reads or extends them."""
+
+    def test_semantics_reset_runs_before_any_scan_work(
+        self, fresh_seeder: _AssetSeeder
+    ):
+        call_order = []
+
+        with (
+            patch("app.assets.seeder.dependencies_available", return_value=True),
+            patch(
+                "app.assets.seeder.run_pending_semantics_steps",
+                side_effect=lambda interrupt_check=None: call_order.append("reset"),
+            ),
+            patch("app.assets.seeder.get_owned_prefixes", return_value=["/models"]),
+            patch(
+                "app.assets.seeder.mark_missing_outside_prefixes_safely",
+                side_effect=lambda prefixes: call_order.append("prune") or 0,
+            ),
+            patch(
+                "app.assets.seeder.sync_temp_references_safely",
+                side_effect=lambda: call_order.append("sync_temp"),
+            ),
+            patch(
+                "app.assets.seeder.sync_root_safely",
+                side_effect=lambda root: call_order.append("sync") or set(),
+            ),
+            patch("app.assets.seeder.collect_paths_for_roots", return_value=[]),
+            patch("app.assets.seeder.build_asset_specs", return_value=([], set(), 0)),
+            patch("app.assets.seeder.insert_asset_specs", return_value=0),
+            patch("app.assets.seeder.get_unenriched_assets_for_roots", return_value=[]),
+            patch("app.assets.seeder.enrich_assets_batch", return_value=(0, 0)),
+        ):
+            fresh_seeder.start(roots=("models",), prune_first=True)
+            fresh_seeder.wait(timeout=5.0)
+
+        assert call_order[0] == "reset", (
+            "reprojection must finish before the scan reads or extends those rows"
+        )
+
+    def test_semantics_reset_can_be_cancelled(self, fresh_seeder: _AssetSeeder):
+        captured = {}
+
+        with (
+            patch("app.assets.seeder.dependencies_available", return_value=True),
+            patch(
+                "app.assets.seeder.run_pending_semantics_steps",
+                side_effect=lambda interrupt_check=None: captured.update(
+                    check=interrupt_check
+                ),
+            ),
+            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=([], set(), 0)),
+            patch("app.assets.seeder.insert_asset_specs", return_value=0),
+            patch("app.assets.seeder.get_unenriched_assets_for_roots", return_value=[]),
+            patch("app.assets.seeder.enrich_assets_batch", return_value=(0, 0)),
+        ):
+            fresh_seeder.start(roots=("models",))
+            fresh_seeder.wait(timeout=5.0)
+
+        assert captured.get("check") is not None
+        assert captured["check"]() is False
+        fresh_seeder._cancel_event.set()
+        assert captured["check"]() is True
+
+
 class TestSeederPhases:
     """Test phased scanning behavior."""