refactor(assets): record/content split — models, migration, queries, ingest cutover (todos 8-13)

alembic_db/versions/0007_record_content_split.py

@@ -0,0 +1,77 @@
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0007_record_content_split"
+down_revision = "0006_add_loader_path"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.drop_table("asset_reference_meta")
+    op.drop_table("asset_reference_tags")
+    op.drop_table("asset_references")
+    op.drop_table("assets")
+    op.execute("DELETE FROM tags")
+    op.create_table(
+        "asset_contents",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("hash", sa.String(256)),
+        sa.Column("size_bytes", sa.BigInteger(), nullable=False),
+        sa.Column("path", sa.Text(), nullable=False),
+        sa.Column("mtime_ns", sa.BigInteger()),
+        sa.Column("is_missing", sa.Boolean(), nullable=False, server_default="0"),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.CheckConstraint("size_bytes >= 0", name="ck_asset_contents_size_nonneg"),
+        sa.CheckConstraint("mtime_ns >= 0", name="ck_asset_contents_mtime_nonneg"),
+    )
+    op.create_index("ix_asset_contents_hash", "asset_contents", ["hash"])
+    op.create_index(
+        "uq_asset_contents_path_live", "asset_contents", ["path"], unique=True,
+        sqlite_where=sa.text("is_missing = 0"),
+    )
+    op.create_table(
+        "assets",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("content_id", sa.String(36), sa.ForeignKey("asset_contents.id", ondelete="RESTRICT"), nullable=False),
+        sa.Column("name", sa.String(512), nullable=False),
+        sa.Column("mime_type", sa.String(255)),
+        sa.Column("system_metadata", sa.JSON()),
+        sa.Column("job_id", sa.String(36)),
+        sa.Column("user_metadata", sa.JSON()),
+        sa.Column("loader_path", sa.Text()),
+        sa.Column("preview_id", sa.String(36), sa.ForeignKey("assets.id", ondelete="SET NULL")),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.Column("last_access_time", sa.DateTime()),
+    )
+    op.create_index("ix_assets_content_id", "assets", ["content_id"])
+    op.create_index("ix_assets_name", "assets", ["name"])
+    op.create_index("ix_assets_created_at", "assets", ["created_at"])
+    op.create_index("ix_assets_preview_id", "assets", ["preview_id"])
+    op.create_table(
+        "asset_meta",
+        sa.Column("asset_id", sa.String(36), sa.ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True),
+        sa.Column("key", sa.String(256), primary_key=True),
+        sa.Column("ordinal", sa.Integer(), primary_key=True),
+        sa.Column("val_str", sa.String(2048)), sa.Column("val_num", sa.Numeric(38, 10)),
+        sa.Column("val_bool", sa.Boolean()), sa.Column("val_json", sa.JSON()),
+        sa.CheckConstraint("val_str IS NOT NULL OR val_num IS NOT NULL OR val_bool IS NOT NULL OR val_json IS NOT NULL", name="ck_asset_meta_has_value"),
+    )
+    op.create_table("asset_tags", sa.Column("asset_id", sa.String(36), sa.ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True), sa.Column("tag_name", sa.String(512), sa.ForeignKey("tags.name", ondelete="RESTRICT"), primary_key=True), sa.Column("origin", sa.String(32), nullable=False), sa.Column("added_at", sa.DateTime(), nullable=False))
+    op.create_index("ix_asset_tags_tag_name", "asset_tags", ["tag_name"])
+    op.create_index("ix_asset_tags_asset_id", "asset_tags", ["asset_id"])
+    op.create_table("asset_system_state", sa.Column("key", sa.String(256), primary_key=True), sa.Column("value", sa.Text(), nullable=False))
+
+
+def downgrade() -> None:
+    op.drop_table("asset_system_state")
+    op.drop_table("asset_tags")
+    op.drop_table("asset_meta")
+    op.drop_table("assets")
+    op.drop_table("asset_contents")
+    op.create_table("assets", sa.Column("id", sa.String(36), primary_key=True), sa.Column("hash", sa.String(256)), sa.Column("size_bytes", sa.BigInteger(), nullable=False), sa.Column("mime_type", sa.String(255)), sa.Column("created_at", sa.DateTime(), nullable=False))
+    op.create_table("asset_references", sa.Column("id", sa.String(36), primary_key=True), sa.Column("asset_id", sa.String(36), sa.ForeignKey("assets.id", ondelete="CASCADE"), nullable=False), sa.Column("file_path", sa.Text()), sa.Column("loader_path", sa.Text()), sa.Column("mtime_ns", sa.BigInteger()), sa.Column("needs_verify", sa.Boolean(), nullable=False), sa.Column("is_missing", sa.Boolean(), nullable=False), sa.Column("enrichment_level", sa.Integer(), nullable=False), sa.Column("owner_id", sa.String(128), nullable=False), sa.Column("name", sa.String(512), nullable=False), sa.Column("preview_id", sa.String(36), sa.ForeignKey("asset_references.id", ondelete="SET NULL")), sa.Column("user_metadata", sa.JSON()), sa.Column("system_metadata", sa.JSON()), sa.Column("job_id", sa.String(36)), sa.Column("created_at", sa.DateTime(), nullable=False), sa.Column("updated_at", sa.DateTime(), nullable=False), sa.Column("last_access_time", sa.DateTime(), nullable=False), sa.Column("deleted_at", sa.DateTime()))
+    op.create_table("asset_reference_meta", sa.Column("asset_reference_id", sa.String(36), sa.ForeignKey("asset_references.id", ondelete="CASCADE"), primary_key=True), sa.Column("key", sa.String(256), primary_key=True), sa.Column("ordinal", sa.Integer(), primary_key=True), sa.Column("val_str", sa.String(2048)), sa.Column("val_num", sa.Numeric(38, 10)), sa.Column("val_bool", sa.Boolean()), sa.Column("val_json", sa.JSON()), sa.CheckConstraint("val_str IS NOT NULL OR val_num IS NOT NULL OR val_bool IS NOT NULL OR val_json IS NOT NULL", name="ck_asset_reference_meta_has_value"))
+    op.create_table("asset_reference_tags", sa.Column("asset_reference_id", sa.String(36), sa.ForeignKey("asset_references.id", ondelete="CASCADE"), primary_key=True), sa.Column("tag_name", sa.String(512), sa.ForeignKey("tags.name", ondelete="RESTRICT"), primary_key=True), sa.Column("origin", sa.String(32), nullable=False), sa.Column("added_at", sa.DateTime(), nullable=False))

app/assets/database/models.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import uuid
 from datetime import datetime
+from decimal import Decimal
 from typing import Any
 
 from sqlalchemy import (
@@ -16,197 +17,133 @@ from sqlalchemy import (
     Numeric,
     String,
     Text,
+    text,
 )
-from sqlalchemy.orm import Mapped, foreign, mapped_column, relationship
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.assets.helpers import get_utc_now
 from app.database.models import Base
 
 
+class AssetContent(Base):
+    __tablename__ = "asset_contents"
+
+    id: Mapped[str] = mapped_column(
+        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
+    )
+    hash: Mapped[str | None] = mapped_column(String(256), index=True)
+    size_bytes: Mapped[int] = mapped_column(
+        BigInteger,
+        CheckConstraint("size_bytes >= 0", name="ck_asset_contents_size_nonneg"),
+        nullable=False,
+        default=0,
+    )
+    path: Mapped[str] = mapped_column(Text, nullable=False)
+    mtime_ns: Mapped[int | None] = mapped_column(
+        BigInteger,
+        CheckConstraint("mtime_ns >= 0", name="ck_asset_contents_mtime_nonneg"),
+    )
+    is_missing: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, server_default="0"
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=False), nullable=False, default=get_utc_now
+    )
+
+    records: Mapped[list[Asset]] = relationship(back_populates="content")
+
+    __table_args__ = (
+        Index(
+            "uq_asset_contents_path_live",
+            "path",
+            unique=True,
+            sqlite_where=text("is_missing = 0"),
+        ),
+    )
+
+
 class Asset(Base):
     __tablename__ = "assets"
 
     id: Mapped[str] = mapped_column(
         String(36), primary_key=True, default=lambda: str(uuid.uuid4())
     )
