diff --git a/app/assets/api/routes.py b/app/assets/api/routes.py
index 80f331f3e..06fa65650 100644
--- a/app/assets/api/routes.py
+++ b/app/assets/api/routes.py
@@ -741,7 +741,6 @@ async def upload_asset(request: web.Request) -> web.Response:
                 name=spec.name or (spec.hash.split(":", 1)[1]),
                 tags=spec.tags,
                 user_metadata=spec.user_metadata or {},
-                tenant_id=tenant_id,
                 mime_type=spec.mime_type,
                 preview_id=spec.preview_id,
             )
@@ -756,7 +755,6 @@ async def upload_asset(request: web.Request) -> web.Response:
                 tags=spec.tags,
                 user_metadata=spec.user_metadata or {},
                 client_filename=parsed.file_client_name,
-                tenant_id=tenant_id,
                 expected_hash=spec.hash,
                 mime_type=spec.mime_type,
                 preview_id=spec.preview_id,
@@ -972,6 +970,7 @@ async def delete_asset_tags(request: web.Request) -> web.Response:
             removed=result.removed,
             not_present=result.not_present,
             total_tags=result.total_tags,
+            protected=result.protected,
         )
     except PermissionError as pe:
         return _build_error_response(403, "FORBIDDEN", str(pe), {"id": reference_id})
diff --git a/app/assets/api/schemas_out.py b/app/assets/api/schemas_out.py
index 849f5f94b..c308de1b9 100644
--- a/app/assets/api/schemas_out.py
+++ b/app/assets/api/schemas_out.py
@@ -78,6 +78,7 @@ class TagsRemove(BaseModel):
     removed: list[str] = Field(default_factory=list)
     not_present: list[str] = Field(default_factory=list)
     total_tags: list[str] = Field(default_factory=list)
+    protected: list[str] = Field(default_factory=list)
 
 
 class TagHistogram(BaseModel):
