review-stack 6/6: tests (8 files, +753/-238)

Review-and-land stack for synap5e/feat/assets-di, generated by review-stack.py. Once approved,
merges DOWN into the layer below (a fast-forward); only the bottom layer
squash-merges into the real base. See ~/adocs/review-stack.md.
Rule: every changed file under tests-unit/ or tests/ (added, modified, or deleted)
Question: Is the code layer well covered, and did any edit weaken an existing check?
Source tip: eca2c74bff
Merge-base: 20d59d2a5f

tests-unit/assets_test/services/test_cached_execution_path.py

@@ -1,27 +1,19 @@
 import os
-import sys
-import types
-from contextlib import contextmanager
 from pathlib import Path
-from unittest.mock import patch
 
 from sqlalchemy import event, select
 
 import folder_paths
 from app.assets.database.models import Asset, AssetContent
+from app.assets.manager import AssetsEnabled
 from comfy_execution.asset_enrichment import (
     register_cached_outputs,
     register_executed_outputs,
 )
 
-
-@contextmanager
-def _assets_enabled(enabled: bool = True):
-    with patch.dict(
-        sys.modules,
-        {"comfy.cli_args": types.SimpleNamespace(args=types.SimpleNamespace(enable_assets=enabled))},
-    ):
-        yield
+class _ArgsStub:
+    enable_assets = True
+    enable_asset_hashing = False
 
 
 def _write_output_file(name: str) -> Path:
@@ -42,15 +34,14 @@ def _wrapper(name: str, node_id: str = "1") -> dict:
 def test_cached_replay_binds_new_record_to_existing_content(mock_create_session):
     path = _write_output_file("cached-execution-bind.png")
     try:
-        with _assets_enabled():
-            executed = register_executed_outputs(_output_ui(path.name), "original-job")
+        manager = AssetsEnabled(_ArgsStub())
+        executed = register_executed_outputs(_output_ui(path.name), "original-job", manager)
         original_id = executed["images"][0]["id"]
         with mock_create_session() as session:
             original_content_id = session.get(Asset, original_id).content_id
 
         wrapper = _wrapper(path.name)
-        with _assets_enabled():
-            enriched = register_cached_outputs(wrapper, "cached-job")
+        enriched = register_cached_outputs(wrapper, "cached-job", manager)
         cached_id = enriched["output"]["images"][0]["id"]
 
         with mock_create_session() as session:
@@ -83,8 +74,8 @@ def test_cached_replay_does_not_update_existing_content(mock_create_session, db_
             update_statements.append(statement)
 
     try:
-        with _assets_enabled():
-            executed = register_executed_outputs(_output_ui(path.name), "original-job")
+        manager = AssetsEnabled(_ArgsStub())
+        executed = register_executed_outputs(_output_ui(path.name), "original-job", manager)
         original_id = executed["images"][0]["id"]
         with mock_create_session() as session:
             original_content = session.get(
@@ -102,8 +93,7 @@ def test_cached_replay_does_not_update_existing_content(mock_create_session, db_
 
         event.listen(db_engine, "before_cursor_execute", capture_updates)
         wrapper = _wrapper(path.name)
-        with _assets_enabled():
-            register_cached_outputs(wrapper, "cached-job")
+        register_cached_outputs(wrapper, "cached-job", manager)
 
         with mock_create_session() as session:
             content = session.get(AssetContent, original_state[0])

tests-unit/assets_test/services/test_manager_isolated.py

@@ -0,0 +1,183 @@
+import dataclasses
+import threading
+from collections.abc import Callable
+from contextlib import AbstractContextManager
+from pathlib import Path
+from unittest.mock import MagicMock, Mock, call
+
+import folder_paths
+import pytest
+from sqlalchemy.orm import Session
+
+from app.assets import lifecycle
+from app.assets import manager as manager_module
+from app.assets.database.models import Asset, AssetContent
+from app.assets.database.queries.records import create_content, create_record
+from app.assets.manager import AssetsEnabled
+from app.assets.seeder import asset_seeder
+from app.assets.services.schemas import RegisteredAsset, UploadAssetView
+
+
+class _ArgsStub:
+    enable_assets = True
+    enable_asset_hashing = False
+
+
+@pytest.fixture
+def enabled_manager() -> AssetsEnabled:
+    return AssetsEnabled(_ArgsStub())
+
+
+@pytest.fixture
+def asset_roots(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> tuple[Path, Path, Path]:
+    output_dir = tmp_path / "output"
+    input_dir = tmp_path / "input"
+    temp_dir = tmp_path / "temp"
+    output_dir.mkdir()
+    input_dir.mkdir()
+    temp_dir.mkdir()
+    monkeypatch.setattr(folder_paths, "get_output_directory", lambda: str(output_dir))
+    monkeypatch.setattr(folder_paths, "get_input_directory", lambda: str(input_dir))
+    monkeypatch.setattr(folder_paths, "get_temp_directory", lambda: str(temp_dir))
+    return output_dir, input_dir, temp_dir
+
+
+def test_executed_and_cached_outputs_share_unhashed_content(
+    enabled_manager: AssetsEnabled,
+    asset_roots: tuple[Path, Path, Path],
+    mock_create_session: Callable[[], AbstractContextManager[Session]],
+) -> None:
+    output_dir, _, _ = asset_roots
+    output_path = output_dir / "executed.png"
+    output_path.write_bytes(b"executed output")
+
+    executed = enabled_manager.register_executed_output(
+        str(output_path), job_id="executed-job"
+    )
+
+    assert isinstance(executed, RegisteredAsset)
+    assert executed.id
+    assert executed.content_id
+    assert executed.name
+    assert executed.job_id == "executed-job"
+    assert not hasattr(executed, "asset_hash")
+    assert {field.name for field in dataclasses.fields(executed)} == {
+        "id",
+        "content_id",
+        "job_id",
+        "name",
+    }
+    with mock_create_session() as session:
+        asset = session.get(Asset, executed.id)
+        content = session.get(AssetContent, executed.content_id)
+        assert asset is not None
+        assert asset.content_id == executed.content_id
+        assert content is not None
+        assert content.hash is None
+
+    cached = enabled_manager.register_cached_output(str(output_path), job_id="cached-job")
+
+    assert isinstance(cached, RegisteredAsset)
+    assert cached.id != executed.id
+    assert cached.content_id == executed.content_id
+    assert cached.job_id == "cached-job"
+    assert (
+        enabled_manager.register_cached_output(
+            str(output_dir / "unknown.png"), job_id="unknown-job"
+        )
+        is None
+    )
+
+
+def test_register_upload_hashes_and_tags_fresh_input_file(
+    enabled_manager: AssetsEnabled,
+    asset_roots: tuple[Path, Path, Path],
+    mock_create_session: Callable[[], AbstractContextManager[Session]],
+) -> None:
+    _, input_dir, _ = asset_roots
+    upload_path = input_dir / "pasted" / "upload.png"
+    upload_path.parent.mkdir()
+    upload_path.write_bytes(b"uploaded input")
+
+    view = enabled_manager.register_upload(
+        str(upload_path),
+        name=upload_path.name,
+        upload_type="input",
+        subfolder="pasted",
+        content_written=True,
+    )
+
+    assert isinstance(view, UploadAssetView)
+    assert isinstance(view.asset, RegisteredAsset)
+    assert view.asset_hash
+    assert "pasted" in view.tags
+
+
+def test_startup_runs_against_memory_db_without_starting_a_scanner_thread(
+    enabled_manager: AssetsEnabled,
+    asset_roots: tuple[Path, Path, Path],
+    mock_create_session: Callable[[], AbstractContextManager[Session]],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _, _, temp_dir = asset_roots
+    (temp_dir / "stale.tmp").write_bytes(b"stale")
+    seeder_start = MagicMock(return_value=False)
+    monkeypatch.setattr(lifecycle, "create_session", mock_create_session)
+    monkeypatch.setattr(lifecycle, "start_asset_seeder", seeder_start)
+
+    thread_count = threading.active_count()
+    enabled_manager.startup()
+    assert threading.active_count() == thread_count, (
+        "start_asset_seeder is mocked, so a new thread means a component other than the seeder spawned one"
+    )
+
+    assert not temp_dir.exists()
+    seeder_start.assert_called_once_with()
+
+
+def test_ensure_scan_started_starts_the_lazy_object_info_scan(
+    enabled_manager: AssetsEnabled, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    seeder_start = MagicMock()
+    monkeypatch.setattr(asset_seeder, "start", seeder_start)
+
+    enabled_manager.ensure_scan_started()
+
+    seeder_start.assert_called_once_with(roots=("models", "input", "output"))
+
+
+def test_shutdown_runs_lifecycle_cleanup_when_seeder_shutdown_times_out(
+    enabled_manager: AssetsEnabled,
+    asset_roots: tuple[Path, Path, Path],
+    mock_create_session: Callable[[], AbstractContextManager[Session]],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _, _, temp_dir = asset_roots
+    stale_file = temp_dir / "stale.tmp"
+    stale_file.write_bytes(b"stale")
+    with mock_create_session() as session:
+        content = create_content(session, path=str(stale_file))
+        asset = create_record(session, content_id=content.id, name=stale_file.name)
+        session.commit()
+        asset_id = asset.id
+        content_id = content.id
+
+    calls = Mock()
+    seeder_shutdown = MagicMock(return_value=False)
+    run_shutdown = MagicMock(wraps=manager_module.run_shutdown)
+    calls.attach_mock(seeder_shutdown, "seeder_shutdown")
+    calls.attach_mock(run_shutdown, "run_shutdown")
+    monkeypatch.setattr(lifecycle, "can_create_session", lambda: True)
+    monkeypatch.setattr(lifecycle, "create_session", mock_create_session)
+    monkeypatch.setattr(asset_seeder, "shutdown", seeder_shutdown)
+    monkeypatch.setattr(manager_module, "run_shutdown", run_shutdown)
+
+    enabled_manager.shutdown()
+
+    assert calls.mock_calls == [call.seeder_shutdown(), call.run_shutdown()]
+    assert not temp_dir.exists()
+    with mock_create_session() as session:
+        assert session.get(Asset, asset_id) is None
+        assert session.get(AssetContent, content_id) is None

tests-unit/assets_test/services/test_noassets_parity.py

@@ -0,0 +1,177 @@
+from collections.abc import Callable
+from contextlib import AbstractContextManager
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import pytest
+from aiohttp import web
+from aiohttp.pytest_plugin import AiohttpClient
+from sqlalchemy.orm import Session
+
+import folder_paths
+from app.assets import lifecycle
+from app.assets.api import routes
+from app.assets.database.models import Asset, AssetContent
+from app.assets.database.queries.records import create_content, create_record
+from app.assets.manager import NoAssets
+from app.assets.mode import hashing_enabled
+from app.assets.seeder import asset_seeder
+from app.assets.services.hash_mode_state import read_stored_mode
+
+
+class _Args:
+    enable_assets = False
+
+    def __init__(self, hashing: bool) -> None:
+        self.enable_asset_hashing = hashing
+
+
+def _no_assets(*, hashing: bool = False) -> NoAssets:
+    return NoAssets(_Args(hashing))
+
+
+@pytest.mark.asyncio
+async def test_noassets_register_routes_returns_service_disabled_and_disables_seeder(
+    aiohttp_client: AiohttpClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(routes, "_ASSETS_ENABLED", False)
+    monkeypatch.setattr(asset_seeder, "_disabled", False)
+    app = web.Application()
+
+    _no_assets().register_routes(app, None)
+    client = await aiohttp_client(app)
+    response = await client.get("/api/assets")
+
+    assert response.status == 503
+    assert (await response.json())["error"]["code"] == "SERVICE_DISABLED"
+    assert asset_seeder.is_disabled()
+
+
+def test_noassets_startup_applies_hash_mode_persists_state_and_cleans_temp_dir(
+    mock_create_session: Callable[[], AbstractContextManager[Session]],
+    session: Session,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    temp_file = tmp_path / "startup-stale.bin"
+    temp_file.write_bytes(b"stale")
+    monkeypatch.setattr(lifecycle, "create_session", mock_create_session)
+    monkeypatch.setattr(folder_paths, "get_temp_directory", lambda: str(tmp_path))
+
+    _no_assets(hashing=True).startup()
+
+    assert hashing_enabled() is True
+    assert read_stored_mode(session) == "on"
+    assert not temp_file.exists()
+
+
+def test_noassets_shutdown_wipes_temp_rows_and_files(
+    mock_create_session: Callable[[], AbstractContextManager[Session]],
+    session: Session,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    manager = _no_assets()
+    monkeypatch.setattr(lifecycle, "create_session", mock_create_session)
+    monkeypatch.setattr(lifecycle, "can_create_session", lambda: True)
+    monkeypatch.setattr(folder_paths, "get_temp_directory", lambda: str(tmp_path))
+    manager.startup()
+
+    tmp_path.mkdir()
+    temp_file = tmp_path / "shutdown-stale.bin"
+    temp_file.write_bytes(b"stale")
+    content = create_content(session, str(temp_file))
+    record = create_record(session, content.id, temp_file.name)
+    session.commit()
+    record_id = record.id
+    content_id = content.id
+
+    manager.shutdown()
+    session.expire_all()
+
+    assert session.get(Asset, record_id) is None
+    assert session.get(AssetContent, content_id) is None
+    assert not temp_file.exists()
+
+
+def test_noassets_shutdown_runs_cleanup_after_unsuccessful_seeder_shutdown(
+    mock_create_session: Callable[[], AbstractContextManager[Session]],
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    temp_file = tmp_path / "seeder-timeout.bin"
+    temp_file.write_bytes(b"stale")
+    calls: list[str] = []
+    monkeypatch.setattr(lifecycle, "create_session", mock_create_session)
+    monkeypatch.setattr(lifecycle, "can_create_session", lambda: True)
+    monkeypatch.setattr(folder_paths, "get_temp_directory", lambda: str(tmp_path))
+
+    def seeder_shutdown() -> bool:
+        calls.append("seeder")
+        return False
+
+    def cleanup() -> None:
+        calls.append("cleanup")
+        lifecycle.run_shutdown()
+
+    with (
+        patch.object(asset_seeder, "shutdown", side_effect=seeder_shutdown),
+        patch("app.assets.manager.run_shutdown", side_effect=cleanup) as cleanup_spy,
+    ):
+        _no_assets().shutdown()
+
+    cleanup_spy.assert_called_once_with()
+    assert calls == ["seeder", "cleanup"]
+    assert not temp_file.exists()
+
+
+def test_noassets_shutdown_without_database_sweeps_files_without_session_access(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    temp_file = tmp_path / "no-database.bin"
+    temp_file.write_bytes(b"stale")
+    create_session = Mock()
+    monkeypatch.setattr(lifecycle, "can_create_session", lambda: False)
+    monkeypatch.setattr(lifecycle, "create_session", create_session)
+    monkeypatch.setattr(folder_paths, "get_temp_directory", lambda: str(tmp_path))
+
+    _no_assets().shutdown()
+
+    assert not temp_file.exists()
+    create_session.assert_not_called()
+
+
+def test_noassets_callbacks_are_noops_without_seeder_side_effects() -> None:
+    manager = _no_assets()
+    with (
+        patch.object(asset_seeder, "start") as start,
+        patch.object(asset_seeder, "pause") as pause,
+        patch.object(asset_seeder, "enqueue_enrich") as enqueue_enrich,
+        patch.object(asset_seeder, "resume") as resume,
+        patch.object(asset_seeder, "set_event_sink") as set_event_sink,
+    ):
+        assert manager.ensure_scan_started() is None
+        assert manager.pause_background_scan() is None
+        assert manager.queue_output_enrichment() is None
+        assert manager.resume_background_scan() is None
+        assert manager.set_event_sink(Mock()) is None
+
+    start.assert_not_called()
+    pause.assert_not_called()
+    enqueue_enrich.assert_not_called()
+    resume.assert_not_called()
+    set_event_sink.assert_not_called()
+
+
+def test_noassets_registration_methods_return_none() -> None:
+    manager = _no_assets()
+
+    assert manager.register_upload(
+        "/tmp/output.png", "output.png", "output", "", content_written=True
+    ) is None
+    assert manager.register_executed_output("/tmp/output.png", "job-id") is None
+    assert manager.register_cached_output("/tmp/output.png", "job-id") is None
+
+
+def test_noassets_is_disabled() -> None:
+    assert _no_assets().enabled is False

tests-unit/assets_test/services/test_output_registration.py

@@ -1,7 +1,4 @@
 import os
-import sys
-import types
-from contextlib import contextmanager
 from pathlib import Path
 from unittest.mock import patch
 
@@ -9,19 +6,16 @@ import folder_paths
 from sqlalchemy import select
 
 from app.assets.database.models import Asset, AssetContent
+from app.assets.manager import AssetsEnabled, NoAssets
 from comfy_execution.asset_enrichment import (
     register_cached_outputs,
     register_executed_outputs,
 )
 
-
-@contextmanager
-def _assets_enabled(enabled: bool = True):
-    with patch.dict(
-        sys.modules,
-        {"comfy.cli_args": types.SimpleNamespace(args=types.SimpleNamespace(enable_assets=enabled))},
-    ):
-        yield
+class _ArgsStub:
+    def __init__(self, enable_assets: bool = True) -> None:
+        self.enable_assets = enable_assets
+        self.enable_asset_hashing = False
 
 
 def _write_output_file(name: str, data: bytes) -> Path:
@@ -44,8 +38,9 @@ def test_executed_adapter_registers_new_output(mock_create_session):
     try:
         output_ui = _output_ui(path.name)
 
-        with _assets_enabled():
-            enriched = register_executed_outputs(output_ui, "exec-job")
+        enriched = register_executed_outputs(
+            output_ui, "exec-job", AssetsEnabled(_ArgsStub())
+        )
 
         new_id = enriched["images"][0]["id"]
         assert "id" not in output_ui["images"][0]
@@ -62,13 +57,12 @@ def test_executed_adapter_registers_new_output(mock_create_session):
 def test_executed_adapter_over_existing_path_marks_old_missing(mock_create_session):
     path = _write_output_file("adapter-executed-replace.png", b"original")
     try:
-        with _assets_enabled():
-            first = register_executed_outputs(_output_ui(path.name), "job-1")
+        manager = AssetsEnabled(_ArgsStub())
+        first = register_executed_outputs(_output_ui(path.name), "job-1", manager)
         old_id = first["images"][0]["id"]
 
         path.write_bytes(b"replacement")
-        with _assets_enabled():
-            second = register_executed_outputs(_output_ui(path.name), "job-2")
+        second = register_executed_outputs(_output_ui(path.name), "job-2", manager)
         new_id = second["images"][0]["id"]
 
         assert new_id != old_id
@@ -92,8 +86,7 @@ def test_executed_adapter_disabled_registers_nothing(mock_create_session):
     try:
         output_ui = _output_ui(path.name)
 
-        with _assets_enabled(False):
-            enriched = register_executed_outputs(output_ui, "job")
+        enriched = register_executed_outputs(output_ui, "job", NoAssets(_ArgsStub(False)))
 
         assert "id" not in enriched["images"][0]
         with mock_create_session() as session:
@@ -112,11 +105,13 @@ def test_executed_adapter_registration_failure_never_raises(mock_create_session)
     try:
         output_ui = _output_ui(path.name)
 
-        with _assets_enabled(), patch(
-            "app.assets.services.ingest.register_executed_output",
+        with patch(
+            "app.assets.manager.ingest_register_executed_output",
             side_effect=RuntimeError("boom"),
         ):
-            enriched = register_executed_outputs(output_ui, "job")
+            enriched = register_executed_outputs(
+                output_ui, "job", AssetsEnabled(_ArgsStub())
+            )
 
         assert "id" not in enriched["images"][0]
     finally:
@@ -128,8 +123,9 @@ def test_cached_adapter_without_live_content_is_nonevent(mock_create_session):
     try:
         wrapper = _wrapper(path.name)
 
-        with _assets_enabled():
-            enriched = register_cached_outputs(wrapper, "cached-job")
+        enriched = register_cached_outputs(
+            wrapper, "cached-job", AssetsEnabled(_ArgsStub())
+        )
 
         assert "id" not in enriched["output"]["images"][0]
         with mock_create_session() as session:

tests-unit/execution_test/test_enrich_output.py

@@ -1,95 +1,49 @@
 import copy
 import os
-import sys
 import tempfile
-import types
 from collections import namedtuple
-from contextlib import contextmanager
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
+
+import folder_paths
+import pytest
+
+from app.assets.manager import NoAssets
+from comfy_execution.asset_enrichment import (
+    emit_cached_output,
+    register_cached_outputs,
+    register_executed_outputs,
+)
+from test_inmemory_assets import AssetCall, InMemoryAssets
 
 _CacheEntry = namedtuple("_CacheEntry", ["ui", "outputs"])
 
 _BASE = os.path.join(tempfile.gettempdir(), "asset-enrichment-test-base")
 
 
-class _FakeAssetDB:
-
-    def __init__(self) -> None:
-        self.live_id_by_path: dict[str, str] = {}
-        self.missing: list[str] = []
-        self.deliveries: list[tuple[str, str, str | None]] = []
-        self._counter = 0
-
-    def _new_id(self) -> str:
-        self._counter += 1
-        return f"asset-{self._counter}"
-
-    def register_executed(self, abs_path: str, job_id: str | None = None):
-        old = self.live_id_by_path.get(abs_path)
-        if old is not None:
-            self.missing.append(old)
-        new_id = self._new_id()
-        self.live_id_by_path[abs_path] = new_id
-        self.deliveries.append((new_id, abs_path, job_id))
-        return types.SimpleNamespace(
-            id=new_id,
-            content_id=f"content-{new_id}",
-            job_id=job_id,
-            name=os.path.basename(abs_path),
-        )
-
-    def register_cached(self, abs_path: str, job_id: str | None = None):
-        if abs_path not in self.live_id_by_path:
-            return None
-        new_id = self._new_id()
-        self.deliveries.append((new_id, abs_path, job_id))
-        return types.SimpleNamespace(
-            id=new_id,
-            content_id=f"content-{self.live_id_by_path[abs_path]}",
-            job_id=job_id,
-            name=os.path.basename(abs_path),
-        )
+class _ArgsStub:
+    enable_assets = False
+    enable_asset_hashing = False
 
 
 class _Server:
+    last_node_id: str | None = None
+    sockets_metadata: dict[str, dict[str, object]] = {}
+
     def __init__(self, client_id: str | None = None) -> None:
         self.client_id = client_id
         self.sent: list[tuple] = []
 
-    def send_sync(self, event, payload, client_id):
-        self.sent.append((event, payload, client_id))
+    def send_sync(self, event, data, sid=None):
+        self.sent.append((event, data, sid))
+
+    def queue_updated(self) -> None:
+        pass
 
 
-@contextmanager
-def _patched(
-    fake: _FakeAssetDB,
-    *,
-    enable_assets: bool = True,
-    directory: str | None = _BASE,
-    file_exists: bool = True,
-    executed_side_effect=None,
-    cached_side_effect=None,
-):
-    reg_exec = MagicMock(side_effect=executed_side_effect or fake.register_executed)
-    reg_cached = MagicMock(side_effect=cached_side_effect or fake.register_cached)
-    modules = {
-        "comfy.cli_args": MagicMock(
-            args=types.SimpleNamespace(enable_assets=enable_assets)
-        ),
-        "folder_paths": MagicMock(
-            get_directory_by_type=MagicMock(return_value=directory)
-        ),
-        "app.assets.services.ingest": MagicMock(
-            register_executed_output=reg_exec,
-            register_cached_output=reg_cached,
-        ),
-    }
-    with patch.dict(sys.modules, modules), patch(
-        "os.path.isfile", return_value=file_exists
-    ):
-        import comfy_execution.asset_enrichment as module
-
-        yield module, reg_exec, reg_cached
+@pytest.fixture
+def output_path_environment(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(folder_paths, "get_directory_by_type", lambda _type: _BASE)
+    monkeypatch.setattr(os.path, "isfile", lambda _path: True)
 
 
 def _output(filename: str, *, subfolder: str = "", type_: str = "output") -> dict:
@@ -122,75 +76,80 @@ def _find_ids(value) -> list:
 
 
 # REQUIRED test names (invoked verbatim downstream). Do not rename.
-def test_executed_new_path_gets_fresh_id() -> None:
-    fake = _FakeAssetDB()
+def test_executed_new_path_gets_fresh_id(output_path_environment) -> None:
+    manager = InMemoryAssets()
     output_ui = _output("new.png")
 
-    with _patched(fake) as (module, reg_exec, _):
-        enriched = module.register_executed_outputs(output_ui, "job-1")
+    enriched = register_executed_outputs(output_ui, "job-1", manager)
 
     assert enriched["images"][0]["id"] == "asset-1"
     assert "id" not in output_ui["images"][0]
-    reg_exec.assert_called_once()
-    _, kwargs = reg_exec.call_args
-    assert kwargs.get("job_id") == "job-1"
-    assert fake.deliveries == [("asset-1", os.path.join(_BASE, "new.png"), "job-1")]
+    assert manager.calls == [
+        AssetCall(
+            "register_executed_output", (os.path.join(_BASE, "new.png"), "job-1")
+        )
+    ]
+    deliveries = manager.deliveries_by_path[os.path.join(_BASE, "new.png")]
+    assert [(delivery.asset.id, delivery.asset.job_id) for delivery in deliveries] == [
+        ("asset-1", "job-1")
+    ]
 
 
-def test_executed_over_existing_path_gets_new_id_and_marks_old_missing() -> None:
-    fake = _FakeAssetDB()
+def test_executed_over_existing_path_gets_new_id_and_marks_old_missing(output_path_environment) -> None:
+    manager = InMemoryAssets()
 
-    first_output = _output("same.png")
-    with _patched(fake) as (module, _, _c):
-        first = module.register_executed_outputs(first_output, "job-1")
+    first = register_executed_outputs(_output("same.png"), "job-1", manager)
     old_id = first["images"][0]["id"]
 
-    second_output = _output("same.png")
-    with _patched(fake) as (module, _, _c):
-        second = module.register_executed_outputs(second_output, "job-2")
+    second = register_executed_outputs(_output("same.png"), "job-2", manager)
     new_id = second["images"][0]["id"]
 
+    deliveries = manager.deliveries_by_path[os.path.join(_BASE, "same.png")]
     assert new_id != old_id
-    assert old_id in fake.missing
-    assert fake.live_id_by_path[os.path.join(_BASE, "same.png")] == new_id
+    assert deliveries[0].superseded is True
+    assert deliveries[-1].asset.id == new_id
 
 
-def test_cached_replay_creates_delivery_with_current_job_id() -> None:
-    fake = _FakeAssetDB()
+def test_cached_replay_creates_delivery_with_current_job_id(output_path_environment) -> None:
+    manager = InMemoryAssets()
 
-    with _patched(fake) as (module, _, _c):
-        module.register_executed_outputs(_output("replay.png"), "seed-job")
-        enriched = module.register_cached_outputs(_wrapper("replay.png"), "replay-job")
+    register_executed_outputs(_output("replay.png"), "seed-job", manager)
+    enriched = register_cached_outputs(_wrapper("replay.png"), "replay-job", manager)
 
     replay_id = enriched["output"]["images"][0]["id"]
-    assert (replay_id, os.path.join(_BASE, "replay.png"), "replay-job") in fake.deliveries
+    deliveries = manager.deliveries_by_path[os.path.join(_BASE, "replay.png")]
+    assert (replay_id, "replay-job") == (deliveries[-1].asset.id, deliveries[-1].asset.job_id)
     assert replay_id != "asset-1"
 
 
-def test_cached_registration_happens_without_client() -> None:
-    fake = _FakeAssetDB()
+def test_cached_registration_happens_without_client(output_path_environment) -> None:
+    manager = InMemoryAssets()
     server = _Server(client_id=None)
     ui_outputs: dict = {}
 
-    with _patched(fake) as (module, _, _c):
-        module.register_executed_outputs(_output("noclient.png"), "seed-job")
-        module.emit_cached_output(
-            server, "node-1", "node-1", _CacheEntry(ui=_wrapper("noclient.png"), outputs=[]),
-            "job-x", ui_outputs,
-        )
+    register_executed_outputs(_output("noclient.png"), "seed-job", manager)
+    emit_cached_output(
+        server,
+        "node-1",
+        "node-1",
+        _CacheEntry(ui=_wrapper("noclient.png"), outputs=[]),
+        "job-x",
+        ui_outputs,
+        manager,
+    )
 
-    assert any(job == "job-x" for (_id, _path, job) in fake.deliveries)
+    deliveries = manager.deliveries_by_path[os.path.join(_BASE, "noclient.png")]
+    assert any(delivery.asset.job_id == "job-x" for delivery in deliveries)
     assert "node-1" in ui_outputs
     assert ui_outputs["node-1"]["output"]["images"][0]["id"] is not None
     assert server.sent == []
 
 
-def test_cache_entry_contains_no_asset_ids() -> None:
-    fake = _FakeAssetDB()
+def test_cache_entry_contains_no_asset_ids(output_path_environment) -> None:
+    manager = InMemoryAssets()
     output_ui = _output("keep.png")
 
-    with _patched(fake) as (module, _, _c):
-        enriched = module.register_executed_outputs(output_ui, "job")
+    enriched = register_executed_outputs(output_ui, "job", manager)
 
     cache_entry = _CacheEntry(
         ui={"meta": {"node_id": "1"}, "output": output_ui}, outputs=[]
@@ -199,134 +158,137 @@ def test_cache_entry_contains_no_asset_ids() -> None:
     assert _find_ids(enriched) == ["asset-1"]
 
 
-def test_cached_ui_object_unmodified_after_emission() -> None:
-    fake = _FakeAssetDB()
+def test_cached_ui_object_unmodified_after_emission(output_path_environment) -> None:
+    manager = InMemoryAssets()
     server = _Server(client_id="client-1")
     cached = _CacheEntry(ui=_wrapper("immut.png"), outputs=[])
     snapshot = copy.deepcopy(cached.ui)
 
-    with _patched(fake) as (module, _, _c):
-        module.register_executed_outputs(_output("immut.png"), "seed-job")
-        module.emit_cached_output(
-            server, "1", "1", cached, "prompt-1", {}
-        )
+    register_executed_outputs(_output("immut.png"), "seed-job", manager)
+    emit_cached_output(server, "1", "1", cached, "prompt-1", {}, manager)
 
     assert cached.ui == snapshot
 
 
-def test_double_emission_yields_single_delivery() -> None:
-    fake = _FakeAssetDB()
+def test_double_emission_yields_single_delivery(output_path_environment) -> None:
+    manager = InMemoryAssets()
     server = _Server(client_id="client-1")
     ui_outputs: dict = {}
     cached = _CacheEntry(ui=_wrapper("dbl.png"), outputs=[])
 
-    with _patched(fake) as (module, _, _c):
-        module.register_executed_outputs(_output("dbl.png"), "seed-job")
-        module.emit_cached_output(server, "1", "1", cached, "prompt-1", ui_outputs)
-        module.emit_cached_output(server, "1", "1", cached, "prompt-1", ui_outputs)
+    register_executed_outputs(_output("dbl.png"), "seed-job", manager)
+    emit_cached_output(server, "1", "1", cached, "prompt-1", ui_outputs, manager)
+    emit_cached_output(server, "1", "1", cached, "prompt-1", ui_outputs, manager)
 
-    cached_deliveries = [d for d in fake.deliveries if d[2] == "prompt-1"]
-    assert len(cached_deliveries) == 1
+    deliveries = manager.deliveries_by_path[os.path.join(_BASE, "dbl.png")]
+    assert len([delivery for delivery in deliveries if delivery.asset.job_id == "prompt-1"]) == 1
 
 
-def test_executed_disabled_returns_unenriched_copy() -> None:
-    fake = _FakeAssetDB()
+def test_executed_disabled_returns_unenriched_copy(output_path_environment) -> None:
+    manager = NoAssets(_ArgsStub())
     output_ui = _output("a.png")
 
-    with _patched(fake, enable_assets=False) as (module, reg_exec, _):
-        enriched = module.register_executed_outputs(output_ui, "job")
+    with patch.object(
+        manager,
+        "register_executed_output",
+        wraps=manager.register_executed_output,
+    ) as register_executed_output:
+        enriched = register_executed_outputs(output_ui, "job", manager)
 
     assert enriched is not output_ui
     assert "id" not in enriched["images"][0]
-    reg_exec.assert_not_called()
+    register_executed_output.assert_not_called()
 
 
-def test_executed_missing_file_is_skipped() -> None:
-    fake = _FakeAssetDB()
+def test_executed_missing_file_is_skipped(monkeypatch: pytest.MonkeyPatch) -> None:
+    manager = InMemoryAssets()
+    monkeypatch.setattr(folder_paths, "get_directory_by_type", lambda _type: _BASE)
+    monkeypatch.setattr(os.path, "isfile", lambda _path: False)
 
-    with _patched(fake, file_exists=False) as (module, reg_exec, _):
-        enriched = module.register_executed_outputs(_output("gone.png"), "job")
+    enriched = register_executed_outputs(_output("gone.png"), "job", manager)
 
     assert "id" not in enriched["images"][0]
-    reg_exec.assert_not_called()
+    assert manager.calls == []
 
 
-def test_executed_path_escape_is_skipped() -> None:
-    fake = _FakeAssetDB()
+def test_executed_path_escape_is_skipped(output_path_environment) -> None:
+    manager = InMemoryAssets()
     output_ui = {"images": [{"filename": "passwd", "subfolder": "../../etc", "type": "output"}]}
 
-    with _patched(fake) as (module, reg_exec, _):
-        enriched = module.register_executed_outputs(output_ui, "job")
+    enriched = register_executed_outputs(output_ui, "job", manager)
 
     assert "id" not in enriched["images"][0]
-    reg_exec.assert_not_called()
+    assert manager.calls == []
 
 
-def test_executed_non_list_value_passes_through() -> None:
-    fake = _FakeAssetDB()
+def test_executed_non_list_value_passes_through(output_path_environment) -> None:
+    manager = InMemoryAssets()
 
-    with _patched(fake) as (module, _, _c):
-        enriched = module.register_executed_outputs({"text": "hello"}, "job")
+    enriched = register_executed_outputs({"text": "hello"}, "job", manager)
 
     assert enriched["text"] == "hello"
 
 
-def test_executed_registration_failure_never_raises() -> None:
-    fake = _FakeAssetDB()
+def test_executed_registration_failure_never_raises(
+    output_path_environment, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    manager = InMemoryAssets()
 
-    def boom(_abs_path, job_id=None):
+    def boom(_abs_path: str, job_id: str | None) -> None:
         raise RuntimeError("registration blew up")
 
-    with _patched(fake, executed_side_effect=boom) as (module, _, _c):
-        enriched = module.register_executed_outputs(_output("boom.png"), "job")
+    monkeypatch.setattr(manager, "register_executed_output", boom)
+
+    enriched = register_executed_outputs(_output("boom.png"), "job", manager)
 
     assert "id" not in enriched["images"][0]
 
 
-def test_cached_strips_legacy_ids_before_replay() -> None:
-    fake = _FakeAssetDB()
+def test_cached_strips_legacy_ids_before_replay(output_path_environment) -> None:
+    manager = InMemoryAssets()
     wrapper = _wrapper("legacy.png")
     wrapper["output"]["images"][0]["id"] = "stale-id"
 
-    with _patched(fake) as (module, _, _c):
-        module.register_executed_outputs(_output("legacy.png"), "seed-job")
-        enriched = module.register_cached_outputs(wrapper, "replay-job")
+    register_executed_outputs(_output("legacy.png"), "seed-job", manager)
+    enriched = register_cached_outputs(wrapper, "replay-job", manager)
 
     assert enriched["output"]["images"][0]["id"] != "stale-id"
     assert wrapper["output"]["images"][0]["id"] == "stale-id"
 
 
-def test_cached_none_wrapper_returns_none() -> None:
-    fake = _FakeAssetDB()
+def test_cached_none_wrapper_returns_none(output_path_environment) -> None:
+    manager = InMemoryAssets()
 
-    with _patched(fake) as (module, _, reg_cached):
-        result = module.register_cached_outputs(None, "job")
+    result = register_cached_outputs(None, "job", manager)
 
     assert result is None
-    reg_cached.assert_not_called()
+    assert manager.calls == []
 
 
-def test_cached_missing_live_content_is_nonevent() -> None:
-    fake = _FakeAssetDB()
+def test_cached_missing_live_content_is_nonevent(output_path_environment) -> None:
+    manager = InMemoryAssets()
 
-    with _patched(fake) as (module, _, _c):
-        enriched = module.register_cached_outputs(_wrapper("orphan.png"), "job")
+    enriched = register_cached_outputs(_wrapper("orphan.png"), "job", manager)
 
     assert "id" not in enriched["output"]["images"][0]
-    assert fake.deliveries == []
+    assert manager.deliveries_by_path == {}
 
 
-def test_emit_cached_sends_enriched_output_to_client() -> None:
-    fake = _FakeAssetDB()
+def test_emit_cached_sends_enriched_output_to_client(output_path_environment) -> None:
+    manager = InMemoryAssets()
     server = _Server(client_id="client-1")
     ui_outputs: dict = {}
 
-    with _patched(fake) as (module, _, _c):
-        module.register_executed_outputs(_output("send.png"), "seed-job")
-        module.emit_cached_output(
-            server, "1", "1", _CacheEntry(ui=_wrapper("send.png"), outputs=[]),
-            "prompt-1", ui_outputs,
-        )
+    register_executed_outputs(_output("send.png"), "seed-job", manager)
+    emit_cached_output(
+        server,
+        "1",
+        "1",
+        _CacheEntry(ui=_wrapper("send.png"), outputs=[]),
+        "prompt-1",
+        ui_outputs,
+        manager,
+    )
 
     assert len(server.sent) == 1
     event, payload, client_id = server.sent[0]

tests-unit/execution_test/test_execute_reentry.py

@@ -1,11 +1,11 @@
 import asyncio
 import os
-import sys
 import tempfile
-import types
 
 import pytest
 
+from test_inmemory_assets import InMemoryAssets
+
 _BASE = os.path.join(tempfile.gettempdir(), "execute-reentry-test-base")
 
 
@@ -100,28 +100,15 @@ def execution_env(monkeypatch):
     except Exception as exc:  # pragma: no cover - environment dependent
         pytest.skip(f"execution module could not be imported in CPU mode: {exc!r}")
 
-    monkeypatch.setattr(args, "enable_assets", True, raising=False)
-
-    counter = {"n": 0}
-
-    def _register(abs_path, job_id=None):
-        counter["n"] += 1
-        return types.SimpleNamespace(id=f"asset-{counter['n']}", job_id=job_id)
-
-    monkeypatch.setitem(
-        sys.modules,
-        "app.assets.services.ingest",
-        types.SimpleNamespace(register_executed_output=_register, register_cached_output=_register),
-    )
     os.makedirs(_BASE, exist_ok=True)
     monkeypatch.setattr(folder_paths, "get_directory_by_type", lambda t: _BASE)
     monkeypatch.setattr(execution, "get_progress_state", lambda: _NoProgress())
     monkeypatch.setitem(nodes.NODE_CLASS_MAPPINGS, "AsyncUINode", _AsyncUINode)
 
-    return execution
+    return execution, InMemoryAssets()
 
 
-async def _drive_async_reentry(execution):
+async def _drive_async_reentry(execution, asset_manager):
     from comfy_execution.graph import DynamicPrompt
 
     unique_id = "1"
@@ -139,7 +126,9 @@ async def _drive_async_reentry(execution):
 
     common = (server, dynprompt, caches, unique_id, {}, executed, "job-1", exec_list)
 
-    r1, _, _ = await execution.execute(*common, pending_subgraph_results, pending_async_nodes, ui_outputs)
+    r1, _, _ = await execution.execute(
+        *common, pending_subgraph_results, pending_async_nodes, ui_outputs, asset_manager
+    )
     ui_outputs_had_uid_after_entry1 = unique_id in ui_outputs
 
     tasks = [t for t in pending_async_nodes.get(unique_id, []) if isinstance(t, asyncio.Task)]
@@ -148,7 +137,9 @@ async def _drive_async_reentry(execution):
     for _ in range(3):
         await asyncio.sleep(0)
 
-    r2, _, _ = await execution.execute(*common, pending_subgraph_results, pending_async_nodes, ui_outputs)
+    r2, _, _ = await execution.execute(
+        *common, pending_subgraph_results, pending_async_nodes, ui_outputs, asset_manager
+    )
 
     cached = caches.outputs.store.get(unique_id)
     return {
@@ -162,10 +153,10 @@ async def _drive_async_reentry(execution):
 
 
 def test_async_reentry_keeps_cache_id_free(execution_env):
-    execution = execution_env
+    execution, asset_manager = execution_env
     from execution import ExecutionResult
 
-    obs = asyncio.run(_drive_async_reentry(execution))
+    obs = asyncio.run(_drive_async_reentry(execution, asset_manager))
 
     assert obs["r1"] == ExecutionResult.PENDING
     assert obs["r2"] == ExecutionResult.SUCCESS
@@ -177,3 +168,4 @@ def test_async_reentry_keeps_cache_id_free(execution_env):
 
     assert obs["cache_entry"] is not None
     assert obs["cache_ids"] == []
+    assert [call.method for call in asset_manager.calls] == ["register_executed_output"]

tests-unit/execution_test/test_inmemory_assets.py

@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, replace
+from typing import TYPE_CHECKING, Any, Callable, TypeAlias
+
+from aiohttp import web
+
+from app.assets.manager import AssetManager
+from app.assets.services.schemas import RegisteredAsset, UploadAssetView
+
+if TYPE_CHECKING:
+    from app.user_manager import UserManager
+
+
+CallArgument: TypeAlias = str | bool | None
+EventSink: TypeAlias = Callable[[str, dict[str, Any]], None]
+
+
+@dataclass(frozen=True, slots=True)
+class AssetCall:
+    method: str
+    arguments: tuple[CallArgument, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class Delivery:
+    abs_path: str
+    asset: RegisteredAsset
+    superseded: bool = False
+
+
+class InMemoryAssets:
+
+    def __init__(self) -> None:
+        self.calls: list[AssetCall] = []
+        self.deliveries_by_path: dict[str, list[Delivery]] = {}
+        self._live_output_by_path: dict[str, RegisteredAsset] = {}
+        self._counter: int = 0
+
+    @property
+    def enabled(self) -> bool:
+        return True
+
+    def startup(self) -> None:
+        return None
+
+    def shutdown(self) -> None:
+        return None
+
+    def register_routes(
+        self, app: web.Application, user_manager: UserManager | None
+    ) -> None:
+        return None
+
+    def ensure_scan_started(self) -> None:
+        return None
+
+    def pause_background_scan(self) -> None:
+        return None
+
+    def queue_output_enrichment(self) -> None:
+        return None
+
+    def resume_background_scan(self) -> None:
+        return None
+
+    def register_upload(
+        self,
+        abs_path: str,
+        name: str,
+        upload_type: str,
+        subfolder: str,
+        *,
+        content_written: bool,
+    ) -> UploadAssetView | None:
+        return None
+
+    def register_executed_output(
+        self, abs_path: str, job_id: str | None
+    ) -> RegisteredAsset | None:
+        self._record("register_executed_output", abs_path, job_id)
+        deliveries = self.deliveries_by_path.setdefault(abs_path, [])
+        for index, delivery in enumerate(deliveries):
+            deliveries[index] = replace(delivery, superseded=True)
+
+        asset_id = self._next_asset_id()
+        asset = RegisteredAsset(
+            id=asset_id,
+            content_id=f"content-{asset_id}",
+            job_id=job_id,
+            name=os.path.basename(abs_path),
+        )
+        deliveries.append(Delivery(abs_path=abs_path, asset=asset))
+        self._live_output_by_path[abs_path] = asset
+        return asset
+
+    def register_cached_output(
+        self, abs_path: str, job_id: str | None
+    ) -> RegisteredAsset | None:
+        self._record("register_cached_output", abs_path, job_id)
+        source = self._live_output_by_path.get(abs_path)
+        if source is None:
+            return None
+
+        asset = RegisteredAsset(
+            id=self._next_asset_id(),
+            content_id=source.content_id,
+            job_id=job_id,
+            name=os.path.basename(abs_path),
+        )
+        self.deliveries_by_path[abs_path].append(
+            Delivery(abs_path=abs_path, asset=asset)
+        )
+        return asset
+
+    def set_event_sink(self, sink: EventSink) -> None:
+        return None
+
+    def _record(self, method: str, *arguments: CallArgument) -> None:
+        self.calls.append(AssetCall(method, arguments))
+
+    def _next_asset_id(self) -> str:
+        self._counter += 1
+        return f"asset-{self._counter}"
+
+
+def test_conforms() -> None:
+    manager: AssetManager = InMemoryAssets()
+    assert manager.enabled, (
+        "structural smoke test: AssetManager is a non-runtime-checkable Protocol, so isinstance is unavailable"
+    )
+    manager.startup()
+    manager.shutdown()
+    manager.register_routes(web.Application(), None)
+    manager.ensure_scan_started()
+    manager.pause_background_scan()
+    manager.queue_output_enrichment()
+    manager.resume_background_scan()
+    assert (
+        manager.register_upload(
+            "/output/upload.png",
+            "upload.png",
+            "output",
+            "",
+            content_written=True,
+        )
+        is None
+    )
+    executed = manager.register_executed_output("/output/executed.png", "job-1")
+    assert executed is not None
+    cached = manager.register_cached_output("/output/executed.png", "job-2")
+    assert cached is not None
+
+    def sink(event: str, payload: dict[str, Any]) -> None:
+        return None
+
+    manager.set_event_sink(sink)
+
+
+def test_register_executed_output_supersedes_prior_deliveries() -> None:
+    manager = InMemoryAssets()
+    abs_path = "/output/image.png"
+
+    first = manager.register_executed_output(abs_path, "job-1")
+    assert first is not None
+    cached = manager.register_cached_output(abs_path, "job-2")
+    assert cached is not None
+    replacement = manager.register_executed_output(abs_path, "job-3")
+    assert replacement is not None
+
+    deliveries = manager.deliveries_by_path[abs_path]
+    assert [delivery.superseded for delivery in deliveries] == [True, True, False]
+    assert cached.content_id == first.content_id
+    assert replacement.id != first.id
+    assert replacement.content_id != first.content_id
+
+
+def test_register_cached_output_returns_none_without_live_content() -> None:
+    manager = InMemoryAssets()
+
+    cached = manager.register_cached_output("/output/missing.png", "job-1")
+
+    assert cached is None
+    assert manager.calls == [
+        AssetCall("register_cached_output", ("/output/missing.png", "job-1"))
+    ]

tests/test_asset_seeder.py

@@ -1,6 +1,7 @@
 """Tests for app.assets.seeder – enqueue_enrich and pending-queue behaviour."""
 
 import threading
+from typing import Any
 from unittest.mock import patch
 
 import pytest
@@ -14,6 +15,33 @@ def seeder():
     return _AssetSeeder()
 
 
+class TestEventSink:
+    def test_delivers_event_and_payload_to_sink(self, seeder):
+        received: list[tuple[str, dict[str, Any]]] = []
+
+        def record_event(event_type: str, data: dict[str, Any]) -> None:
+            received.append((event_type, data))
+
+        event_type = "assets.seed.started"
+        payload = {"roots": ["models"], "total": 1, "phase": "fast"}
+        seeder.set_event_sink(record_event)
+
+        seeder._emit_event(event_type, payload)
+
+        assert received == [(event_type, payload)]
+
+    def test_noop_when_no_sink_is_set(self, seeder):
+        assert seeder._emit_event("assets.seed.resumed", {}) is None
+
+    def test_swallows_raising_sink_exception(self, seeder):
+        def raise_from_sink(event_type: str, data: dict[str, Any]) -> None:
+            raise RuntimeError("boom")
+
+        seeder.set_event_sink(raise_from_sink)
+
+        assert seeder._emit_event("assets.seed.error", {"message": "failure"}) is None
+
+
 # ---------------------------------------------------------------------------
 # _reset_to_idle
 # ---------------------------------------------------------------------------