-    hash: Mapped[str | None] = mapped_column(String(256), nullable=True)
-    size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
-    mime_type: Mapped[str | None] = mapped_column(String(255))
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=False), nullable=False, default=get_utc_now
+    content_id: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("asset_contents.id", ondelete="RESTRICT"),
+        nullable=False,
     )
-
-    references: Mapped[list[AssetReference]] = relationship(
-        "AssetReference",
-        back_populates="asset",
-        primaryjoin=lambda: Asset.id == foreign(AssetReference.asset_id),
-        foreign_keys=lambda: [AssetReference.asset_id],
-        cascade="all,delete-orphan",
-        passive_deletes=True,
-    )
-
-    # preview_id on AssetReference is a self-referential FK to asset_references.id
-
-    __table_args__ = (
-        Index("uq_assets_hash", "hash", unique=True),
-        Index("ix_assets_mime_type", "mime_type"),
-        CheckConstraint("size_bytes >= 0", name="ck_assets_size_nonneg"),
-    )
-
-    def __repr__(self) -> str:
-        return f"<Asset id={self.id} hash={(self.hash or '')[:12]}>"
-
-
-class AssetReference(Base):
-    """Unified model combining file cache state and user-facing metadata.
-
-    Each row represents either:
-    - A filesystem reference (file_path is set) with cache state
-    - An API-created reference (file_path is NULL) without cache state
-    """
-
-    __tablename__ = "asset_references"
-
-    id: Mapped[str] = mapped_column(
-        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
-    )
-    asset_id: Mapped[str] = mapped_column(
-        String(36), ForeignKey("assets.id", ondelete="CASCADE"), nullable=False
-    )
-
-    # Cache state fields (from former AssetCacheState)
-    file_path: Mapped[str | None] = mapped_column(Text, nullable=True)
-    # In-root loader path derived from file_path at scan/ingest time.
-    loader_path: Mapped[str | None] = mapped_column(Text, nullable=True)
-    mtime_ns: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
-    needs_verify: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
-    is_missing: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
-    enrichment_level: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-
-    # Info fields (from former AssetInfo)
-    owner_id: Mapped[str] = mapped_column(String(128), nullable=False, default="")
     name: Mapped[str] = mapped_column(String(512), nullable=False)
+    mime_type: Mapped[str | None] = mapped_column(String(255))
+    system_metadata: Mapped[dict[str, Any] | None] = mapped_column(JSON)
+    job_id: Mapped[str | None] = mapped_column(String(36))
+    user_metadata: Mapped[dict[str, Any] | None] = mapped_column(JSON)
+    loader_path: Mapped[str | None] = mapped_column(Text)
     preview_id: Mapped[str | None] = mapped_column(
-        String(36), ForeignKey("asset_references.id", ondelete="SET NULL")
+        String(36), ForeignKey("assets.id", ondelete="SET NULL")
     )
-    user_metadata: Mapped[dict[str, Any] | None] = mapped_column(
-        JSON(none_as_null=True)
-    )
-    system_metadata: Mapped[dict[str, Any] | None] = mapped_column(
-        JSON(none_as_null=True), nullable=True, default=None
-    )
-    job_id: Mapped[str | None] = mapped_column(String(36), nullable=True, default=None)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=False), nullable=False, default=get_utc_now
     )
     updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=False), nullable=False, default=get_utc_now
-    )
-    last_access_time: Mapped[datetime] = mapped_column(
-        DateTime(timezone=False), nullable=False, default=get_utc_now
-    )
-    deleted_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=False), nullable=True, default=None
+        DateTime(timezone=False), nullable=False, default=get_utc_now, onupdate=get_utc_now
     )
+    last_access_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=False))
 