diff --git a/app/assets/database/queries/common.py b/app/assets/database/queries/common.py
deleted file mode 100644
index 70d496971..000000000
--- a/app/assets/database/queries/common.py
+++ /dev/null
@@ -1,17 +0,0 @@
-"""Shared utilities for database query modules (B-schema stub)."""
-
-
-MAX_BIND_PARAMS = 800
-
-
-def calculate_rows_per_statement(cols: int) -> int:
-    return max(1, MAX_BIND_PARAMS // max(1, cols))
-
-
-def iter_chunks(seq, n: int):
-    for index in range(0, len(seq), n):
-        yield seq[index : index + n]
-
-
-def iter_row_chunks(rows, cols_per_row: int):
-    yield from iter_chunks(rows, calculate_rows_per_statement(cols_per_row))
diff --git a/app/assets/database/queries/tags.py b/app/assets/database/queries/tags.py
index a35254407..015ea98e9 100644
--- a/app/assets/database/queries/tags.py
+++ b/app/assets/database/queries/tags.py
@@ -2,10 +2,9 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 
-from typing import Iterable, Sequence
+from typing import Sequence
 
 from sqlalchemy import func, select
-from sqlalchemy.dialects import sqlite
 from sqlalchemy.orm import Session
 
 from app.assets.database.models import (
@@ -17,7 +16,7 @@ from app.assets.database.models import (
 from app.assets.database.queries.records import (
     build_record_tag_filter_clauses,
 )
-from app.assets.helpers import escape_sql_like_string, normalize_tags
+from app.assets.helpers import escape_sql_like_string
 
 
 @dataclass(frozen=True)
@@ -38,30 +37,6 @@ class RemoveTagsResult:
     protected: list[str] = field(default_factory=list)
 
 
-def validate_tags_exist(session: Session, tags: list[str]) -> None:
-    """Raise ValueError if any of the given tag names do not exist."""
-    existing_tag_names = set(
-        name
-        for (name,) in session.execute(select(Tag.name).where(Tag.name.in_(tags))).all()
-    )
-    missing = [t for t in tags if t not in existing_tag_names]
-    if missing:
-        raise ValueError(f"Unknown tags: {missing}")
-
-
-def ensure_tags_exist(session: Session, names: Iterable[str]) -> None:
-    wanted = normalize_tags(list(names))
-    if not wanted:
-        return
-    rows = [{"name": n} for n in list(dict.fromkeys(wanted))]
-    ins = (
-        sqlite.insert(Tag)
-        .values(rows)
-        .on_conflict_do_nothing(index_elements=[Tag.name])
-    )
-    session.execute(ins)
-
-
 def list_tags_with_usage(
     session: Session,
     prefix: str | None = None,
diff --git a/app/assets/helpers.py b/app/assets/helpers.py
index 2e964162b..ba0b324e2 100644
--- a/app/assets/helpers.py
+++ b/app/assets/helpers.py
@@ -1,26 +1,4 @@
-import os
 from datetime import datetime, timezone
-from typing import Sequence
-
-
-def select_live_path(states: Sequence) -> str:
-    """
-    Return the best on-disk path among cache states:
-      1) Prefer a path that exists with pending_verification == False (already verified).
-      2) Otherwise, pick the first path that exists.
-      3) Otherwise return empty string.
-    """
-    alive = [
-        s
-        for s in states
-        if getattr(s, "file_path", None) and os.path.isfile(s.file_path)
-    ]
-    if not alive:
-        return ""
-    for s in alive:
-        if not getattr(s, "pending_verification", False):
-            return s.file_path
-    return alive[0].file_path
 
 
 def escape_sql_like_string(s: str, escape: str = "!") -> tuple[str, str]:
diff --git a/app/assets/mode.py b/app/assets/mode.py
index 7ea2705c8..0a91ef66a 100644
--- a/app/assets/mode.py
+++ b/app/assets/mode.py
@@ -19,4 +19,9 @@ def init(args: _HashingArguments) -> None:
 
 def hashing_enabled() -> bool:
     """Return whether startup enabled asset hashing."""
-    return bool(getattr(_args, "enable_asset_hashing", False))
+    if _args is None:
+        raise RuntimeError(
+            "app.assets.mode.init() was not called before hashing_enabled(); "
+            "hash-mode state is uninitialised"
+        )
+    return bool(_args.enable_asset_hashing)
diff --git a/app/assets/scanner.py b/app/assets/scanner.py
index 408ca493e..930f04af7 100644
--- a/app/assets/scanner.py
+++ b/app/assets/scanner.py
@@ -130,13 +130,11 @@ def sync_references_with_filesystem(
     session,
     root: RootType,
     collect_existing_paths: bool = False,
-    update_missing_tags: bool = False,
 ) -> set[str] | None:
     return sync_prefixes_with_filesystem(
         session,
         get_scan_prefixes_for_root(root),
         collect_existing_paths=collect_existing_paths,
-        update_missing_tags=update_missing_tags,
     )
 
 
@@ -144,14 +142,8 @@ def sync_prefixes_with_filesystem(
     session: Session,
     prefixes: list[str],
     collect_existing_paths: bool = False,
-    update_missing_tags: bool = False,
 ) -> set[str] | None:
-    """Mark disappeared content missing and return live filesystem paths.
-
-    ``update_missing_tags`` remains for callers from the pre-B scanner API. The B
-    schema has one authoritative missing state on content, so its record-tag
-    projection is always maintained rather than conditionally enabled.
-    """
+    """Mark disappeared content missing and return live filesystem paths."""
     if not prefixes:
         return set() if collect_existing_paths else None
 
@@ -193,7 +185,6 @@ def sync_root_safely(root: RootType) -> set[str]:
                 sess,
                 root,
                 collect_existing_paths=True,
-                update_missing_tags=True,
             )
             sess.commit()
             return survivors or set()
diff --git a/app/assets/services/asset_management.py b/app/assets/services/asset_management.py
index c775e52ff..347546c10 100644
--- a/app/assets/services/asset_management.py
+++ b/app/assets/services/asset_management.py
@@ -52,9 +52,7 @@ def _record_to_detail_result(session, record) -> AssetDetailResult:
 
 def get_asset_detail(
     reference_id: str,
-    tenant_id: str = "",
 ) -> AssetDetailResult | None:
-    del tenant_id
     with create_session() as session:
         record = get_record_by_id(session, reference_id)
         if record is None:
@@ -68,11 +66,9 @@ def update_asset_metadata(
     tags: Sequence[str] | None = None,
     user_metadata: UserMetadata = None,
     tag_origin: str = "manual",
-    tenant_id: str = "",
     mime_type: str | None = None,
     preview_id: str | None = None,
 ) -> AssetDetailResult:
-    del tenant_id
     with create_session() as session:
         record = get_record_by_id(session, reference_id)
         if record is None:
@@ -132,11 +128,8 @@ def update_asset_metadata(
 
 def delete_asset_reference(
     reference_id: str,
-    tenant_id: str = "",
-    delete_content_if_orphan: bool = True,
 ) -> bool:
     """Hard-delete an asset record. Content rows and files are untouched (D-3 floor)."""
-    del tenant_id, delete_content_if_orphan
     with create_session() as session:
         if get_record_by_id(session, reference_id) is None:
             return False
@@ -156,7 +149,6 @@ def asset_exists(asset_hash: str) -> bool:
 
 def resolve_hash_to_path(
     asset_hash: str,
-    tenant_id: str = "",
 ) -> DownloadResolutionResult | None:
     """Resolve a blake3 hash to an on-disk file path via lookup_for_view.
 
@@ -165,7 +157,6 @@ def resolve_hash_to_path(
     temp content returns None. Updates last_access_time on every record pointing
     at the served content.
     """
-    del tenant_id
     try:
         canonical = validate_blake3_hash(asset_hash)
     except ValueError:
@@ -213,9 +204,7 @@ def get_preview_file_paths(preview_ids: list[str]) -> dict[str, str]:
 
 def resolve_asset_for_download(
     reference_id: str,
-    tenant_id: str = "",
 ) -> DownloadResolutionResult:
-    del tenant_id
     with create_session() as session:
         record = get_record_by_id(session, reference_id)
         if record is None:
diff --git a/app/assets/services/ingest.py b/app/assets/services/ingest.py
index 406a43202..469c57736 100644
--- a/app/assets/services/ingest.py
+++ b/app/assets/services/ingest.py
@@ -5,7 +5,7 @@ import os
 from types import SimpleNamespace
 from typing import Any, Sequence
 
-from sqlalchemy import event, func, select
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app.assets import mode
@@ -243,7 +243,6 @@ def upload_from_temp_path(
     tags: list[str] | None = None,
     user_metadata: dict | None = None,
     client_filename: str | None = None,
-    tenant_id: str = "",
     expected_hash: str | None = None,
     mime_type: str | None = None,
     preview_id: str | None = None,
@@ -350,7 +349,6 @@ def register_file_in_place(
     abs_path: str,
     name: str,
     tags: list[str],
-    tenant_id: str = "",
     mime_type: str | None = None,
 ) -> UploadResult:
     """Register an already-saved file in the asset database without moving it.
@@ -428,7 +426,6 @@ def create_from_hash(
     name: str,
     tags: list[str] | None = None,
     user_metadata: dict | None = None,
-    tenant_id: str = "",
     mime_type: str | None = None,
     preview_id: str | None = None,
 ) -> UploadResult | None:
@@ -474,12 +471,6 @@ def register_cached_output(abs_path: str, job_id: str | None = None):
     locator = os.path.abspath(abs_path)
     try:
         with create_session() as session:
-            update_count = [0]
-
-            @event.listens_for(session, "after_bulk_update")
-            def _count_update(update_context):
-                update_count[0] += 1
-
             existing = session.scalars(
                 select(AssetContent).where(
                     AssetContent.path == locator, AssetContent.is_missing.is_(False)
@@ -522,14 +513,6 @@ def register_cached_output(abs_path: str, job_id: str | None = None):
                     tags=path_tags,
                     system_metadata=system_metadata,
                 )
-                if update_count[0] != 0:
-                    logging.error(
-                        "Cached save must not UPDATE any row; got %d for %s; discarding",
-                        update_count[0],
-                        locator,
-                    )
-                    session.rollback()
-                    return None
                 session.commit()
             except Exception:
                 session.rollback()
diff --git a/app/assets/services/lookup.py b/app/assets/services/lookup.py
index bbff93f98..0efab6a0f 100644
--- a/app/assets/services/lookup.py
+++ b/app/assets/services/lookup.py
@@ -53,10 +53,7 @@ def qualified_content_iterator(session: Session, hash: str) -> Iterator[AssetCon
 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,
-    )
+    return next(qualified_content_iterator(session, hash), None)
 
 
 def lookup_for_upload_dedup(
@@ -64,8 +61,6 @@ def lookup_for_upload_dedup(
 ) -> Asset | AssetContent | 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(
diff --git a/app/assets/services/tagging.py b/app/assets/services/tagging.py
index a441c963c..b335b24b0 100644
--- a/app/assets/services/tagging.py
+++ b/app/assets/services/tagging.py
@@ -18,9 +18,7 @@ def apply_tags(
     reference_id: str,
     tags: list[str],
     origin: str = "manual",
-    tenant_id: str = "",
 ) -> AddTagsResult:
-    del tenant_id
     with create_session() as session:
         if session.get(Asset, reference_id) is None:
             raise ValueError(f"Asset {reference_id} not found")
@@ -64,9 +62,7 @@ def apply_tags(
 def remove_tags(
     reference_id: str,
     tags: list[str],
-    tenant_id: str = "",
 ) -> RemoveTagsResult:
-    del tenant_id
     with create_session() as session:
         if session.get(Asset, reference_id) is None:
             raise ValueError(f"Asset {reference_id} not found")
@@ -125,9 +121,7 @@ def list_tags(
     offset: int = 0,
     order: str = "count_desc",
     include_zero: bool = True,
-    tenant_id: str = "",
 ) -> tuple[list[TagUsage], int]:
-    del tenant_id
     limit = max(1, min(1000, limit))
     offset = max(0, offset)
 
@@ -145,7 +139,6 @@ def list_tags(
 
 
 def list_tag_histogram(
-    tenant_id: str = "",
     include_tags: Sequence[str] | None = None,
     exclude_tags: Sequence[str] | None = None,
     name_contains: str | None = None,
@@ -153,7 +146,6 @@ def list_tag_histogram(
     # Appended last so pre-existing positional callers keep binding correctly.
     any_tags: Sequence[str] | None = None,
 ) -> dict[str, int]:
-    del tenant_id
     with create_session() as session:
         return list_tag_counts_for_filtered_assets(
             session,
diff --git a/tests-unit/assets_test/conftest.py b/tests-unit/assets_test/conftest.py
index 75bdcd979..b54b4a56b 100644
--- a/tests-unit/assets_test/conftest.py
+++ b/tests-unit/assets_test/conftest.py
@@ -215,8 +215,9 @@ def _post_multipart_asset(
 @pytest.fixture
 def make_asset_bytes() -> Callable[[str, int], bytes]:
     # Salt content per test so it never collides with assets left over from
-    # earlier tests. Delete is now always a soft delete (content is preserved),
-    # so the suite can no longer rely on hard-deleting content for isolation.
+    # earlier tests. Delete hard-deletes the record but preserves content
+    # (content rows and files are untouched), so the suite cannot rely on delete
+    # removing content for isolation.
     # Deterministic within a test: the same (name, size) yields the same bytes.
     salt = uuid.uuid4().bytes
 
@@ -262,8 +263,9 @@ def seeded_asset(request: pytest.FixtureRequest, http: requests.Session, api_bas
         tags = ["models", "model_type:checkpoints", "unit-tests", "alpha"]
     meta = {"purpose": "test", "epoch": 1, "flags": ["x", "y"], "nullable": None}
     # Unique content per test so the seed always creates a fresh asset (201).
-    # Delete is now always a soft delete, so content from a prior test survives
-    # and would otherwise dedup this upload into an existing asset (200).
+    # Delete preserves content (only the record is hard-deleted), so content
+    # from a prior test survives and would otherwise dedup this upload into an
+    # existing asset (200).
     content = uuid.uuid4().bytes + b"A" * (4096 - 16)
     files = {"file": (name, content, "application/octet-stream")}
     form_data = {
diff --git a/tests-unit/assets_test/services/conftest.py b/tests-unit/assets_test/services/conftest.py
index 09a51e0e8..41c222907 100644
--- a/tests-unit/assets_test/services/conftest.py
+++ b/tests-unit/assets_test/services/conftest.py
@@ -7,6 +7,7 @@ import pytest
 from sqlalchemy import create_engine, event
 from sqlalchemy.orm import Session, Session as SASession
 
+from app.assets import mode
 from app.assets.database.models import Base
 
 
@@ -16,6 +17,20 @@ def autoclean_unit_test_assets():
     yield
 
 
+@pytest.fixture(autouse=True)
+def initialised_hash_mode():
+    # mode._args is process-global and leaks across tests; hashing_enabled() now
+    # raises when it was never initialised. Give every service test a determinate
+    # hashing-off baseline. Tests needing hashing on override via their own
+    # fixture or by patching mode.hashing_enabled after this runs.
+    class _HashingOff:
+        enable_asset_hashing = False
+
+    mode.init(_HashingOff())
+    yield
+    mode.init(None)
+
+
 @pytest.fixture
 def db_engine():
     """In-memory SQLite engine for fast unit tests."""
diff --git a/tests-unit/assets_test/services/test_tag_protection.py b/tests-unit/assets_test/services/test_tag_protection.py
index 9901da76d..5bd8f4168 100644
--- a/tests-unit/assets_test/services/test_tag_protection.py
+++ b/tests-unit/assets_test/services/test_tag_protection.py
@@ -1,3 +1,4 @@
+import json
 from types import SimpleNamespace
 
 import pytest
@@ -66,7 +67,7 @@ async def test_other_tags_unaffected(monkeypatch):
     def remove_tags(**kwargs):
         calls.append(("remove", kwargs["tags"]))
         return SimpleNamespace(
-            removed=kwargs["tags"], not_present=[], total_tags=[]
+            removed=kwargs["tags"], not_present=[], total_tags=[], protected=[]
         )
 
     monkeypatch.setattr(routes, "apply_tags", apply_tags)
@@ -80,3 +81,24 @@ async def test_other_tags_unaffected(monkeypatch):
     assert add_response.status == 200
     assert remove_response.status == 200
     assert calls == [("add", ["manual"]), ("remove", ["manual"])]
+
+
+@pytest.mark.asyncio
+async def test_remove_tags_response_exposes_protected_bucket(monkeypatch):
+    """The DELETE /tags route must serialise the ``protected`` bucket rather than
+    drop it: a present-but-automatic tag the service reports as protected has to
+    reach the HTTP body so the contract matches RemoveTagsResult (review2-18)."""
+    monkeypatch.setattr(routes, "USER_MANAGER", _UserManager())
+    monkeypatch.setattr(
+        routes,
+        "remove_tags",
+        lambda **_kwargs: SimpleNamespace(
+            removed=[], not_present=[], total_tags=["auto"], protected=["auto"]
+        ),
+    )
+
+    response = await routes.delete_asset_tags.__wrapped__(_JsonRequest(["auto"]))
+
+    assert response.status == 200
+    body = json.loads(response.body)
+    assert body["protected"] == ["auto"]