feat: Implement live asset management and deduplication logic in InfiniteScroll and asset handling

js/entry.js

@@ -40,6 +40,7 @@ import {
 } from "./features/bottomPanel/GeneratedFeedTab.js";
 import { createJobFinalizeQueue } from "./features/stacks/finalizeQueue.js";
 import { createExecutionAssetBuffer } from "./features/stacks/executionAssetBuffer.js";
+import { createLiveAssetGate } from "./features/stacks/liveAssetGate.js";
 import { extractOutputFiles } from "./utils/extractOutputFiles.js";
 import { post } from "./api/client.js";
 import { ENDPOINTS } from "./api/endpoints.js";
@@ -387,9 +388,9 @@ const executionStarts = createTTLCache({
     maxSize: EXECUTION_STARTS_MAX,
 });
 const executionAssetBuffer = createExecutionAssetBuffer();
+const liveAssetGate = createLiveAssetGate();
 let _lastStacksUpdateSignature = "";
 let _lastStacksUpdateAt = 0;
-const _pendingStackJobIds = new Set();
 
 function isJobTrackingDebugEnabled() {
     try {
@@ -421,6 +422,7 @@ const _stackFinalizeQueue = createJobFinalizeQueue({
                 job_id: targetJobId,
             });
             if (!res?.ok) {
+                liveAssetGate.clearPendingJob(targetJobId);
                 reportError(
                     new Error(String(res?.error || "Auto stack failed")),
                     "entry.execution_end.auto_stack",
@@ -451,11 +453,11 @@ function notifyStacksUpdated(detail = {}) {
     });
     try {
         const targeted = String(detail?.targeted_job_id || "").trim();
-        if (targeted) _pendingStackJobIds.delete(targeted);
+        if (targeted) liveAssetGate.clearPendingJob(targeted);
         const stacks = Array.isArray(detail?.stacks) ? detail.stacks : [];
         for (const item of stacks) {
             const jobId = String(item?.job_id || "").trim();
-            if (jobId) _pendingStackJobIds.delete(jobId);
+            if (jobId) liveAssetGate.clearPendingJob(jobId);
         }
     } catch (e) {
         console.debug?.(e);
@@ -622,7 +624,7 @@ function flushBufferedGenerationIndex(jobId) {
         file_count: files.length,
         files: files.map((item) => String(item?.filename || item?.filepath || "")),
     });
-    _pendingStackJobIds.add(promptId);
+    liveAssetGate.markPendingJob(promptId, { files, expectedCount: files.length });
     if (!files.length) {
         scheduleFinalizeExecutionStacks(promptId);
         return;
@@ -634,7 +636,7 @@ function flushBufferedGenerationIndex(jobId) {
     })
         .then((res) => {
             if (!res?.ok) {
-                _pendingStackJobIds.delete(promptId);
+                liveAssetGate.clearPendingJob(promptId);
                 reportError(
                     new Error(String(res?.error || "Indexing generated files failed")),
                     "entry.executed.index",
@@ -648,21 +650,17 @@ function flushBufferedGenerationIndex(jobId) {
             scheduleFinalizeExecutionStacks(promptId, 200);
         })
         .catch((error) => {
-            _pendingStackJobIds.delete(promptId);
+            liveAssetGate.clearPendingJob(promptId);
             reportError(error, "entry.executed.index");
         });
 }
 
-function shouldDeferLiveAssetEvent(detail) {
+function prepareLiveAssetEvent(detail) {
     try {
-        const jobId = String(detail?.job_id || "").trim();
-        if (!jobId) return false;
-        if (!_pendingStackJobIds.has(jobId)) return false;
-        const stackId = String(detail?.stack_id || "").trim();
-        return !stackId;
+        return liveAssetGate.prepare(detail);
     } catch (e) {
         console.debug?.(e);
-        return false;
+        return { detail, jobId: "", defer: false };
     }
 }
 
@@ -898,7 +896,7 @@ app.registerExtension({
                                 reason: String(event?.type || ""),
                             });
                             executionAssetBuffer.clear(promptId);
-                            _pendingStackJobIds.delete(promptId);
+                            liveAssetGate.clearPendingJob(promptId);
                         }
                         if (String(event?.type || "") === "execution_success") {
                             flushBufferedGenerationIndex(promptId);
@@ -1002,15 +1000,17 @@ app.registerExtension({
                     try {
                         const grid = getActiveGridContainer();
                         if (grid && event?.detail) {
-                            if (shouldDeferLiveAssetEvent(event.detail)) return;
-                            pushGeneratedAsset(event.detail);
+                            const liveEvent = prepareLiveAssetEvent(event.detail);
+                            if (liveEvent?.defer) return;
+                            const detail = liveEvent?.detail || event.detail;
+                            pushGeneratedAsset(detail);
                             // Only handle direct updates when the active grid shows output or "all" scope.
                             const scope = grid.dataset?.mjrScope || "output";
                             if (scope !== "output" && scope !== "all") return;
                             const query = grid.dataset?.mjrQuery || "*";
                             const canDirectUpsert = String(query).trim() === "*";
                             if (canDirectUpsert) {
-                                const handled = !!upsertAsset(grid, event.detail);
+                                const handled = !!upsertAsset(grid, detail);
                                 if (handled) {
                                     // Signal that we handled this event so handleCountersUpdate
                                     // skips its own reload within the next window.
@@ -1039,9 +1039,10 @@ app.registerExtension({
                     try {
                         const grid = getActiveGridContainer();
                         if (grid && event?.detail) {
-                            if (shouldDeferLiveAssetEvent(event.detail)) return;
-                            pushGeneratedAsset(event.detail);
-                            const detail = event.detail;
+                            const liveEvent = prepareLiveAssetEvent(event.detail);
+                            if (liveEvent?.defer) return;
+                            const detail = liveEvent?.detail || event.detail;
+                            pushGeneratedAsset(detail);
                             const assetId = String(detail?.id || "").trim();
                             const kind = String(detail?.kind || "").trim();
                             const filename = String(detail?.filename || "").trim();

js/features/grid/InfiniteScroll.js

@@ -224,6 +224,24 @@ export function getUpsertBatchState(gridContainer, upsertState) {
     return s;
 }
 
+function dedupeAssetsByKey(state, deps) {
+    const seenIds = new Set();
+    const seenKeys = new Set();
+    const deduped = [];
+    for (const asset of Array.isArray(state?.assets) ? state.assets : []) {
+        const assetId = asset?.id != null ? String(asset.id) : "";
+        const key = deps.assetKey(asset);
+        if (assetId && seenIds.has(assetId)) continue;
+        if (key && seenKeys.has(key)) continue;
+        if (assetId) seenIds.add(assetId);
+        if (key) seenKeys.add(key);
+        deduped.push(asset);
+    }
+    state.assets = deduped;
+    state.assetIdSet = seenIds;
+    state.seenKeys = seenKeys;
+}
+
 /**
  * Flush all pending upsert operations into the grid.
  *
@@ -288,8 +306,15 @@ export function flushUpsertBatch(gridContainer, deps) {
             const existingIndex = state.assets.findIndex((a) => String(a.id) === assetId);
             if (existingIndex > -1) {
                 const existingAsset = state.assets[existingIndex];
+                const previousKey = deps.assetKey(existingAsset);
                 Object.assign(existingAsset, asset);
-                state.assets[existingIndex] = { ...existingAsset };
+                const mergedAsset = { ...existingAsset };
+                const nextKey = deps.assetKey(mergedAsset);
+                state.assets[existingIndex] = mergedAsset;
+                if (previousKey && previousKey !== nextKey) {
+                    state.seenKeys?.delete?.(previousKey);
+                    if (nextKey) state.seenKeys?.add?.(nextKey);
+                }
                 modified = true;
             } else {
                 const alreadySeen =
@@ -306,6 +331,7 @@ export function flushUpsertBatch(gridContainer, deps) {
             }
         }
         if (modified && vg) {
+            dedupeAssetsByKey(state, deps);
             try {
                 gridContainer.dataset.mjrHiddenPngSiblings = String(
                     Number(state.hiddenPngSiblings || 0) || 0,

js/features/stacks/liveAssetGate.js

@@ -0,0 +1,79 @@
+function _normalizeJobId(jobId) {
+    return String(jobId || "").trim();
+}
+
+function _assetFileKey(detail) {
+    const filename = String(detail?.filename || "").trim();
+    const filepath = String(detail?.filepath || "").trim();
+    const subfolder = String(detail?.subfolder || "").trim();
+    const type = String(detail?.type || detail?.source || "output")
+        .trim()
+        .toLowerCase();
+    const rootId = String(detail?.root_id || detail?.custom_root_id || "").trim();
+    return `${type}|${rootId}|${subfolder}|${filename || filepath}`;
+}
+
+function _copyWithJobId(detail, jobId) {
+    if (!jobId) return detail;
+    if (String(detail?.job_id || "").trim() === jobId) return detail;
+    return { ...(detail || {}), job_id: jobId };
+}
+
+export function createLiveAssetGate() {
+    const pendingJobs = new Map();
+
+    function markPendingJob(jobId, { files = [], expectedCount = null } = {}) {
+        const normalizedJobId = _normalizeJobId(jobId);
+        if (!normalizedJobId) return;
+        const fileKeys = new Set();
+        for (const file of Array.isArray(files) ? files : []) {
+            const key = _assetFileKey(file);
+            if (key) fileKeys.add(key);
+        }
+        pendingJobs.set(normalizedJobId, {
+            expectedCount: Math.max(0, Number(expectedCount ?? fileKeys.size) || 0),
+            fileKeys,
+        });
+    }
+
+    function clearPendingJob(jobId) {
+        const normalizedJobId = _normalizeJobId(jobId);
+        if (!normalizedJobId) return;
+        pendingJobs.delete(normalizedJobId);
+    }
+
+    function resolveJobId(detail) {
+        const explicit = _normalizeJobId(detail?.job_id);
+        if (explicit) return explicit;
+        const key = _assetFileKey(detail);
+        if (!key) return "";
+        for (const [jobId, meta] of pendingJobs.entries()) {
+            if (meta?.fileKeys?.has?.(key)) return jobId;
+        }
+        return "";
+    }
+
+    function prepare(detail) {
+        const jobId = resolveJobId(detail);
+        if (!jobId) return { detail, jobId: "", defer: false };
+        const normalizedDetail = _copyWithJobId(detail, jobId);
+        const pending = pendingJobs.get(jobId);
+        if (!pending) return { detail: normalizedDetail, jobId, defer: false };
+        const stackId = String(normalizedDetail?.stack_id || "").trim();
+        const expectedCount = Math.max(0, Number(pending?.expectedCount || 0) || 0);
+        const isMultiOutput = expectedCount > 1;
+        return {
+            detail: normalizedDetail,
+            jobId,
+            defer: isMultiOutput && !stackId,
+        };
+    }
+
+    return {
+        markPendingJob,
+        clearPendingJob,
+        resolveJobId,
+        prepare,
+        _pendingJobs: pendingJobs,
+    };
+}

js/tests/infinite_scroll_upsert.vitest.mjs

@@ -0,0 +1,79 @@
+import { describe, expect, it, vi } from "vitest";
+
+import {
+    flushUpsertBatch,
+    getUpsertBatchState,
+    upsertAsset,
+} from "../features/grid/InfiniteScroll.js";
+
+function assetKey(asset) {
+    const stackId = String(asset?.stack_id || "").trim();
+    if (stackId) return `stack:${stackId}`;
+    return `id:${String(asset?.id || "")}`;
+}
+
+describe("InfiniteScroll upsert", () => {
+    it("refreshes seen keys when an existing asset gains a stack id", () => {
+        const gridContainer = { dataset: { mjrSort: "mtime_desc" } };
+        const state = {
+            assets: [{ id: 1, filename: "one.png", mtime: 10 }],
+            seenKeys: new Set(["id:1"]),
+            assetIdSet: new Set(["1"]),
+            hiddenPngSiblings: 0,
+        };
+        const setItems = vi.fn();
+        const deps = {
+            upsertState: new Map(),
+            getOrCreateState: () => state,
+            ensureVirtualGrid: () => ({ setItems }),
+            assetKey,
+            loadMajoorSettings: () => ({}),
+        };
+
+        const batchState = getUpsertBatchState(gridContainer, deps.upsertState);
+        batchState.pending.set("1", { id: 1, stack_id: "stack-a", filename: "one.png", mtime: 10 });
+
+        flushUpsertBatch(gridContainer, deps);
+
+        expect(state.seenKeys.has("id:1")).toBe(false);
+        expect(state.seenKeys.has("stack:stack-a")).toBe(true);
+        expect(state.assets).toHaveLength(1);
+        expect(state.assets[0].stack_id).toBe("stack-a");
+        expect(setItems).toHaveBeenCalledTimes(1);
+    });
+
+    it("dedupes cards after a key change collides with an existing stack card", () => {
+        const gridContainer = { dataset: { mjrSort: "mtime_desc" } };
+        const state = {
+            assets: [
+                { id: 1, filename: "one.png", mtime: 20 },
+                { id: 2, filename: "two.png", mtime: 10, stack_id: "stack-a" },
+            ],
+            seenKeys: new Set(["id:1", "stack:stack-a"]),
+            assetIdSet: new Set(["1", "2"]),
+            hiddenPngSiblings: 0,
+        };
+        const setItems = vi.fn();
+        const deps = {
+            upsertState: new Map(),
+            getOrCreateState: () => state,
+            ensureVirtualGrid: () => ({ setItems }),
+            assetKey,
+            loadMajoorSettings: () => ({}),
+            maxBatchSize: 50,
+            debounceMs: 0,
+        };
+
+        expect(upsertAsset(gridContainer, { id: 1, stack_id: "stack-a", mtime: 30 }, deps)).toBe(
+            true,
+        );
+        flushUpsertBatch(gridContainer, deps);
+
+        expect(state.assets).toHaveLength(1);
+        expect(state.assets[0].id).toBe(1);
+        expect(state.seenKeys.has("stack:stack-a")).toBe(true);
+        expect(state.assetIdSet.has("1")).toBe(true);
+        expect(state.assetIdSet.has("2")).toBe(false);
+        expect(setItems).toHaveBeenCalledTimes(1);
+    });
+});

js/tests/live_asset_gate.vitest.mjs

@@ -0,0 +1,73 @@
+import { describe, expect, it } from "vitest";
+
+import { createLiveAssetGate } from "../features/stacks/liveAssetGate.js";
+
+describe("createLiveAssetGate", () => {
+    it("defers multi-output assets until stack assignment is available", () => {
+        const gate = createLiveAssetGate();
+        gate.markPendingJob("job-multi", {
+            expectedCount: 2,
+            files: [
+                { filename: "a.png", subfolder: "gen", type: "output" },
+                { filename: "b.png", subfolder: "gen", type: "output" },
+            ],
+        });
+
+        const pending = gate.prepare({
+            filename: "a.png",
+            subfolder: "gen",
+            type: "output",
+            job_id: "job-multi",
+        });
+        expect(pending.defer).toBe(true);
+        expect(pending.detail.job_id).toBe("job-multi");
+
+        const stacked = gate.prepare({
+            filename: "a.png",
+            subfolder: "gen",
+            type: "output",
+            job_id: "job-multi",
+            stack_id: "42",
+        });
+        expect(stacked.defer).toBe(false);
+        expect(stacked.detail.stack_id).toBe("42");
+    });
+
+    it("allows mono-output assets through normal upsert while preserving job id", () => {
+        const gate = createLiveAssetGate();
+        gate.markPendingJob("job-single", {
+            expectedCount: 1,
+            files: [{ filename: "solo.png", subfolder: "gen", type: "output" }],
+        });
+
+        const prepared = gate.prepare({
+            filename: "solo.png",
+            subfolder: "gen",
+            type: "output",
+        });
+
+        expect(prepared.defer).toBe(false);
+        expect(prepared.jobId).toBe("job-single");
+        expect(prepared.detail.job_id).toBe("job-single");
+        expect(prepared.detail.stack_id).toBeUndefined();
+    });
+
+    it("stops deferring once the pending job is cleared", () => {
+        const gate = createLiveAssetGate();
+        gate.markPendingJob("job-clear", {
+            expectedCount: 3,
+            files: [{ filename: "x.png", subfolder: "gen", type: "output" }],
+        });
+        gate.clearPendingJob("job-clear");
+
+        const prepared = gate.prepare({
+            filename: "x.png",
+            subfolder: "gen",
+            type: "output",
+            job_id: "job-clear",
+        });
+
+        expect(prepared.defer).toBe(false);
+        expect(prepared.detail.job_id).toBe("job-clear");
+    });
+});

mjr_am_backend/adapters/db/schema.py

@@ -137,6 +137,7 @@ COLUMN_DEFINITIONS = {
         ("rating", "rating INTEGER DEFAULT 0"),
         ("tags", "tags TEXT DEFAULT ''"),
         ("tags_text", "tags_text TEXT DEFAULT ''"),
+        ("metadata_text", "metadata_text TEXT DEFAULT ''"),
         ("workflow_hash", "workflow_hash TEXT"),
         ("has_workflow", "has_workflow BOOLEAN DEFAULT 0"),
         ("has_generation_data", "has_generation_data BOOLEAN DEFAULT 0"),

mjr_am_backend/routes/handlers/assets_impl.py

@@ -946,7 +946,8 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
         asset_row = await _query_one(
             f"""
             SELECT id, filename, subfolder, filepath, source, root_id, kind, ext, size, mtime,
-                   width, height, duration, enhanced_caption,
+                   width, height, duration, enhanced_caption, job_id, stack_id,
+                   source_node_id, source_node_type,
                    created_at, updated_at, indexed_at, content_hash, phash, hash_state
             FROM assets
             WHERE {filepath_where}
@@ -1013,9 +1014,33 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
                 if asset:
                     await db.aexecute(
                         """
-                        INSERT OR REPLACE INTO assets
-                        (id, filename, subfolder, filepath, source, root_id, kind, ext, size, mtime, width, height, duration, enhanced_caption, created_at, updated_at, indexed_at, content_hash, phash, hash_state)
-                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                        INSERT INTO assets
+                        (id, filename, subfolder, filepath, source, root_id, kind, ext, size, mtime, width, height, duration, enhanced_caption, job_id, stack_id, source_node_id, source_node_type, created_at, updated_at, indexed_at, content_hash, phash, hash_state)
+                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                        ON CONFLICT(id) DO UPDATE SET
+                            filename = excluded.filename,
+                            subfolder = excluded.subfolder,
+                            filepath = excluded.filepath,
+                            source = excluded.source,
+                            root_id = excluded.root_id,
+                            kind = excluded.kind,
+                            ext = excluded.ext,
+                            size = excluded.size,
+                            mtime = excluded.mtime,
+                            width = excluded.width,
+                            height = excluded.height,
+                            duration = excluded.duration,
+                            enhanced_caption = excluded.enhanced_caption,
+                            job_id = excluded.job_id,
+                            stack_id = excluded.stack_id,
+                            source_node_id = excluded.source_node_id,
+                            source_node_type = excluded.source_node_type,
+                            created_at = excluded.created_at,
+                            updated_at = excluded.updated_at,
+                            indexed_at = excluded.indexed_at,
+                            content_hash = excluded.content_hash,
+                            phash = excluded.phash,
+                            hash_state = excluded.hash_state
                         """,
                         (
                             asset.get("id"),
@@ -1032,6 +1057,10 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
                             asset.get("height"),
                             asset.get("duration"),
                             asset.get("enhanced_caption"),
+                            asset.get("job_id"),
+                            asset.get("stack_id"),
+                            asset.get("source_node_id"),
+                            asset.get("source_node_type"),
                             asset.get("created_at"),
                             asset.get("updated_at"),
                             asset.get("indexed_at"),
@@ -1043,9 +1072,19 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
                 if asset_meta:
                     await db.aexecute(
                         """
-                        INSERT OR REPLACE INTO asset_metadata
+                        INSERT INTO asset_metadata
                         (asset_id, rating, tags, tags_text, workflow_hash, has_workflow, has_generation_data, metadata_quality, metadata_raw, metadata_text)
                         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                        ON CONFLICT(asset_id) DO UPDATE SET
+                            rating = excluded.rating,
+                            tags = excluded.tags,
+                            tags_text = excluded.tags_text,
+                            workflow_hash = excluded.workflow_hash,
+                            has_workflow = excluded.has_workflow,
+                            has_generation_data = excluded.has_generation_data,
+                            metadata_quality = excluded.metadata_quality,
+                            metadata_raw = excluded.metadata_raw,
+                            metadata_text = excluded.metadata_text
                         """,
                         (
                             asset_meta.get("asset_id"),
@@ -1063,9 +1102,15 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
                 if scan_journal:
                     await db.aexecute(
                         """
-                        INSERT OR REPLACE INTO scan_journal
+                        INSERT INTO scan_journal
                         (filepath, dir_path, state_hash, mtime, size, last_seen)
                         VALUES (?, ?, ?, ?, ?, ?)
+                        ON CONFLICT(filepath) DO UPDATE SET
+                            dir_path = excluded.dir_path,
+                            state_hash = excluded.state_hash,
+                            mtime = excluded.mtime,
+                            size = excluded.size,
+                            last_seen = excluded.last_seen
                         """,
                         (
                             scan_journal.get("filepath"),
@@ -1079,9 +1124,14 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
                 if metadata_cache:
                     await db.aexecute(
                         """
-                        INSERT OR REPLACE INTO metadata_cache
+                        INSERT INTO metadata_cache
                         (filepath, state_hash, metadata_hash, metadata_raw, last_updated)
                         VALUES (?, ?, ?, ?, ?)
+                        ON CONFLICT(filepath) DO UPDATE SET
+                            state_hash = excluded.state_hash,
+                            metadata_hash = excluded.metadata_hash,
+                            metadata_raw = excluded.metadata_raw,
+                            last_updated = excluded.last_updated
                         """,
                         (
                             metadata_cache.get("filepath"),

tests/database/test_schema_heal.py

@@ -74,6 +74,7 @@ async def test_schema_self_heals_missing_columns(tmp_path):
         for name in [
             "tags",
             "tags_text",
+            "metadata_text",
             "workflow_hash",
             "has_workflow",
             "has_generation_data",
@@ -86,7 +87,7 @@ async def test_schema_self_heals_missing_columns(tmp_path):
         assert (await db.aquery("SELECT source, root_id, enhanced_caption, content_hash, phash, hash_state FROM assets LIMIT 0")).ok
         assert (await db.aquery(
             """
-            SELECT tags, metadata_raw, has_workflow, has_generation_data, metadata_quality
+            SELECT tags, tags_text, metadata_text, metadata_raw, has_workflow, has_generation_data, metadata_quality
             FROM asset_metadata
             LIMIT 0
             """