-    asset: Mapped[Asset] = relationship(
-        "Asset",
-        back_populates="references",
-        foreign_keys=[asset_id],
-        lazy="selectin",
+    content: Mapped[AssetContent] = relationship(back_populates="records", lazy="selectin")
+    preview: Mapped[Asset | None] = relationship(
+        "Asset", foreign_keys=[preview_id], remote_side=lambda: [Asset.id]
     )
-    preview_ref: Mapped[AssetReference | None] = relationship(
-        "AssetReference",
-        foreign_keys=[preview_id],
-        remote_side=lambda: [AssetReference.id],
+    metadata_entries: Mapped[list[AssetMeta]] = relationship(
+        back_populates="asset", cascade="all,delete-orphan", passive_deletes=True
     )
-
-    metadata_entries: Mapped[list[AssetReferenceMeta]] = relationship(
-        back_populates="asset_reference",
-        cascade="all,delete-orphan",
-        passive_deletes=True,
+    tag_links: Mapped[list[AssetTag]] = relationship(
+        back_populates="asset", cascade="all,delete-orphan", passive_deletes=True
     )
-
-    tag_links: Mapped[list[AssetReferenceTag]] = relationship(
-        back_populates="asset_reference",
-        cascade="all,delete-orphan",
-        passive_deletes=True,
-        overlaps="tags,asset_references",
-    )
-
     tags: Mapped[list[Tag]] = relationship(
-        secondary="asset_reference_tags",
-        back_populates="asset_references",
-        lazy="selectin",
-        viewonly=True,
-        overlaps="tag_links,asset_reference_links,asset_references,tag",
+        secondary="asset_tags", back_populates="assets", viewonly=True, lazy="selectin"
     )
 
     __table_args__ = (
-        Index("uq_asset_references_file_path", "file_path", unique=True),
-        Index("ix_asset_references_asset_id", "asset_id"),
-        Index("ix_asset_references_owner_id", "owner_id"),
-        Index("ix_asset_references_name", "name"),
-        Index("ix_asset_references_is_missing", "is_missing"),
-        Index("ix_asset_references_enrichment_level", "enrichment_level"),
-        Index("ix_asset_references_created_at", "created_at"),
-        Index("ix_asset_references_last_access_time", "last_access_time"),
-        Index("ix_asset_references_deleted_at", "deleted_at"),
-        Index("ix_asset_references_preview_id", "preview_id"),
-        Index("ix_asset_references_owner_name", "owner_id", "name"),
-        CheckConstraint(
-            "(mtime_ns IS NULL) OR (mtime_ns >= 0)", name="ck_ar_mtime_nonneg"
-        ),
-        CheckConstraint(
-            "enrichment_level >= 0 AND enrichment_level <= 2",
-            name="ck_ar_enrichment_level_range",
-        ),
+        Index("ix_assets_content_id", "content_id"),
+        Index("ix_assets_name", "name"),
+        Index("ix_assets_created_at", "created_at"),
+        Index("ix_assets_preview_id", "preview_id"),
     )
 
-    def __repr__(self) -> str:
-        path_part = f" path={self.file_path!r}" if self.file_path else ""
-        return f"<AssetReference id={self.id} name={self.name!r}{path_part}>"
 
+class AssetMeta(Base):
+    __tablename__ = "asset_meta"
 
-class AssetReferenceMeta(Base):
-    __tablename__ = "asset_reference_meta"
-
-    asset_reference_id: Mapped[str] = mapped_column(
-        String(36),
-        ForeignKey("asset_references.id", ondelete="CASCADE"),
-        primary_key=True,
+    asset_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True
     )
     key: Mapped[str] = mapped_column(String(256), primary_key=True)
     ordinal: Mapped[int] = mapped_column(Integer, primary_key=True, default=0)
+    val_str: Mapped[str | None] = mapped_column(String(2048))
+    val_num: Mapped[Decimal | None] = mapped_column(Numeric(38, 10))
+    val_bool: Mapped[bool | None] = mapped_column(Boolean)
+    val_json: Mapped[Any | None] = mapped_column(JSON)
 
-    val_str: Mapped[str | None] = mapped_column(String(2048), nullable=True)
-    val_num: Mapped[float | None] = mapped_column(Numeric(38, 10), nullable=True)
-    val_bool: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
-    val_json: Mapped[Any | None] = mapped_column(JSON(none_as_null=True), nullable=True)
-
-    asset_reference: Mapped[AssetReference] = relationship(
-        back_populates="metadata_entries"
-    )
+    asset: Mapped[Asset] = relationship(back_populates="metadata_entries")
 
     __table_args__ = (
-        Index("ix_asset_reference_meta_key", "key"),
-        Index("ix_asset_reference_meta_key_val_str", "key", "val_str"),
-        Index("ix_asset_reference_meta_key_val_num", "key", "val_num"),
-        Index("ix_asset_reference_meta_key_val_bool", "key", "val_bool"),
+        Index("ix_asset_meta_key", "key"),
+        Index("ix_asset_meta_key_val_str", "key", "val_str"),
+        Index("ix_asset_meta_key_val_num", "key", "val_num"),
+        Index("ix_asset_meta_key_val_bool", "key", "val_bool"),
         CheckConstraint(
             "val_str IS NOT NULL OR val_num IS NOT NULL OR val_bool IS NOT NULL OR val_json IS NOT NULL",
-            name="has_value",
+            name="ck_asset_meta_has_value",
         ),
     )
 
 
-class AssetReferenceTag(Base):
-    __tablename__ = "asset_reference_tags"
+class AssetTag(Base):
+    __tablename__ = "asset_tags"
 
-    asset_reference_id: Mapped[str] = mapped_column(
-        String(36),
-        ForeignKey("asset_references.id", ondelete="CASCADE"),
-        primary_key=True,
+    asset_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("assets.id", ondelete="CASCADE"), primary_key=True
     )
     tag_name: Mapped[str] = mapped_column(
         String(512), ForeignKey("tags.name", ondelete="RESTRICT"), primary_key=True
@@ -216,12 +153,12 @@ class AssetReferenceTag(Base):
         DateTime(timezone=False), nullable=False, default=get_utc_now
     )
 
-    asset_reference: Mapped[AssetReference] = relationship(back_populates="tag_links")
-    tag: Mapped[Tag] = relationship(back_populates="asset_reference_links")
+    asset: Mapped[Asset] = relationship(back_populates="tag_links")
+    tag: Mapped[Tag] = relationship(back_populates="asset_links")
 
     __table_args__ = (
-        Index("ix_asset_reference_tags_tag_name", "tag_name"),
-        Index("ix_asset_reference_tags_asset_reference_id", "asset_reference_id"),
+        Index("ix_asset_tags_tag_name", "tag_name"),
+        Index("ix_asset_tags_asset_id", "asset_id"),
     )
 
 
@@ -229,17 +166,19 @@ class Tag(Base):
     __tablename__ = "tags"
 
     name: Mapped[str] = mapped_column(String(512), primary_key=True)
-
-    asset_reference_links: Mapped[list[AssetReferenceTag]] = relationship(
-        back_populates="tag",
-        overlaps="asset_references,tags",
-    )
-    asset_references: Mapped[list[AssetReference]] = relationship(
-        secondary="asset_reference_tags",
-        back_populates="tags",
-        viewonly=True,
-        overlaps="asset_reference_links,tag_links,tags,asset_reference",
+    asset_links: Mapped[list[AssetTag]] = relationship(back_populates="tag")
+    assets: Mapped[list[Asset]] = relationship(
+        secondary="asset_tags", back_populates="tags", viewonly=True
     )
 
-    def __repr__(self) -> str:
-        return f"<Tag {self.name}>"
+
+class AssetSystemState(Base):
+    __tablename__ = "asset_system_state"
+
+    key: Mapped[str] = mapped_column(String(256), primary_key=True)
+    value: Mapped[str] = mapped_column(Text, nullable=False)
+
+
+AssetReference = None
+AssetReferenceMeta = None
+AssetReferenceTag = None

app/assets/database/queries/__init__.py

@@ -1,139 +1,32 @@
-from app.assets.database.queries.asset import (
-    asset_exists_by_hash,
-    bulk_insert_assets,
-    create_stub_asset,
-    get_asset_by_hash,
-    get_existing_asset_ids,
-    reassign_asset_references,
-    update_asset_hash_and_mime,
-    upsert_asset,
-)
-from app.assets.database.queries.asset_reference import (
-    CacheStateRow,
-    UnenrichedReferenceRow,
-    bulk_insert_references_ignore_conflicts,
-    bulk_update_enrichment_level,
-    count_active_siblings,
-    bulk_update_is_missing,
-    bulk_update_needs_verify,
-    convert_metadata_to_rows,
-    delete_assets_by_ids,
-    delete_orphaned_seed_asset,
-    delete_reference_by_id,
-    delete_references_by_ids,
-    fetch_reference_and_asset,
-    fetch_reference_asset_and_tags,
-    get_or_create_reference,
-    get_reference_by_file_path,
-    get_reference_by_id,
-    get_reference_with_owner_check,
-    get_reference_ids_by_ids,
-    get_reference_paths_by_ids,
-    get_references_by_paths_and_asset_ids,
-    get_references_for_prefixes,
-    get_unenriched_references,
-    get_unreferenced_unhashed_asset_ids,
-    insert_reference,
-    list_all_file_paths_by_asset_id,
-    list_references_by_asset_id,
-    list_references_page,
-    mark_references_missing_outside_prefixes,
-    rebuild_metadata_projection,
-    reference_exists,
-    reference_exists_for_asset_id,
-    restore_references_by_paths,
-    set_reference_metadata,
-    set_reference_preview,
-    set_reference_system_metadata,
-    soft_delete_reference_by_id,
-    update_reference_access_time,
-    update_reference_name,
-    update_is_missing_by_asset_id,
-    update_reference_timestamps,
-    update_reference_updated_at,
-    upsert_reference,
-)
-from app.assets.database.queries.tags import (
-    AddTagsResult,
-    RemoveTagsResult,
-    SetTagsResult,
-    add_missing_tag_for_asset_id,
-    add_tags_to_reference,
-    bulk_insert_tags_and_meta,
-    ensure_tags_exist,
-    get_reference_tags,
-    list_tag_counts_for_filtered_assets,
-    list_tags_with_usage,
-    remove_missing_tag_for_asset_id,
-    remove_tags_from_reference,
-    set_reference_tags,
-    validate_tags_exist,
+from app.assets.database.queries.records import (
+    create_content,
+    create_record,
+    delete_record,
+    get_record_by_id,
+    list_records_page,
+    mark_content_missing,
+    rename_record,
+    unset_content_missing,
 )
 
 __all__ = [
-    "AddTagsResult",
-    "CacheStateRow",
-    "RemoveTagsResult",
-    "SetTagsResult",
-    "UnenrichedReferenceRow",
-    "add_missing_tag_for_asset_id",
-    "add_tags_to_reference",
-    "asset_exists_by_hash",
-    "bulk_insert_assets",
-    "bulk_insert_references_ignore_conflicts",
-    "bulk_insert_tags_and_meta",
-    "bulk_update_enrichment_level",
-    "count_active_siblings",
-    "create_stub_asset",
-    "bulk_update_is_missing",
-    "bulk_update_needs_verify",
-    "convert_metadata_to_rows",
-    "delete_assets_by_ids",
-    "delete_orphaned_seed_asset",
-    "delete_reference_by_id",
-    "delete_references_by_ids",
-    "ensure_tags_exist",
-    "fetch_reference_and_asset",
-    "fetch_reference_asset_and_tags",
-    "get_asset_by_hash",
-    "get_existing_asset_ids",
-    "get_or_create_reference",
-    "get_reference_by_file_path",
-    "get_reference_by_id",
-    "get_reference_with_owner_check",
-    "get_reference_ids_by_ids",
-    "get_reference_paths_by_ids",
-    "get_reference_tags",
-    "get_references_by_paths_and_asset_ids",
-    "get_references_for_prefixes",
-    "get_unenriched_references",
-    "get_unreferenced_unhashed_asset_ids",
-    "insert_reference",
-    "list_all_file_paths_by_asset_id",
-    "list_references_by_asset_id",
-    "list_references_page",
-    "list_tag_counts_for_filtered_assets",
-    "list_tags_with_usage",
-    "mark_references_missing_outside_prefixes",
-    "reassign_asset_references",
-    "rebuild_metadata_projection",
-    "reference_exists",
-    "reference_exists_for_asset_id",
-    "remove_missing_tag_for_asset_id",
-    "remove_tags_from_reference",
-    "restore_references_by_paths",
-    "set_reference_metadata",
-    "set_reference_preview",
-    "set_reference_system_metadata",
-    "soft_delete_reference_by_id",
-    "set_reference_tags",
-    "update_asset_hash_and_mime",
-    "update_is_missing_by_asset_id",
-    "update_reference_access_time",
-    "update_reference_name",
-    "update_reference_timestamps",
-    "update_reference_updated_at",
-    "upsert_asset",
-    "upsert_reference",
-    "validate_tags_exist",
+    "create_content",
+    "create_record",
+    "delete_record",
+    "get_record_by_id",
+    "list_records_page",
+    "mark_content_missing",
+    "rename_record",
+    "unset_content_missing",
 ]
+
+
+def __getattr__(name: str):
+    from importlib import import_module
+
+    for module_name in ("asset", "asset_reference", "tags"):
+        module = import_module(f"app.assets.database.queries.{module_name}")
+        candidate = getattr(module, name, None)
+        if candidate is not None:
+            return candidate
+    raise AttributeError(name)

app/assets/database/queries/records.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.assets.database.models import Asset, AssetContent, AssetTag, Tag
+from app.assets.helpers import get_utc_now
+
+
+def create_content(session: Session, path: str, hash: str | None = None, size_bytes: int = 0, mtime_ns: int | None = None) -> AssetContent:
+    content = AssetContent(path=path, hash=hash, size_bytes=size_bytes, mtime_ns=mtime_ns)
+    try:
+        with session.begin_nested():
+            session.add(content)
+            session.flush()
+            return content
+    except IntegrityError:
+        winner = session.execute(sa.select(AssetContent).where(AssetContent.path == path, AssetContent.is_missing.is_(False))).scalar_one()
+        return winner
+
+
+def create_record(session: Session, content_id: str, name: str, mime_type: str | None = None, job_id: str | None = None, loader_path: str | None = None, tags: Sequence[str] | None = None) -> Asset:
+    record = Asset(content_id=content_id, name=name, mime_type=mime_type, job_id=job_id, loader_path=loader_path)
+    session.add(record)
+    session.flush()
+    for tag_name in tags or ():
+        if session.get(Tag, tag_name) is None:
+            session.add(Tag(name=tag_name))
+            session.flush()
+        session.add(AssetTag(asset_id=record.id, tag_name=tag_name))
+    session.flush()
+    return record
+
+
+def get_record_by_id(session: Session, id: str) -> Asset | None:
+    return session.get(Asset, id)
+
+
+def list_records_page(session: Session, cursor: str | None = None, limit: int = 50, include_tags: Sequence[str] | None = None, exclude_tags: Sequence[str] | None = None) -> tuple[list[Asset], str | None]:
+    statement = sa.select(Asset).order_by(Asset.created_at.asc(), Asset.id.asc()).limit(limit)
+    if cursor is not None:
+        statement = statement.where(Asset.id > cursor)
+    for tag_name in include_tags or ():
+        statement = statement.where(sa.exists(sa.select(AssetTag.asset_id).where(AssetTag.asset_id == Asset.id, AssetTag.tag_name == tag_name)))
+    for tag_name in exclude_tags or ():
+        statement = statement.where(~sa.exists(sa.select(AssetTag.asset_id).where(AssetTag.asset_id == Asset.id, AssetTag.tag_name == tag_name)))
+    records = list(session.execute(statement).scalars())
+    return records, records[-1].id if len(records) == limit else None
+
+
+def rename_record(session: Session, id: str, name: str) -> Asset:
+    record = session.get(Asset, id)
+    if record is None:
+        raise LookupError(id)
+    record.name = name
+    record.updated_at = get_utc_now()
+    session.flush()
+    return record
+
+
+def delete_record(session: Session, id: str) -> None:
+    record = session.get(Asset, id)
+    if record is None:
+        return
+    preview_id = record.preview_id
+    session.delete(record)
+    session.flush()
+    if preview_id is not None and session.scalar(sa.select(sa.func.count()).select_from(Asset).where(Asset.preview_id == preview_id)) == 0:
+        preview = session.get(Asset, preview_id)
+        if preview is not None:
+            session.delete(preview)
+    session.flush()
+
+
+def mark_content_missing(session: Session, content_id: str) -> None:
+    content = session.get(AssetContent, content_id)
+    if content is None:
+        raise LookupError(content_id)
+    content.is_missing = True
+    if session.get(Tag, "missing") is None:
+        session.add(Tag(name="missing"))
+        session.flush()
+    for record_id in session.scalars(sa.select(Asset.id).where(Asset.content_id == content_id)):
+        if session.get(AssetTag, {"asset_id": record_id, "tag_name": "missing"}) is None:
+            session.add(AssetTag(asset_id=record_id, tag_name="missing", origin="automatic"))
+    session.flush()
+
+
+def unset_content_missing(session: Session, content_id: str) -> None:
+    content = session.get(AssetContent, content_id)
+    if content is None:
+        raise LookupError(content_id)
+    content.is_missing = False
+    session.execute(sa.delete(AssetTag).where(AssetTag.tag_name == "missing", AssetTag.asset_id.in_(sa.select(Asset.id).where(Asset.content_id == content_id))))
+    session.flush()

app/assets/mode.py

@@ -0,0 +1,22 @@
+"""Runtime hash-mode accessor."""
+
+from __future__ import annotations
+
+from typing import Protocol
+
+
+class _HashingArguments(Protocol):
+    enable_asset_hashing: bool
+
+
+_args: _HashingArguments | None = None
+
+
+def init(args: _HashingArguments) -> None:
+    global _args
+    _args = args
+
+
+def hashing_enabled() -> bool:
+    """Return whether startup enabled asset hashing."""
+    return bool(getattr(_args, "enable_asset_hashing", False))

app/assets/services/__init__.py

@@ -1,93 +1,42 @@
-from app.assets.services.asset_management import (
-    asset_exists,
-    delete_asset_reference,
-    get_asset_by_hash,
-    get_asset_detail,
-    list_assets_page,
-    get_preview_file_paths,
-    resolve_asset_for_download,
-    set_asset_preview,
-    update_asset_metadata,
-)
-from app.assets.services.bulk_ingest import (
-    BulkInsertResult,
-    batch_insert_seed_assets,
-    cleanup_unreferenced_assets,
-)
-from app.assets.services.file_utils import (
-    get_mtime_ns,
-    get_size_and_mtime_ns,
-    list_files_recursively,
-    verify_file_unchanged,
-)
+"""Re-export surface for app.assets.services — keeps routes.py and tests importable during the B-schema cutover."""
 from app.assets.services.ingest import (
     DependencyMissingError,
     HashMismatchError,
-    create_from_hash,
-    ingest_existing_file,
-    register_output_files,
     upload_from_temp_path,
+    create_from_hash,
+    register_file_in_place,
 )
-from app.assets.database.queries import (
-    AddTagsResult,
-    RemoveTagsResult,
-)
-from app.assets.services.schemas import (
-    AssetData,
-    AssetDetailResult,
-    AssetSummaryData,
-    DownloadResolutionResult,
-    IngestResult,
-    ListAssetsResult,
-    ReferenceData,
-    RegisterAssetResult,
-    TagUsage,
-    UploadResult,
-    UserMetadata,
+from app.assets.services.asset_management import (
+    get_asset_detail,
+    update_asset_metadata,
+    delete_asset_reference,
+    set_asset_preview,
+    asset_exists,
+    list_assets_page,
+    get_preview_file_paths,
+    resolve_asset_for_download,
 )
 from app.assets.services.tagging import (
     apply_tags,
-    list_tags,
     remove_tags,
+    list_tags,
 )
 
 __all__ = [
-    "AddTagsResult",
-    "AssetData",
-    "AssetDetailResult",
-    "AssetSummaryData",
-    "ReferenceData",
-    "BulkInsertResult",
     "DependencyMissingError",
-    "DownloadResolutionResult",
     "HashMismatchError",
-    "IngestResult",
-    "ListAssetsResult",
-    "RegisterAssetResult",
-    "RemoveTagsResult",
-    "TagUsage",
-    "UploadResult",
-    "UserMetadata",
-    "apply_tags",
-    "asset_exists",
-    "batch_insert_seed_assets",
+    "upload_from_temp_path",
     "create_from_hash",
-    "delete_asset_reference",
-    "get_asset_by_hash",
+    "register_file_in_place",
     "get_asset_detail",
-    "ingest_existing_file",
-    "register_output_files",
-    "get_mtime_ns",
-    "get_size_and_mtime_ns",
+    "update_asset_metadata",
+    "delete_asset_reference",
+    "set_asset_preview",
+    "asset_exists",
     "list_assets_page",
-    "list_files_recursively",
-    "list_tags",
-    "cleanup_unreferenced_assets",
-    "remove_tags",
     "get_preview_file_paths",
     "resolve_asset_for_download",
-    "set_asset_preview",
-    "update_asset_metadata",
-    "upload_from_temp_path",
-    "verify_file_unchanged",
+    "apply_tags",
+    "remove_tags",
+    "list_tags",
 ]

app/assets/services/ingest.py

@@ -25,7 +25,7 @@ from app.assets.database.queries import (
     set_reference_tags,
     update_asset_hash_and_mime,
     upsert_asset,
-    upsert_reference,
+    upsert_reference as _legacy_upsert_reference,  # wave-3-fixes: replaced by B-schema write paths in Wave 3
     validate_tags_exist,
 )
 from app.assets.helpers import get_utc_now, normalize_tags
@@ -85,7 +85,7 @@ def _ingest_file_from_path(
             mime_type=mime_type,
         )
 
-        ref_created, ref_updated = upsert_reference(
+        ref_created, ref_updated = _legacy_upsert_reference(  # wave-3-fixes: replaced in Wave 3
             session,
             asset_id=asset.id,
             file_path=locator,
@@ -176,10 +176,8 @@ def register_output_files(
         if not os.path.isfile(abs_path):
             continue
         try:
-            if ingest_existing_file(
-                abs_path, user_metadata=user_metadata, job_id=job_id
-            ):
-                registered += 1
+            register_output_file_b(abs_path, job_id=job_id)
+            registered += 1
         except Exception:
             logging.exception("Failed to register output: %s", abs_path)
     return registered
@@ -685,3 +683,62 @@ def create_from_hash(
         tags=result.tags,
         created_new=False,
     )
+
+
+_verification_queue: list[str] = []
+
+
+def register_output_file_b(abs_path: str, job_id: str | None = None):
+    from sqlalchemy import select
+
+    from app.assets import mode
+    from app.assets.database.models import AssetContent
+    from app.assets.database.queries.records import (
+        create_content,
+        create_record,
+        mark_content_missing,
+    )
+    from app.assets.services.snapshot_hash import snapshot_hash
+
+    locator = os.path.abspath(abs_path)
+    size_bytes, mtime_ns = get_size_and_mtime_ns(locator)
+    mime_type = mimetypes.guess_type(locator, strict=False)[0]
+    name, path_tags = get_name_and_tags_from_asset_path(locator)
+    with create_session() as session:
+        existing = session.scalars(
+            select(AssetContent).where(
+                AssetContent.path == locator, AssetContent.is_missing.is_(False)
+            )
+        ).first()
+        if existing is not None:
+            mark_content_missing(session, existing.id)
+        content_hash = None
+        if mode.hashing_enabled():
+            content_hash = snapshot_hash(locator)
+            if content_hash is None:
+                _verification_queue.append(locator)
+        content = create_content(
+            session, locator, content_hash, size_bytes, mtime_ns
+        )
+        record = create_record(
+            session,
+            content.id,
+            name,
+            mime_type=mime_type,
+            job_id=job_id,
+            loader_path=compute_loader_path(locator),
+            tags=path_tags,
+        )
+        session.commit()
+        record_id = record.id
+        record_content_id = record.content_id
+        record_job_id = record.job_id
+        record_name = record.name
+
+    from types import SimpleNamespace
+    return SimpleNamespace(
+        id=record_id,
+        content_id=record_content_id,
+        job_id=record_job_id,
+        name=record_name,
+    )

app/assets/services/lookup.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import os
+from collections.abc import Iterator
+from pathlib import Path
+
+import folder_paths
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.assets import mode
+from app.assets.database.models import Asset, AssetContent
+
+
+def is_temp_path(path: str) -> bool:
+    try:
+        temp_root = Path(os.path.abspath(folder_paths.get_temp_directory()))
+        candidate = Path(os.path.abspath(path))
+        return candidate.is_relative_to(temp_root)
+    except OSError:
+        return False
+
+
+def _stat_consistent(content: AssetContent) -> bool:
+    """Check file exists and stored stat values are consistent.
+
+    mtime_ns=None means "not yet measured" — always consistent.
+    size_bytes=0 with mtime_ns=None means "stub row" — always consistent.
+    """
+    try:
+        stat = os.stat(content.path)
+    except FileNotFoundError:
+        return False
+    if content.mtime_ns is not None and stat.st_mtime_ns != content.mtime_ns:
+        return False
+    # Only check size when mtime is also stored (fully enriched row)
+    if content.mtime_ns is not None and stat.st_size != content.size_bytes:
+        return False
+    return True
+
+
+def qualified_content_iterator(session: Session, hash: str) -> Iterator[AssetContent]:
+    rows = session.scalars(
+        select(AssetContent)
+        .where(AssetContent.hash == hash, AssetContent.is_missing.is_(False))
+        .order_by(AssetContent.created_at, AssetContent.id)
+    )
+    for row in rows:
+        if os.path.isfile(row.path) and _stat_consistent(row):
+            yield row
+
+
+def lookup_for_from_hash(session: Session, hash: str) -> AssetContent | None:
+    if not mode.hashing_enabled():
+        return None
+    return next(
+        (content for content in qualified_content_iterator(session, hash) if not is_temp_path(content.path)),
+        None,
+    )
+
+
+def lookup_for_upload_dedup(
+    session: Session, hash: str, name: str
+) -> Asset | AssetContent | None:
+    if not mode.hashing_enabled():
+        return None
+    first_content = None
+    for content in qualified_content_iterator(session, hash):
+        if is_temp_path(content.path):
+            continue
+        if first_content is None:
+            first_content = content
+        match = session.scalars(
+            select(Asset).where(Asset.content_id == content.id, Asset.name == name)
+        ).first()
+        if match is not None:
+            return match
+    return first_content
+
+
+def lookup_for_view(session: Session, hash: str) -> AssetContent | None:
+    return next(qualified_content_iterator(session, hash), None)

app/assets/services/snapshot_hash.py

@@ -0,0 +1,43 @@
+"""Stable file snapshot hashing."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+from blake3 import blake3
+
+
+@dataclass(frozen=True, slots=True)
+class _Snapshot:
+    dev: int
+    ino: int
+    mtime_ns: int
+    size: int
+
+
+def _snapshot(stat_result: os.stat_result) -> _Snapshot:
+    return _Snapshot(
+        dev=stat_result.st_dev,
+        ino=stat_result.st_ino,
+        mtime_ns=stat_result.st_mtime_ns,
+        size=stat_result.st_size,
+    )
+
+
+def snapshot_hash(path: str, chunk_size: int = 8 * 1024 * 1024) -> str | None:
+    """Return a BLAKE3 digest only when all path and descriptor snapshots match."""
+    pre_stat = _snapshot(os.stat(path))
+    hasher = blake3()
+    with open(path, "rb") as file:
+        open_stat = _snapshot(os.fstat(file.fileno()))
+        while chunk := file.read(chunk_size):
+            hasher.update(chunk)
+        post_hash_stat = _snapshot(os.fstat(file.fileno()))
+    try:
+        post_stat = _snapshot(os.stat(path))
+    except FileNotFoundError:
+        return None
+    if len({pre_stat, open_stat, post_hash_stat, post_stat}) != 1:
+        return None
+    return hasher.hexdigest()

app/database/db.py

@@ -134,6 +134,7 @@ def _init_memory_db(db_url):
 def _init_file_db(db_url):
     """Initialize a file-backed SQLite database using Alembic migrations."""
     db_path = get_db_path()
+    _acquire_file_lock(db_path)
     db_exists = os.path.exists(db_path)
 
     config = get_alembic_config()
@@ -168,6 +169,8 @@ def _init_file_db(db_url):
 
         try:
             command.upgrade(config, target_rev)
+            if backup_path:
+                os.remove(backup_path)
             logging.info(f"Database upgraded from {current_rev} to {target_rev}")
         except Exception as e:
             if backup_path:
@@ -177,11 +180,7 @@ def _init_file_db(db_url):
             logging.exception("Error upgrading database: ")
             raise e
 
-    # Acquire an OS-level file lock after migrations are complete.
-    # Alembic uses its own connection, so we must wait until it's done
-    # before locking — otherwise our own lock blocks the migration.
     conn.close()
-    _acquire_file_lock(db_path)
 
     global Session
     Session = sessionmaker(bind=engine)

tests-unit/app_test/test_migration_0007.py

@@ -0,0 +1,127 @@
+"""Tests specific to migration 0007 (record/content split)."""
+import os
+import sqlite3
+
+import pytest
+from alembic import command
+from alembic.config import Config
+
+_BASELINE_0006 = "0006_add_loader_path"
+
+
+def _make_config(db_path: str) -> Config:
+    root = os.path.join(os.path.dirname(__file__), "../..")
+    cfg = Config(os.path.abspath(os.path.join(root, "alembic.ini")))
+    cfg.set_main_option("script_location", os.path.abspath(os.path.join(root, "alembic_db")))
+    cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db_path}")
+    return cfg
+
+
+@pytest.fixture
+def db_at_0006(tmp_path):
+    db_path = str(tmp_path / "test.db")
+    cfg = _make_config(db_path)
+    command.upgrade(cfg, _BASELINE_0006)
+    yield cfg, db_path
+
+
+def test_0007_upgrade_from_0006(db_at_0006):
+    """Upgrade from 0006 to head succeeds; new tables present, old gone."""
+    cfg, db_path = db_at_0006
+    command.upgrade(cfg, "head")
+    with sqlite3.connect(db_path) as conn:
+        tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
+    assert "assets" in tables
+    assert "asset_contents" in tables
+    assert "asset_system_state" in tables
+    assert "asset_references" not in tables
+
+
+def test_0007_schema_has_expected_columns(db_at_0006):
+    """After upgrade, key columns exist on new tables."""
+    cfg, db_path = db_at_0006
+    command.upgrade(cfg, "head")
+    with sqlite3.connect(db_path) as conn:
+        asset_cols = {r[1] for r in conn.execute("PRAGMA table_info(assets)")}
+        content_cols = {r[1] for r in conn.execute("PRAGMA table_info(asset_contents)")}
+    assert {"id", "content_id", "name", "loader_path", "updated_at", "last_access_time"} <= asset_cols
+    assert {"id", "hash", "path", "is_missing", "mtime_ns"} <= content_cols
+
+
+def test_0007_downgrade_restores_0006_schema(db_at_0006):
+    """Downgrade from head back to 0006 restores asset_references."""
+    cfg, db_path = db_at_0006
+    command.upgrade(cfg, "head")
+    command.downgrade(cfg, _BASELINE_0006)
+    with sqlite3.connect(db_path) as conn:
+        tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
+    assert "asset_references" in tables
+    assert "asset_contents" not in tables
+
+
+def test_0007_invariants_on_migrated_db(db_at_0006):
+    """After upgrade, schema invariants hold."""
+    cfg, db_path = db_at_0006
+    command.upgrade(cfg, "head")
+    with sqlite3.connect(db_path) as conn:
+        conn.execute("PRAGMA foreign_keys = ON")
+        # Insert a content row and a record
+        conn.execute(
+            "INSERT INTO asset_contents(id, path, is_missing, size_bytes, created_at) "
+            "VALUES ('c1', '/tmp/f1', 0, 0, '2024-01-01')"
+        )
+        conn.execute(
+            "INSERT INTO assets(id, content_id, name, created_at, updated_at) "
+            "VALUES ('a1', 'c1', 'test', '2024-01-01', '2024-01-01')"
+        )
+        conn.commit()
+
+        # Duplicate live path should fail (partial unique)
+        with pytest.raises(sqlite3.IntegrityError):
+            conn.execute(
+                "INSERT INTO asset_contents(id, path, is_missing, size_bytes, created_at) "
+                "VALUES ('c2', '/tmp/f1', 0, 0, '2024-01-01')"
+            )
+            conn.commit()
+        conn.rollback()
+
+        # Live + missing same path should succeed
+        conn.execute(
+            "INSERT INTO asset_contents(id, path, is_missing, size_bytes, created_at) "
+            "VALUES ('c3', '/tmp/f1', 1, 0, '2024-01-01')"
+        )
+        conn.commit()
+
+        # Equal-hash rows should coexist (no hash UNIQUE)
+        conn.execute("UPDATE asset_contents SET hash='abc123' WHERE id='c1'")
+        conn.execute(
+            "INSERT INTO asset_contents(id, path, hash, is_missing, size_bytes, created_at) "
+            "VALUES ('c4', '/tmp/f2', 'abc123', 0, 0, '2024-01-01')"
+        )
+        conn.commit()
+
+
+def test_0007_orm_parity(db_at_0006, tmp_path):
+    """Base.metadata.create_all produces same table names as alembic upgrade."""
+    from sqlalchemy import create_engine, inspect
+
+    from app.database.models import Base
+
+    cfg, db_path = db_at_0006
+    command.upgrade(cfg, "head")
+    with sqlite3.connect(db_path) as conn:
+        alembic_tables = {
+            r[0]
+            for r in conn.execute(
+                "SELECT name FROM sqlite_master WHERE type='table' "
+                "AND name NOT LIKE 'alembic%' AND name NOT LIKE 'sqlite%'"
+            )
+        }
+
+    orm_db = str(tmp_path / "orm.db")
+    engine = create_engine(f"sqlite:///{orm_db}")
+    Base.metadata.create_all(engine)
+    inspector = inspect(engine)
+    orm_tables = set(inspector.get_table_names())
+
+    assert alembic_tables == orm_tables, f"Mismatch: alembic={alembic_tables}, orm={orm_tables}"

tests-unit/assets_test/queries/conftest.py

@@ -5,6 +5,18 @@ from sqlalchemy.orm import Session
 from app.assets.database.models import Base
 
 
+@pytest.fixture(scope="session", autouse=True)
+def assert_asset_metadata_tables():
+    assert set(Base.metadata.tables) == {
+        "assets",
+        "asset_contents",
+        "asset_meta",
+        "asset_tags",
+        "tags",
+        "asset_system_state",
+    }
+
+
 @pytest.fixture
 def session():
     """In-memory SQLite session for fast unit tests."""

tests-unit/assets_test/queries/test_lookup.py

@@ -0,0 +1,121 @@
+"""Tests for the B-schema hash lookup policies."""
+from datetime import datetime
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy import create_engine, update
+from sqlalchemy.orm import Session
+
+import app.assets.mode as mode_module
+from app.assets.database.models import AssetContent
+from app.assets.database.queries.records import create_content, create_record
+from app.assets.services.lookup import (
+    is_temp_path as _is_temp_path,
+    lookup_for_from_hash,
+    lookup_for_upload_dedup,
+    lookup_for_view,
+)
+from app.database.models import Base
+
+
+@pytest.fixture
+def session(tmp_path):
+    engine = create_engine(
+        f"sqlite:///{tmp_path}/test.db", connect_args={"check_same_thread": False}
+    )
+    Base.metadata.create_all(engine)
+    with Session(engine) as sess:
+        yield sess
+
+
+@pytest.fixture(autouse=True)
+def enable_hashing():
+    class FakeArgs:
+        enable_asset_hashing = True
+
+    mode_module.init(FakeArgs())
+    yield
+    mode_module.init(None)
+
+
+def _make_file(tmp_path, name: str, content: bytes = b"bytes") -> str:
+    p = tmp_path / name
+    p.write_bytes(content)
+    return str(p)
+
+
+def test_temp_only_match_from_hash_returns_none(session, tmp_path):
+    f = _make_file(tmp_path, "f.png")
+    create_content(session, path=f, hash="abc123")
+    session.commit()
+    with patch("app.assets.services.lookup.is_temp_path", return_value=True):
+        result = lookup_for_from_hash(session, "abc123")
+    assert result is None
+
+
+def test_temp_only_match_view_still_serves(session, tmp_path):
+    f = _make_file(tmp_path, "f2.png")
+    create_content(session, path=f, hash="abc123")
+    session.commit()
+    with patch("app.assets.services.lookup.is_temp_path", return_value=True):
+        result = lookup_for_view(session, "abc123")
+    assert result is not None
+
+
+def test_sibling_prefix_not_temp(tmp_path):
+    with patch("folder_paths.get_temp_directory", return_value=str(tmp_path / "temp")):
+        assert not _is_temp_path(str(tmp_path / "temp-other" / "f.png"))
+        assert _is_temp_path(str(tmp_path / "temp" / "f.png"))
+
+
+def test_off_mode_from_hash_returns_none(session, tmp_path):
+    class FakeArgs:
+        enable_asset_hashing = False
+
+    mode_module.init(FakeArgs())
+    f = _make_file(tmp_path, "f3.png")
+    create_content(session, path=f, hash="abc123")
+    session.commit()
+    result = lookup_for_from_hash(session, "abc123")
+    assert result is None
+
+
+def test_off_mode_dedup_returns_none(session, tmp_path):
+    class FakeArgs:
+        enable_asset_hashing = False
+
+    mode_module.init(FakeArgs())
+    f = _make_file(tmp_path, "f4.png")
+    create_content(session, path=f, hash="abc123")
+    session.commit()
+    result = lookup_for_upload_dedup(session, "abc123", "test.png")
+    assert result is None
+
+
+def test_stale_older_newer_live_returns_newer(session, tmp_path):
+    """Oldest candidate with missing file is skipped; newer live candidate returned."""
+    f_newer = _make_file(tmp_path, "newer.png")
+    old_time = datetime(2020, 1, 1)
+    new_time = datetime(2024, 1, 1)
+    c_old = create_content(session, path="/nonexistent/old.png", hash="xyz")
+    c_new = create_content(session, path=f_newer, hash="xyz")
+    session.execute(update(AssetContent).where(AssetContent.id == c_old.id).values(created_at=old_time))
+    session.execute(update(AssetContent).where(AssetContent.id == c_new.id).values(created_at=new_time))
+    session.commit()
+
+    result = lookup_for_from_hash(session, "xyz")
+    assert result is not None
+    assert result.id == c_new.id
+
+
+def test_dedup_returns_matching_name_entity(session, tmp_path):
+    """Upload dedup returns the entity with the matching name, not just any entity."""
+    f = _make_file(tmp_path, "match.png")
+    content = create_content(session, path=f, hash="dup")
+    record = create_record(session, content_id=content.id, name="match.png")
+    session.commit()
+
+    result = lookup_for_upload_dedup(session, "dup", "match.png")
+    assert result is not None
+    assert hasattr(result, "name"), "Should return an Asset record"
+    assert result.name == "match.png"

tests-unit/assets_test/queries/test_records.py

@@ -0,0 +1,105 @@
+"""Tests for the B-schema record/content query layer."""
+import pytest
+from sqlalchemy import create_engine, select, update
+from sqlalchemy.orm import Session
+
+from app.assets.database.models import Asset, AssetContent
+from app.assets.database.queries.records import (
+    create_content,
+    create_record,
+    delete_record,
+    get_record_by_id,
+    list_records_page,
+    mark_content_missing,
+    unset_content_missing,
+)
+from app.database.models import Base
+
+
+@pytest.fixture
+def session():
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
+    Base.metadata.create_all(engine)
+    with Session(engine) as sess:
+        yield sess
+
+
+def test_listing_includes_missing_by_default(session):
+    """Missing-content records appear in unfiltered listings."""
+    content = create_content(session, path="/tmp/f1")
+    record = create_record(session, content_id=content.id, name="test")
+    mark_content_missing(session, content.id)
+    session.commit()
+
+    results, _ = list_records_page(session)
+    assert any(r.id == record.id for r in results), "Missing record should appear in default listing"
+
+
+def test_listing_excludes_missing_with_filter(session):
+    """exclude_tags=['missing'] hides missing-content records."""
+    content = create_content(session, path="/tmp/f2")
+    record = create_record(session, content_id=content.id, name="test2")
+    mark_content_missing(session, content.id)
+    session.commit()
+
+    results, _ = list_records_page(session, exclude_tags=["missing"])
+    assert not any(r.id == record.id for r in results), "Missing record should be excluded"
+
+
+def test_preview_cleanup_on_delete(session):
+    """Deleting the last record referencing a preview deletes the preview record but not its content."""
+    preview_content = create_content(session, path="/tmp/preview")
+    preview_record = create_record(session, content_id=preview_content.id, name="preview")
+    content1 = create_content(session, path="/tmp/f3")
+    content2 = create_content(session, path="/tmp/f4")
+    r1 = create_record(session, content_id=content1.id, name="r1")
+    r2 = create_record(session, content_id=content2.id, name="r2")
+
+    # Both records reference the same preview
+    session.execute(update(Asset).where(Asset.id == r1.id).values(preview_id=preview_record.id))
+    session.execute(update(Asset).where(Asset.id == r2.id).values(preview_id=preview_record.id))
+    session.commit()
+
+    # Delete r1 — preview should survive (r2 still references it)
+    delete_record(session, r1.id)
+    session.commit()
+    assert get_record_by_id(session, preview_record.id) is not None, "Preview should survive"
+
+    # Delete r2 — preview should be deleted (no more references)
+    delete_record(session, r2.id)
+    session.commit()
+    assert get_record_by_id(session, preview_record.id) is None, "Preview should be deleted"
+
+    # Preview content row should still exist (D-3 floor)
+    content_row = session.execute(
+        select(AssetContent).where(AssetContent.id == preview_content.id)
+    ).scalar_one_or_none()
+    assert content_row is not None, "Preview content row should survive (D-3 floor)"
+
+
+def test_concurrent_create_content_same_path(tmp_path):
+    """Concurrent inserts for the same live path: exactly one live row wins."""
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import Session
+
+    db_path = str(tmp_path / "concurrent.db")
+    engine = create_engine(f"sqlite:///{db_path}", connect_args={"check_same_thread": False})
+    Base.metadata.create_all(engine)
+
+    with Session(engine) as s1, Session(engine) as s2:
+        c1 = create_content(s1, path="/tmp/shared")
+        s1.commit()
+        # Second session: same path — should get the winner back
+        c2 = create_content(s2, path="/tmp/shared")
+        s2.commit()
+
+    with Session(engine) as s:
+        live_rows = list(
+            s.execute(
+                select(AssetContent).where(
+                    AssetContent.path == "/tmp/shared",
+                    AssetContent.is_missing.is_(False),
+                )
+            ).scalars()
+        )
+    assert len(live_rows) == 1, f"Expected exactly one live row, got {len(live_rows)}"

tests-unit/assets_test/services/test_ingest_b.py

@@ -0,0 +1,89 @@
+"""Tests for the B-schema ingest service (register_output_file_b)."""
+import os
+
+import pytest
+from sqlalchemy import select
+
+import app.assets.mode as mode_module
+from app.assets.database.models import Asset, AssetContent
+
+
+@pytest.fixture(autouse=True)
+def hashing_off():
+    class FakeArgs:
+        enable_asset_hashing = False
+
+    mode_module.init(FakeArgs())
+    yield
+    mode_module.init(None)
+
+
+def test_new_path_save_off_mode_hash_null(mock_create_session):
+    """New file registered in off mode: content row has hash=NULL."""
+    import folder_paths
+    from app.assets.services.ingest import register_output_file_b
+
+    output_dir = folder_paths.get_output_directory()
+    os.makedirs(output_dir, exist_ok=True)
+    f = os.path.join(output_dir, "test_ingest_b_new.png")
+    with open(f, "wb") as fh:
+        fh.write(b"pixels")
+
+    try:
+        record = register_output_file_b(f, job_id="job1")
+        record_id = record.id
+        with mock_create_session() as session:
+            content = session.execute(
+                select(AssetContent).where(AssetContent.path == os.path.abspath(f))
+            ).scalar_one()
+            assert content.hash is None
+            assert content.is_missing is False
+            asset = session.execute(
+                select(Asset).where(Asset.id == record_id)
+            ).scalar_one()
+            assert asset.job_id == "job1"
+    finally:
+        if os.path.exists(f):
+            os.unlink(f)
+
+
+def test_overwrite_at_live_path_marks_old_missing(mock_create_session):
+    """Overwriting a live path marks the old content row missing; old record's job_id unchanged."""
+    import folder_paths
+    from app.assets.services.ingest import register_output_file_b
+
+    output_dir = folder_paths.get_output_directory()
+    os.makedirs(output_dir, exist_ok=True)
+    f = os.path.join(output_dir, "test_ingest_b_overwrite.png")
+
+    try:
+        with open(f, "wb") as fh:
+            fh.write(b"v1")
+        r1 = register_output_file_b(f, job_id="job1")
+        old_content_id = r1.content_id
+        old_record_id = r1.id
+
+        with open(f, "wb") as fh:
+            fh.write(b"v2")
+        r2 = register_output_file_b(f, job_id="job2")
+        new_record_id = r2.id
+
+        with mock_create_session() as session:
+            old_content = session.execute(
+                select(AssetContent).where(AssetContent.id == old_content_id)
+            ).scalar_one()
+            assert old_content.is_missing is True, "Old content should be marked missing"
+
+            old_record = session.execute(
+                select(Asset).where(Asset.id == old_record_id)
+            ).scalar_one()
+            assert old_record.job_id == "job1", "Old record's job_id must not be mutated"
+
+            new_record = session.execute(
+                select(Asset).where(Asset.id == new_record_id)
+            ).scalar_one()
+            assert new_record.job_id == "job2"
+            assert new_record_id != old_record_id
+    finally:
+        if os.path.exists(f):
+            os.unlink(f)

tests-unit/assets_test/services/test_snapshot_hash.py

@@ -0,0 +1,75 @@
+import builtins
+import os
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+from blake3 import blake3
+
+from app.assets.services.snapshot_hash import snapshot_hash
+
+
+class _MutatingReader:
+    def __init__(self, file, mutate: Callable[[], None]) -> None:
+        self._file = file
+        self._mutate = mutate
+        self._did_mutate = False
+
+    def __enter__(self):
+        self._file.__enter__()
+        return self
+
+    def __exit__(self, *args):
+        return self._file.__exit__(*args)
+
+    def fileno(self) -> int:
+        return self._file.fileno()
+
+    def read(self, size: int = -1) -> bytes:
+        result = self._file.read(size)
+        if not self._did_mutate:
+            self._did_mutate = True
+            self._mutate()
+        return result
+
+
+def test_snapshot_hash_returns_digest_for_quiescent_file(tmp_path: Path) -> None:
+    payload = b"quiescent" * 1024
+    path = tmp_path / "asset.bin"
+    path.write_bytes(payload)
+
+    assert snapshot_hash(str(path), chunk_size=64) == blake3(payload).hexdigest()
+
+
+@pytest.mark.parametrize("mutation", ["bytes", "replace", "unlink", "truncate", "append"])
+def test_snapshot_hash_returns_none_when_file_drifts(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mutation: str
+) -> None:
+    path = tmp_path / "asset.bin"
+    path.write_bytes(b"original" * 1024)
+    original_open = builtins.open
+
+    def mutate() -> None:
+        match mutation:
+            case "bytes":
+                path.write_bytes(b"changed" * 1024)
+            case "replace":
+                replacement = tmp_path / "replacement.bin"
+                replacement.write_bytes(b"replacement")
+                os.replace(replacement, path)
+            case "unlink":
+                path.unlink()
+            case "truncate":
+                path.write_bytes(b"")
+            case "append":
+                with original_open(path, "ab") as output:
+                    output.write(b"more")
+            case unreachable:
+                raise AssertionError(f"unexpected mutation {unreachable}")
+
+    def open_with_mutation(*args, **kwargs):
+        return _MutatingReader(original_open(*args, **kwargs), mutate)
+
+    monkeypatch.setattr(builtins, "open", open_with_mutation)
+
+    assert snapshot_hash(str(path), chunk_size=64) is None