From 44c0c57f6ecca2dfe289e09dac50832072f9964f Mon Sep 17 00:00:00 2001 From: Ewald <129623145+MajoorWaldi@users.noreply.github.com> Date: Sun, 29 Mar 2026 01:03:49 +0100 Subject: [PATCH] feat: Add asset management and real-time listener functionalities - Implemented entry UI registration for the Assets Manager, including sidebar and bottom panel integration. - Created real-time listeners for execution events, asset updates, and enrichment status. - Developed execution runtime controller to manage job tracking, asset buffering, and indexing. - Introduced lazy initialization for vector services to optimize resource usage. - Added HTTP handlers for plugin management, enabling listing, enabling, and reloading of plugins. --- .gitignore | 4 + docs/AI_FEATURES.md | 4 +- js/api/client.js | 8 + js/api/endpoints.js | 1 + js/app/bootstrap.js | 31 +- js/app/config.js | 1 + js/app/settings/SettingsPanel.js | 12 +- js/app/settings/settingsCore.js | 33 + js/app/settings/settingsScanning.js | 46 +- js/entry.js | 1420 ++--------------- js/features/bottomPanel/GeneratedFeedTab.js | 23 +- js/features/grid/GridView.js | 2 + js/features/grid/GridView_impl.js | 81 + js/features/panel/AssetsManagerPanel.js | 20 + .../panel/controllers/gridController.js | 6 +- js/features/runtime/entryDebugApi.js | 50 + js/features/runtime/entryRuntimeLifecycle.js | 186 +++ js/features/runtime/entryUiRegistration.js | 151 ++ .../runtime/registerRealtimeListeners.js | 305 ++++ .../stacks/executionRuntimeController.js | 558 +++++++ mjr_am_backend/config.py | 30 +- mjr_am_backend/deps.py | 24 +- mjr_am_backend/features/index/service.py | 33 +- .../features/index/vector_runtime.py | 66 + mjr_am_backend/features/index/watcher.py | 33 + mjr_am_backend/features/runtime/bootstrap.py | 4 + mjr_am_backend/routes/handlers/__init__.py | 2 + mjr_am_backend/routes/handlers/health.py | 70 + mjr_am_backend/routes/handlers/plugins.py | 67 + mjr_am_backend/routes/handlers/scan.py | 147 +- mjr_am_backend/routes/handlers/stacks.py | 11 + .../routes/handlers/vector_search.py | 63 +- mjr_am_backend/routes/registry.py | 8 + mjr_am_backend/settings.py | 98 +- tests/features/test_scan_routes_more.py | 13 + 35 files changed, 2197 insertions(+), 1414 deletions(-) create mode 100644 js/features/runtime/entryDebugApi.js create mode 100644 js/features/runtime/entryRuntimeLifecycle.js create mode 100644 js/features/runtime/entryUiRegistration.js create mode 100644 js/features/runtime/registerRealtimeListeners.js create mode 100644 js/features/stacks/executionRuntimeController.js create mode 100644 mjr_am_backend/features/index/vector_runtime.py create mode 100644 mjr_am_backend/routes/handlers/plugins.py diff --git a/.gitignore b/.gitignore index 05cab401..7936e44e 100644 --- a/.gitignore +++ b/.gitignore @@ -198,3 +198,7 @@ docs/QUALITY_IMPROVEMENT_PLAN.md docs/COMFYUI_CORE_ANALYSIS.md .github/workflows/pr-review.yml scripts/pr_reviewer.py +zz_test.txt +unsaved_329.txt +draftindex_329.txt +draft0348_tail.txt diff --git a/docs/AI_FEATURES.md b/docs/AI_FEATURES.md index 8023f45b..94e13c93 100644 --- a/docs/AI_FEATURES.md +++ b/docs/AI_FEATURES.md @@ -2,8 +2,8 @@ **Comprehensive guide to AI-powered features in Majoor Assets Manager** -*Version: 2.4.0* -*Last Updated: March 5, 2026* +*Version: 2.4.2* +*Last Updated: March 21, 2026* --- diff --git a/js/api/client.js b/js/api/client.js index f7b93ff1..442fee95 100644 --- a/js/api/client.js +++ b/js/api/client.js @@ -943,6 +943,14 @@ export async function setVectorSearchSettings(enabled = true) { return post(ENDPOINTS.SETTINGS_VECTOR_SEARCH, { enabled: !!enabled }); } +export async function getExecutionGroupingSettings() { + return get(ENDPOINTS.SETTINGS_EXECUTION_GROUPING); +} + +export async function setExecutionGroupingSettings(enabled = true) { + return post(ENDPOINTS.SETTINGS_EXECUTION_GROUPING, { enabled: !!enabled }); +} + export async function getHuggingFaceSettings() { return get(ENDPOINTS.SETTINGS_HUGGINGFACE); } diff --git a/js/api/endpoints.js b/js/api/endpoints.js index 6ab57037..c0f179c5 100644 --- a/js/api/endpoints.js +++ b/js/api/endpoints.js @@ -41,6 +41,7 @@ export const ENDPOINTS = { SETTINGS_OUTPUT_DIRECTORY: "/mjr/am/settings/output-directory", SETTINGS_METADATA_FALLBACK: "/mjr/am/settings/metadata-fallback", SETTINGS_VECTOR_SEARCH: "/mjr/am/settings/vector-search", + SETTINGS_EXECUTION_GROUPING: "/mjr/am/settings/execution-grouping", SETTINGS_HUGGINGFACE: "/mjr/am/settings/huggingface", SETTINGS_AI_LOGGING: "/mjr/am/settings/ai-logging", diff --git a/js/app/bootstrap.js b/js/app/bootstrap.js index 4fae233c..8bb06d47 100644 --- a/js/app/bootstrap.js +++ b/js/app/bootstrap.js @@ -8,15 +8,44 @@ import { APP_CONFIG } from "./config.js"; // Auto-scan state (once per session) let startupScanDone = false; +let startupScanTimer = null; + +function isExecutionIdle() { + try { + const runtime = window?.__MJR_EXECUTION_RUNTIME__; + return !String(runtime?.active_prompt_id || "").trim(); + } catch (error) { + console.debug?.(error); + return true; + } +} /** * Trigger background scan as soon as the ComfyUI frontend loads (if enabled). * This helps warming the DB before the Assets Manager panel is opened. */ -export async function triggerStartupScan() { +export async function triggerStartupScan(options = {}) { if (!APP_CONFIG || typeof APP_CONFIG !== "object") return; if (!APP_CONFIG.AUTO_SCAN_ON_STARTUP) return; if (startupScanDone) return; + const delayMs = Math.max(0, Number(options?.delayMs) || 0); + const idleOnly = options?.idleOnly !== false; + if (delayMs > 0) { + if (startupScanTimer) clearTimeout(startupScanTimer); + startupScanTimer = setTimeout(() => { + startupScanTimer = null; + void triggerStartupScan({ ...options, delayMs: 0 }); + }, delayMs); + return; + } + if (idleOnly && !isExecutionIdle()) { + if (startupScanTimer) clearTimeout(startupScanTimer); + startupScanTimer = setTimeout(() => { + startupScanTimer = null; + void triggerStartupScan({ ...options, delayMs: 0 }); + }, 5000); + return; + } startupScanDone = true; try { diff --git a/js/app/config.js b/js/app/config.js index a608ef6e..a6742695 100644 --- a/js/app/config.js +++ b/js/app/config.js @@ -73,6 +73,7 @@ export const APP_DEFAULTS = Object.freeze({ // Auto-scan on startup AUTO_SCAN_ON_STARTUP: false, + EXECUTION_GROUPING_ENABLED: true, // Rating/tags hydration (grid) RT_HYDRATE_CONCURRENCY: 2, diff --git a/js/app/settings/SettingsPanel.js b/js/app/settings/SettingsPanel.js index e0215ee3..8946f8e8 100644 --- a/js/app/settings/SettingsPanel.js +++ b/js/app/settings/SettingsPanel.js @@ -9,7 +9,7 @@ import { safeDispatchCustomEvent } from "../../utils/events.js"; import { initI18n, setFollowComfyLanguage, startComfyLanguageSync } from "../i18n.js"; import { debounce } from "../../utils/debounce.js"; -import { getWatcherStatus, toggleWatcher } from "../../api/client.js"; +import { getWatcherStatus } from "../../api/client.js"; import { loadMajoorSettings, @@ -17,6 +17,7 @@ import { applySettingsToConfig, syncBackendSecuritySettings, syncBackendVectorSearchSettings, + syncBackendExecutionGroupingSettings, } from "./settingsCore.js"; import { startRuntimeStatusDashboard } from "./settingsRuntime.js"; import { registerGridSettings } from "./settingsGrid.js"; @@ -189,15 +190,8 @@ function ensureMajoorSettingsContext(app, onApplied, { initRuntime = false } = { startComfyLanguageSync(runtimeApp); void syncBackendSecuritySettings(); void syncBackendVectorSearchSettings(); + void syncBackendExecutionGroupingSettings(); startRuntimeStatusDashboard(); - try { - const desired = !!settings?.watcher?.enabled; - setTimeout(() => { - toggleWatcher(desired).catch(() => {}); - }, 0); - } catch (e) { - console.debug?.(e); - } _settingsRuntimeInitialized = true; } return _settingsContext; diff --git a/js/app/settings/settingsCore.js b/js/app/settings/settingsCore.js index aca0329a..d4eb062b 100644 --- a/js/app/settings/settingsCore.js +++ b/js/app/settings/settingsCore.js @@ -8,6 +8,7 @@ import { getSecuritySettings, bootstrapSecurityToken, getVectorSearchSettings, + getExecutionGroupingSettings, } from "../../api/client.js"; import { safeDispatchCustomEvent } from "../../utils/events.js"; import { SettingsStore } from "./SettingsStore.js"; @@ -140,6 +141,9 @@ export const DEFAULT_SETTINGS = { vectorSearchEnabled: true, verboseAiLogs: false, }, + executionGrouping: { + enabled: APP_DEFAULTS.EXECUTION_GROUPING_ENABLED, + }, workflowMinimap: { enabled: false, nodeColors: true, @@ -211,6 +215,7 @@ export const loadMajoorSettings = () => { "cache", "search", "ai", + "executionGrouping", "workflowMinimap", "ui", "security", @@ -292,6 +297,9 @@ export const applySettingsToConfig = (settings) => { ); APP_CONFIG.DEFAULT_PAGE_SIZE = pageSize; APP_CONFIG.AUTO_SCAN_ON_STARTUP = !!settings.autoScan?.onStartup; + APP_CONFIG.EXECUTION_GROUPING_ENABLED = !!( + settings.executionGrouping?.enabled ?? APP_DEFAULTS.EXECUTION_GROUPING_ENABLED + ); APP_CONFIG.STATUS_POLL_INTERVAL = Math.max( 1000, Number(settings.status?.pollInterval) || APP_DEFAULTS.STATUS_POLL_INTERVAL, @@ -652,3 +660,28 @@ export async function syncBackendVectorSearchSettings() { console.warn("[Majoor] failed to sync backend vector search settings", error); } } + +export async function syncBackendExecutionGroupingSettings() { + try { + const res = await getExecutionGroupingSettings(); + if (!res?.ok) return; + const prefs = res.data?.prefs; + if (!prefs || typeof prefs !== "object") return; + + const settings = loadMajoorSettings(); + settings.executionGrouping = settings.executionGrouping || {}; + settings.executionGrouping.enabled = _safeBool( + prefs.enabled, + settings.executionGrouping.enabled ?? APP_DEFAULTS.EXECUTION_GROUPING_ENABLED, + ); + saveMajoorSettings(settings); + applySettingsToConfig(settings); + safeDispatchCustomEvent( + "mjr-settings-changed", + { key: "executionGrouping.enabled" }, + { warnPrefix: "[Majoor]" }, + ); + } catch (error) { + console.warn("[Majoor] failed to sync backend execution grouping settings", error); + } +} diff --git a/js/app/settings/settingsScanning.js b/js/app/settings/settingsScanning.js index 2704d56a..f0505c92 100644 --- a/js/app/settings/settingsScanning.js +++ b/js/app/settings/settingsScanning.js @@ -3,7 +3,12 @@ */ import { APP_DEFAULTS } from "../config.js"; -import { toggleWatcher, getWatcherSettings, updateWatcherSettings } from "../../api/client.js"; +import { + toggleWatcher, + getWatcherSettings, + updateWatcherSettings, + setExecutionGroupingSettings, +} from "../../api/client.js"; import { comfyToast } from "../toast.js"; import { t } from "../i18n.js"; import { _safeNum } from "./settingsUtils.js"; @@ -22,6 +27,45 @@ const SETTINGS_CATEGORY = "Majoor Assets Manager"; export function registerScanningSettings(safeAddSetting, settings, notifyApplied) { const cat = (section, label) => [SETTINGS_CATEGORY, section, label]; + safeAddSetting({ + id: `${SETTINGS_PREFIX}.ExecutionGrouping.Enabled`, + category: cat(t("cat.scanning"), "Execution grouping"), + name: "Execution job/stack grouping", + tooltip: + "Enable or disable all live job_id / stack_id tracking, grouping, and stack finalization logic.", + type: "boolean", + defaultValue: !!( + settings.executionGrouping?.enabled ?? APP_DEFAULTS.EXECUTION_GROUPING_ENABLED + ), + onChange: async (value) => { + const previous = !!( + settings.executionGrouping?.enabled ?? APP_DEFAULTS.EXECUTION_GROUPING_ENABLED + ); + const normalized = !!value; + settings.executionGrouping = settings.executionGrouping || {}; + settings.executionGrouping.enabled = normalized; + saveMajoorSettings(settings); + applySettingsToConfig(settings); + notifyApplied("executionGrouping.enabled"); + try { + const res = await setExecutionGroupingSettings(normalized); + if (!res?.ok) { + throw new Error(res?.error || "Failed to update execution grouping setting"); + } + settings.executionGrouping.enabled = !!res?.data?.prefs?.enabled; + saveMajoorSettings(settings); + applySettingsToConfig(settings); + notifyApplied("executionGrouping.enabled"); + } catch (error) { + settings.executionGrouping.enabled = previous; + saveMajoorSettings(settings); + applySettingsToConfig(settings); + notifyApplied("executionGrouping.enabled"); + comfyToast(error?.message || "Failed to update execution grouping setting", "error"); + } + }, + }); + safeAddSetting({ id: `${SETTINGS_PREFIX}.AutoScan.OnStartup`, category: cat(t("cat.scanning"), t("setting.scan.startup.name").replace("Majoor: ", "")), diff --git a/js/entry.js b/js/entry.js index ec7b10cb..cb6b8a9b 100644 --- a/js/entry.js +++ b/js/entry.js @@ -8,11 +8,7 @@ import { checkMajoorVersion } from "./app/versionCheck.js"; import { ensureStyleLoaded } from "./app/style.js"; import { buildMajoorSettings, registerMajoorSettings } from "./app/settings.js"; import { - activateSidebarTabCompat, getComfyApi, - registerBottomPanelTabCompat, - registerSidebarTabCompat, - registerCommandCompat, setComfyApi, setComfyApp, waitForComfyApi, @@ -29,18 +25,12 @@ import { NODE_STREAM_REACTIVATION_DOC, } from "./features/viewer/nodeStream/nodeStreamFeatureFlag.js"; import { loadAssets, upsertAsset, removeAssetsFromGrid } from "./features/grid/GridView.js"; +import { getActiveGridContainer } from "./features/panel/AssetsManagerPanel.js"; import { - renderAssetsManager, - getActiveGridContainer, -} from "./features/panel/AssetsManagerPanel.js"; -import { - getGeneratedFeedBottomPanelTab, pushGeneratedAsset, refreshGeneratedFeedHosts, } 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 { createExecutionRuntimeController } from "./features/stacks/executionRuntimeController.js"; import { extractOutputFiles } from "./utils/extractOutputFiles.js"; import { post } from "./api/client.js"; import { ENDPOINTS } from "./api/endpoints.js"; @@ -48,27 +38,41 @@ import { comfyToast } from "./app/toast.js"; import { t } from "./app/i18n.js"; import { getEnrichmentState, setEnrichmentState } from "./app/runtimeState.js"; import { reportError } from "./utils/logging.js"; -import { createTTLCache } from "./utils/ttlCache.js"; import { app } from "../../scripts/app.js"; +import { registerRealtimeListeners } from "./features/runtime/registerRealtimeListeners.js"; +import { exposeDebugApis } from "./features/runtime/entryDebugApi.js"; +import { + buildBottomPanelTabs, + buildNativeCommands, + getMajoorNodeMenuItems, + registerAssetsSidebar, + registerGeneratedBottomPanel, + registerNativeCommands, +} from "./features/runtime/entryUiRegistration.js"; +import { + ENTRY_RUNTIME_KEY, + cleanupEntryRuntime, + installEntryRuntimeController, + registerCleanableListener as registerCleanableListener, + removeApiHandlers, + removeRuntimeWindowHandlers, +} from "./features/runtime/entryRuntimeLifecycle.js"; -// Lazy-loaded modules — resolved on first use to speed up startup. /** @type {import("./features/viewer/LiveStreamTracker.js") | null} */ -let _liveStreamMod = null; +let liveStreamModule = null; /** @type {import("./features/viewer/nodeStream/NodeStreamController.js") | null} */ -let _nodeStreamMod = null; +let nodeStreamModule = null; const UI_FLAGS = { useComfyThemeUI: true, }; -// Runtime cleanup key (used for hot-reload cleanup in ComfyUI) -const ENTRY_RUNTIME_KEY = "__MJR_ENTRY_RUNTIME__"; const SIDEBAR_TAB_ID = "majoor-assets"; const EXECUTION_RUNTIME_KEY = "__MJR_EXECUTION_RUNTIME__"; function ensureExecutionRuntime() { try { - if (typeof window === "undefined") + if (typeof window === "undefined") { return { active_prompt_id: null, queue_remaining: null, @@ -77,6 +81,7 @@ function ensureExecutionRuntime() { progress_max: null, cached_nodes: [], }; + } if (!window[EXECUTION_RUNTIME_KEY] || typeof window[EXECUTION_RUNTIME_KEY] !== "object") { window[EXECUTION_RUNTIME_KEY] = { active_prompt_id: null, @@ -104,615 +109,119 @@ function emitRuntimeStatus(partial = {}) { } } -function openAssetsManagerPanel(runtimeApp) { - const opened = activateSidebarTabCompat(runtimeApp, SIDEBAR_TAB_ID); - if (!opened) { - try { - window.dispatchEvent(new Event(EVENTS.OPEN_ASSETS_MANAGER)); - } catch (e) { - console.debug?.(e); - } - } - return opened; -} - -function triggerRefreshGrid() { - try { - window.dispatchEvent( - new CustomEvent(EVENTS.RELOAD_GRID, { detail: { reason: "command" } }), - ); - } catch (e) { - console.debug?.(e); - } -} - -function registerNativeCommands(runtimeApp) { - const commands = [ - { - id: "mjr.openAssetsManager", - label: t("manager.title", "Assets Manager"), - icon: "pi pi-folder-open", - function: () => openAssetsManagerPanel(runtimeApp), - }, - { - id: "mjr.scanAssets", - label: t("command.scanAssets", "Scan assets"), - icon: "pi pi-refresh", - function: () => triggerStartupScan(), - }, - { - id: "mjr.toggleFloatingViewer", - label: t("command.toggleFloatingViewer", "Toggle floating viewer"), - icon: "pi pi-images", - function: () => window.dispatchEvent(new Event(EVENTS.MFV_TOGGLE)), - }, - { - id: "mjr.refreshAssetsGrid", - label: t("command.refreshAssetsGrid", "Refresh assets grid"), - icon: "pi pi-sync", - function: () => triggerRefreshGrid(), - }, - ]; - for (const command of commands) { - registerCommandCompat(runtimeApp, command); - } -} - -function isMajoorTrackableNode(node) { - const comfyClass = String( - node?.comfyClass || node?.type || node?.constructor?.type || "", - ).trim(); - if (!comfyClass) return false; - return /save|load|preview/i.test(comfyClass); -} - -function nodeMenuEntry(label, callback) { - return { - content: label, - callback, - }; -} - -function removeApiHandlers(api) { - if (!api) return; - try { - if (api._mjrAssetUpdateReloadTimer) { - clearTimeout(api._mjrAssetUpdateReloadTimer); - api._mjrAssetUpdateReloadTimer = null; - } - if (api._mjrExecutedHandler) { - api.removeEventListener("executed", api._mjrExecutedHandler); - } - if (api._mjrAssetAddedHandler) { - api.removeEventListener("mjr-asset-added", api._mjrAssetAddedHandler); - } - if (api._mjrAssetUpdatedHandler) { - api.removeEventListener("mjr-asset-updated", api._mjrAssetUpdatedHandler); - } - if (api._mjrScanCompleteHandler) { - api.removeEventListener(EVENTS.SCAN_COMPLETE, api._mjrScanCompleteHandler); - } - if (api._mjrScanProgressHandler) { - api.removeEventListener(EVENTS.SCAN_PROGRESS, api._mjrScanProgressHandler); - } - if (api._mjrAssetIndexingHandler) { - api.removeEventListener(EVENTS.ASSET_INDEXING, api._mjrAssetIndexingHandler); - } - if (api._mjrAssetIndexedHandler) { - api.removeEventListener(EVENTS.ASSET_INDEXED, api._mjrAssetIndexedHandler); - } - if (api._mjrExecutionStartHandler) { - api.removeEventListener("execution_start", api._mjrExecutionStartHandler); - } - if (api._mjrExecutionEndHandler) { - api.removeEventListener("execution_success", api._mjrExecutionEndHandler); - api.removeEventListener("execution_error", api._mjrExecutionEndHandler); - api.removeEventListener("execution_interrupted", api._mjrExecutionEndHandler); - } - if (api._mjrStacksUpdatedHandler) { - api.removeEventListener("mjr.stacks.updated", api._mjrStacksUpdatedHandler); - } - if (api._mjrEnrichmentStatusHandler) { - api.removeEventListener(EVENTS.ENRICHMENT_STATUS, api._mjrEnrichmentStatusHandler); - } - if (api._mjrDbRestoreStatusHandler) { - api.removeEventListener(EVENTS.DB_RESTORE_STATUS, api._mjrDbRestoreStatusHandler); - } - if (api._mjrRuntimeStatusHandler) { - api.removeEventListener("progress", api._mjrRuntimeStatusHandler); - api.removeEventListener("status", api._mjrRuntimeStatusHandler); - api.removeEventListener(EVENTS.RUNTIME_STATUS, api._mjrRuntimeStatusHandler); - api.removeEventListener("execution_cached", api._mjrExecutionCachedHandler); - } - } catch (error) { - reportError(error, "entry.removeApiHandlers"); - } -} - -function removeRuntimeWindowHandlers(runtime) { - try { - const handler = runtime?.assetsDeletedHandler; - if (handler && typeof window !== "undefined") { - window.removeEventListener(EVENTS.ASSETS_DELETED, handler); - } - } catch (error) { - reportError(error, "entry.removeRuntimeWindowHandlers"); - } -} - -function cleanupEntryRuntime(runtime) { - if (!runtime || typeof runtime !== "object") return; - try { - const cleanups = Array.isArray(runtime._listenerCleanupFns) - ? runtime._listenerCleanupFns - : []; - for (const cleanup of cleanups) { +function setupLazyModules(runtimeApp) { + void import("./features/viewer/LiveStreamTracker.js") + .then((mod) => { + liveStreamModule = mod; try { - cleanup?.(); + mod.initLiveStreamTracker(runtimeApp); } catch (e) { - console.debug?.(e); - } - } - runtime._listenerCleanupFns = []; - } catch (e) { - console.warn("[MJR teardown]", e); - } - try { - const controllers = Array.isArray(runtime._cleanupControllers) - ? runtime._cleanupControllers - : []; - for (const controller of controllers) { - try { - controller?.abort?.(); - } catch (e) { - console.debug?.(e); - } - } - runtime._cleanupControllers = []; - } catch (e) { - console.warn("[MJR teardown]", e); - } - try { - removeApiHandlers(runtime.api || null); - removeRuntimeWindowHandlers(runtime); - } catch (e) { - console.warn("[MJR teardown]", e); - } -} - -function _registerCleanableListener(runtime, target, event, handler, options = undefined) { - if (!runtime || !target?.addEventListener || typeof handler !== "function") return null; - const listenerOptions = options && typeof options === "object" ? { ...options } : options; - if (typeof AbortController !== "undefined") { - try { - const controller = new AbortController(); - const nextOptions = - listenerOptions && typeof listenerOptions === "object" - ? { ...listenerOptions, signal: controller.signal } - : { signal: controller.signal }; - target.addEventListener(event, handler, nextOptions); - runtime._cleanupControllers = Array.isArray(runtime._cleanupControllers) - ? runtime._cleanupControllers - : []; - runtime._cleanupControllers.push(controller); - return controller; - } catch (e) { - console.debug?.(e); - } - } - target.addEventListener(event, handler, listenerOptions); - runtime._listenerCleanupFns = Array.isArray(runtime._listenerCleanupFns) - ? runtime._listenerCleanupFns - : []; - runtime._listenerCleanupFns.push(() => { - try { - target.removeEventListener(event, handler, listenerOptions); - } catch (e) { - console.debug?.(e); - } - }); - return null; -} - -function installEntryRuntimeController() { - try { - if (typeof window !== "undefined") { - // Replace previous runtime entry to allow cleanup on hot-reload. - // Handlers are removed explicitly via removeApiHandlers / removeRuntimeWindowHandlers. - try { - const prev = window[ENTRY_RUNTIME_KEY]; - cleanupEntryRuntime(prev); - } catch (e) { - console.warn("[MJR teardown]", e); - } - // Tear down MFV module-level listeners before re-registering (NM-3, NM-4). - try { - _liveStreamMod?.teardownLiveStreamTracker(window.__MJR_RUNTIME_APP__); - } catch (e) { - console.warn("[MJR teardown]", e); - } - try { - _nodeStreamMod?.teardownNodeStream(window.__MJR_RUNTIME_APP__); - } catch (e) { - console.warn("[MJR teardown]", e); - } - try { - teardownFloatingViewerManager(); - } catch (e) { - console.warn("[MJR teardown]", e); - } - window[ENTRY_RUNTIME_KEY] = { - api: null, - assetsDeletedHandler: null, - _cleanupControllers: [], - _listenerCleanupFns: [], - }; - } - } catch (e) { - console.warn("[MJR teardown]", e); - } -} - -function _safeCssEscape(value) { - const str = String(value ?? ""); - try { - if (typeof CSS !== "undefined" && typeof CSS.escape === "function") { - return CSS.escape(str); - } - } catch (e) { - console.debug?.(e); - } - return str.replace(/([!"#$%&'()*+,./:;<=>?@[\\\]^`{|}~])/g, "\\$1"); -} - -// Deduplication for executed events (ComfyUI can fire multiple times for same file) -// Window (ms) to ignore duplicate file events from the same execution -const DEDUPE_TTL_MS = 2000; -const RECENT_FILES_MAX = 2000; // Hard cap to prevent unbounded growth -const recentFiles = createTTLCache({ ttlMs: DEDUPE_TTL_MS, maxSize: RECENT_FILES_MAX }); -const INDEX_RETRYABLE_CODES = new Set([ - "DB_MAINTENANCE", - "TIMEOUT", - "NETWORK_ERROR", - "SERVICE_UNAVAILABLE", -]); -const INDEX_RETRY_MAX_ATTEMPTS = 8; -const INDEX_RETRY_BASE_DELAY_MS = 1200; -const INDEX_RETRY_MAX_DELAY_MS = 15_000; - -// Track execution start times to compute generation duration -const EXECUTION_START_TTL_MS = 10 * 60 * 1000; -const EXECUTION_STARTS_MAX = 500; // Hard cap to prevent unbounded growth -const executionStarts = createTTLCache({ - ttlMs: EXECUTION_START_TTL_MS, - maxSize: EXECUTION_STARTS_MAX, -}); -const executionAssetBuffer = createExecutionAssetBuffer(); -const liveAssetGate = createLiveAssetGate(); -let _lastStacksUpdateSignature = ""; -let _lastStacksUpdateAt = 0; - -function isJobTrackingDebugEnabled() { - try { - if (typeof window === "undefined") return false; - if (window.__MJR_DEBUG_JOB_TRACKING__ === true) return true; - return String(window.localStorage?.getItem("mjr.debug.jobTracking") || "").trim() === "1"; - } catch (e) { - console.debug?.(e); - return false; - } -} - -function debugJobTracking(stage, detail = {}) { - if (!isJobTrackingDebugEnabled()) return; - try { - console.debug("[Majoor][job-tracking]", stage, detail); - } catch (e) { - console.debug?.(e); - } -} - -const _stackFinalizeQueue = createJobFinalizeQueue({ - defaultDelayMs: 900, - async postJob(targetJobId) { - debugJobTracking("auto-stack:start", { job_id: targetJobId }); - try { - const res = await post(ENDPOINTS.STACKS_AUTO_STACK, { - mode: "job_id", - job_id: targetJobId, - }); - if (!res?.ok) { - liveAssetGate.clearPendingJob(targetJobId); - reportError( - new Error(String(res?.error || "Auto stack failed")), - "entry.execution_end.auto_stack", - ); - return; - } - debugJobTracking("auto-stack:done", { - job_id: targetJobId, - stacks: Array.isArray(res?.data?.stacks) ? res.data.stacks.length : 0, - }); - notifyStacksUpdated(res?.data || { targeted_job_id: targetJobId }); - } catch (error) { - reportError(error, "entry.execution_end.auto_stack"); - } - }, -}); - -function notifyStacksUpdated(detail = {}) { - debugJobTracking("stacks-updated", { - targeted_job_id: String(detail?.targeted_job_id || ""), - stacks: Array.isArray(detail?.stacks) - ? detail.stacks.map((item) => ({ - job_id: String(item?.job_id || ""), - stack_id: String(item?.stack_id || ""), - asset_count: Number(item?.asset_count || 0) || 0, - })) - : [], - }); - try { - const targeted = String(detail?.targeted_job_id || "").trim(); - 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) liveAssetGate.clearPendingJob(jobId); - } - } catch (e) { - console.debug?.(e); - } - try { - const stacks = Array.isArray(detail?.stacks) ? detail.stacks : []; - const signature = JSON.stringify({ - targeted_job_id: String(detail?.targeted_job_id || ""), - stacks: stacks.map((item) => ({ - job_id: String(item?.job_id || ""), - stack_id: String(item?.stack_id || ""), - asset_count: Number(item?.asset_count || 0) || 0, - })), - }); - const now = Date.now(); - if ( - signature && - signature === _lastStacksUpdateSignature && - now - _lastStacksUpdateAt < 1500 - ) { - return; - } - _lastStacksUpdateSignature = signature; - _lastStacksUpdateAt = now; - } catch (e) { - console.debug?.(e); - } - try { - window.dispatchEvent( - new CustomEvent("mjr:stacks-updated", { detail: { ...(detail || {}) } }), - ); - } catch (e) { - console.debug?.(e); - } - try { - refreshGeneratedFeedHosts(); - } catch (e) { - console.debug?.(e); - } - try { - const grid = getActiveGridContainer(); - const scope = String(grid?.dataset?.mjrScope || "") - .trim() - .toLowerCase(); - if (grid && (scope === "output" || scope === "all")) { - window.dispatchEvent( - new CustomEvent(EVENTS.RELOAD_GRID, { - detail: { reason: "stacks-updated", ...(detail || {}) }, - }), - ); - } - } catch (e) { - console.debug?.(e); - } -} - -function rememberExecutionStart(promptId, timestampMs) { - if (!promptId) return; - const ts = Number(timestampMs) || Date.now(); - executionStarts.set(String(promptId), ts, { at: ts }); - pruneExecutionStarts(); -} - -function pruneExecutionStarts() { - executionStarts.prune(); -} - -function dedupeFiles(files) { - recentFiles.prune(); - // Filter out recently seen files - const fresh = []; - for (const f of files) { - // toLowerCase for case-insensitive path dedup (Windows paths may differ in casing) - const key = `${f.type || ""}|${f.subfolder || ""}|${f.filename || ""}`.toLowerCase(); - if (!recentFiles.has(key)) { - recentFiles.set(key, true); - fresh.push(f); - } - } - return fresh; -} - -function _indexRetryDelayMs(attempt) { - const a = Math.max(1, Number(attempt) || 1); - const exp = Math.min(6, a - 1); - return Math.min(INDEX_RETRY_MAX_DELAY_MS, INDEX_RETRY_BASE_DELAY_MS * 2 ** exp); -} - -function _shouldRetryIndexResponse(res) { - const code = String(res?.code || "") - .trim() - .toUpperCase(); - return INDEX_RETRYABLE_CODES.has(code); -} - -function scheduleGenerationIndex(files, attempt = 1, meta = {}) { - const payloadFiles = Array.isArray(files) ? files : []; - if (!payloadFiles.length) return; - const payload = { - files: payloadFiles, - origin: "generation", - }; - if (meta?.prompt_id || meta?.promptId) { - payload.prompt_id = String(meta.prompt_id || meta.promptId); - } - debugJobTracking("index:direct", { - prompt_id: String(payload.prompt_id || ""), - attempt, - files: payloadFiles.map((item) => String(item?.filename || item?.filepath || "")), - }); - post(ENDPOINTS.INDEX_FILES, payload) - .then((res) => { - if (res?.ok) return; - if (_shouldRetryIndexResponse(res) && attempt < INDEX_RETRY_MAX_ATTEMPTS) { - const delayMs = _indexRetryDelayMs(attempt); - setTimeout(() => scheduleGenerationIndex(payloadFiles, attempt + 1, meta), delayMs); - return; - } - try { - const code = String(res?.code || "INDEX_FAILED"); - const msg = String(res?.error || "Indexing generated files failed"); - reportError(new Error(`${code}: ${msg}`), "entry.executed.index"); - } catch (e) { - console.debug?.(e); + console.warn("[MJR setup] initLiveStreamTracker failed:", e); } }) - .catch((error) => reportError(error, "entry.executed.index")); -} + .catch((e) => console.warn("[MJR setup] LiveStreamTracker load failed:", e)); -function resolveGenerationPromptId(meta = {}) { - const explicitPromptId = String(meta?.prompt_id || meta?.promptId || "").trim(); - if (explicitPromptId) return explicitPromptId; - try { - return String(ensureExecutionRuntime()?.active_prompt_id || "").trim(); - } catch (e) { - console.debug?.(e); - return ""; - } -} - -function bufferGenerationIndex(files, meta = {}) { - const promptId = resolveGenerationPromptId(meta); - debugJobTracking("buffer:add", { - prompt_id: promptId, - explicit_prompt_id: String(meta?.prompt_id || meta?.promptId || ""), - file_count: Array.isArray(files) ? files.length : 0, - files: Array.isArray(files) - ? files.map((item) => String(item?.filename || item?.filepath || "")) - : [], - }); - if (!promptId) { - scheduleGenerationIndex(files, 1, meta); + if (!NODE_STREAM_FEATURE_ENABLED) { + console.debug( + `[Majoor] Node Stream disabled by feature flag. See ${NODE_STREAM_REACTIVATION_DOC}`, + ); return; } - executionAssetBuffer.add(promptId, files); -} -function flushBufferedGenerationIndex(jobId) { - const promptId = String(jobId || "").trim(); - if (!promptId) return; - const files = executionAssetBuffer.take(promptId); - debugJobTracking("buffer:flush", { - prompt_id: promptId, - file_count: files.length, - files: files.map((item) => String(item?.filename || item?.filepath || "")), - }); - liveAssetGate.markPendingJob(promptId, { files, expectedCount: files.length }); - if (!files.length) { - scheduleFinalizeExecutionStacks(promptId); - return; - } - post(ENDPOINTS.INDEX_FILES, { - files, - origin: "generation", - prompt_id: promptId, - }) - .then((res) => { - if (!res?.ok) { - liveAssetGate.clearPendingJob(promptId); - reportError( - new Error(String(res?.error || "Indexing generated files failed")), - "entry.executed.index", - ); - return; + void import("./features/viewer/nodeStream/NodeStreamController.js") + .then((mod) => { + nodeStreamModule = mod; + try { + mod.initNodeStream({ + app: runtimeApp, + onOutput: (fileData) => floatingViewerManager.feedNodeStream(fileData), + }); + } catch (e) { + console.warn("[MJR setup] initNodeStream failed:", e); } - debugJobTracking("index:flush:done", { - prompt_id: promptId, - file_count: files.length, - }); - scheduleFinalizeExecutionStacks(promptId, 200); }) - .catch((error) => { - liveAssetGate.clearPendingJob(promptId); - reportError(error, "entry.executed.index"); - }); + .catch((e) => console.warn("[MJR setup] NodeStream load failed:", e)); } -function prepareLiveAssetEvent(detail) { +async function setupApiListeners(runtimeApp, executionRuntime) { + const api = + (await waitForComfyApi({ app: runtimeApp, timeoutMs: 4000 })) || getComfyApi(runtimeApp); + setComfyApi(api || null); + if (!api) { + console.warn("Majoor API not available, real-time updates disabled"); + return; + } + + let runtime = null; try { - return liveAssetGate.prepare(detail); + runtime = typeof window !== "undefined" ? window[ENTRY_RUNTIME_KEY] || null : null; } catch (e) { console.debug?.(e); - return { detail, jobId: "", defer: false }; } + + removeApiHandlers(runtime?.api || null, reportError); + if (api !== runtime?.api) removeApiHandlers(api, reportError); + removeRuntimeWindowHandlers(runtime, reportError); + + await registerRealtimeListeners({ + api, + runtime, + executionRuntime, + appRef: app, + liveStreamModule, + ensureExecutionRuntime, + emitRuntimeStatus, + getActiveGridContainer, + pushGeneratedAsset, + upsertAsset, + removeAssetsFromGrid, + getEnrichmentState, + setEnrichmentState, + comfyToast, + t, + triggerStartupScan, + reportError, + registerCleanableListener, + }); } -function scheduleFinalizeExecutionStacks(jobId, delayMs = 900) { - debugJobTracking("finalize:schedule", { job_id: String(jobId || ""), delay_ms: delayMs }); - _stackFinalizeQueue.schedule(jobId, delayMs); -} +const executionRuntime = createExecutionRuntimeController({ + post, + ENDPOINTS, + reportError, + extractOutputFiles, + ensureExecutionRuntime, + emitRuntimeStatus, + refreshGeneratedFeedHosts, + getActiveGridContainer, +}); app.registerExtension({ name: "Majoor.AssetsManager", settings: buildMajoorSettings(app), - commands: [ - { - id: "mjr.openAssetsManager", - label: "Open Assets Manager", - icon: "pi pi-folder-open", - function: () => openAssetsManagerPanel(app), - }, - { - id: "mjr.scanAssets", - label: "Scan Assets", - icon: "pi pi-refresh", - function: () => triggerStartupScan(), - }, - { - id: "mjr.toggleFloatingViewer", - label: "Toggle Floating Viewer", - icon: "pi pi-images", - function: () => window.dispatchEvent(new Event(EVENTS.MFV_TOGGLE)), - }, - { - id: "mjr.refreshAssetsGrid", - label: "Refresh Assets Grid", - icon: "pi pi-sync", - function: () => triggerRefreshGrid(), - }, - ], + commands: buildNativeCommands(app, { + sidebarTabId: SIDEBAR_TAB_ID, + triggerStartupScan, + }), async setup() { - installEntryRuntimeController(); - // Seed the bridge with the module-imported app so waitForComfyApp - // resolves immediately (newer ComfyUI may not expose app on window/globalThis). + installEntryRuntimeController({ + cleanupEntryRuntimeFn: cleanupEntryRuntime, + teardownLiveStreamTracker: (runtimeApp) => + liveStreamModule?.teardownLiveStreamTracker(runtimeApp), + teardownNodeStream: (runtimeApp) => nodeStreamModule?.teardownNodeStream(runtimeApp), + teardownFloatingViewerManager, + reportError, + }); + setComfyApp(app); const runtimeApp = (await waitForComfyApp({ timeoutMs: 12000 })) || app; setComfyApp(runtimeApp); - // Store app reference for teardown use in subsequent hot-reloads. try { if (typeof window !== "undefined") window.__MJR_RUNTIME_APP__ = runtimeApp; } catch (e) { console.debug?.(e); } - // Initialize core services testAPI(); ensureStyleLoaded({ enabled: UI_FLAGS.useComfyThemeUI }); @@ -722,731 +231,76 @@ app.registerExtension({ console.warn("[MJR setup] initDragDrop failed:", e); } - // Lazy-load LiveStreamTracker in the background (not needed for initial render). - void import("./features/viewer/LiveStreamTracker.js") - .then((mod) => { - _liveStreamMod = mod; - try { - mod.initLiveStreamTracker(runtimeApp); - } catch (e) { - console.warn("[MJR setup] initLiveStreamTracker failed:", e); - } - }) - .catch((e) => console.warn("[MJR setup] LiveStreamTracker load failed:", e)); - - if (NODE_STREAM_FEATURE_ENABLED) { - // Lazy-load NodeStream in the background (feature-flagged, not needed at startup). - void import("./features/viewer/nodeStream/NodeStreamController.js") - .then((mod) => { - _nodeStreamMod = mod; - try { - mod.initNodeStream({ - app: runtimeApp, - onOutput: (fileData) => floatingViewerManager.feedNodeStream(fileData), - }); - } catch (e) { - console.warn("[MJR setup] initNodeStream failed:", e); - } - }) - .catch((e) => console.warn("[MJR setup] NodeStream load failed:", e)); - } else { - console.debug( - `[Majoor] Node Stream disabled by feature flag. See ${NODE_STREAM_REACTIVATION_DOC}`, - ); - } + setupLazyModules(runtimeApp); registerMajoorSettings(runtimeApp, () => { const grid = getActiveGridContainer(); if (grid) loadAssets(grid); }); - registerNativeCommands(runtimeApp); + registerNativeCommands(runtimeApp, { + sidebarTabId: SIDEBAR_TAB_ID, + triggerStartupScan, + }); setTimeout(() => { void checkMajoorVersion(); }, 5000); - // Initialize API listeners in the background so sidebar tab registration - // is not blocked by API readiness polling. - void (async () => { - // Get ComfyUI API – errors are caught below to avoid unhandled rejections. - const api = - (await waitForComfyApi({ app: runtimeApp, timeoutMs: 4000 })) || - getComfyApi(runtimeApp); - setComfyApi(api || null); - if (api) { - let runtime = null; - try { - runtime = - typeof window !== "undefined" ? window[ENTRY_RUNTIME_KEY] || null : null; - } catch (e) { - console.debug?.(e); - } - removeApiHandlers(runtime?.api || null); - // Only clean up stale handlers on the new api if it is a different object - // than the previous runtime's api (avoids redundant double-removal). - if (api !== runtime?.api) removeApiHandlers(api); - removeRuntimeWindowHandlers(runtime); + void setupApiListeners(runtimeApp, executionRuntime).catch((e) => + reportError(e, "entry.api_setup"), + ); - // Listen for ComfyUI execution - extract output files and send to backend - api._mjrExecutedHandler = (event) => { - try { - const outputFiles = dedupeFiles(extractOutputFiles(event?.detail?.output)); - if (!outputFiles.length) return; - - const promptId = event?.detail?.prompt_id || event?.detail?.promptId; - const sourceNodeId = String( - event?.detail?.node || event?.detail?.display_node || "", - ).trim(); - const startTs = promptId ? executionStarts.get(String(promptId)) : null; - const genTimeMs = startTs ? Math.max(0, Date.now() - startTs) : 0; - - // Resolve node class_type from the active workflow graph - let sourceNodeType = ""; - try { - if (sourceNodeId && typeof app !== "undefined" && app?.graph) { - const graphNode = app.graph.getNodeById(Number(sourceNodeId)); - if (graphNode) - sourceNodeType = String( - graphNode.type || graphNode.comfyClass || "", - ).trim(); - } - } catch (e) { - console.debug?.(e); - } - - const files = outputFiles.map((f) => ({ - ...f, - ...(genTimeMs > 0 ? { generation_time_ms: genTimeMs } : {}), - ...(sourceNodeId ? { source_node_id: sourceNodeId } : {}), - ...(sourceNodeType ? { source_node_type: sourceNodeType } : {}), - })); - debugJobTracking("executed:event", { - prompt_id: String(promptId || ""), - source_node_id: sourceNodeId, - source_node_type: sourceNodeType, - file_count: files.length, - files: files.map((item) => String(item?.filename || item?.filepath || "")), - }); - - // Notify the MFV (and any other subscriber) that new files were generated. - try { - window.dispatchEvent( - new CustomEvent(EVENTS.NEW_GENERATION_OUTPUT, { - detail: { files }, - }), - ); - } catch (e) { - console.debug?.(e); - } - - bufferGenerationIndex(files, { prompt_id: promptId }); - } catch (error) { - reportError(error, "entry.executed"); - } - }; - _registerCleanableListener(runtime, api, "executed", api._mjrExecutedHandler); - - // Track execution start/end to compute duration - api._mjrExecutionStartHandler = (event) => { - try { - const promptId = event?.detail?.prompt_id || event?.detail?.promptId; - const ts = event?.detail?.timestamp; - debugJobTracking("execution:start", { - prompt_id: String(promptId || ""), - timestamp: Number(ts) || null, - }); - rememberExecutionStart(promptId, ts); - // Feed jobId to LiveStreamTracker so b_preview_with_metadata - // events from stale executions are filtered out (ComfyUI v1.42+). - _liveStreamMod?.setCurrentJobId(promptId || null); - emitRuntimeStatus({ - active_prompt_id: promptId || null, - cached_nodes: [], - progress_node: null, - progress_value: null, - progress_max: null, - }); - } catch (error) { - reportError(error, "entry.execution_start"); - } - }; - api._mjrExecutionEndHandler = (event) => { - try { - const promptId = event?.detail?.prompt_id || event?.detail?.promptId; - debugJobTracking("execution:end", { - type: String(event?.type || ""), - prompt_id: String(promptId || ""), - }); - if (promptId) { - executionStarts.delete(String(promptId)); - } - pruneExecutionStarts(); - // Clear jobId filter once execution completes. - _liveStreamMod?.setCurrentJobId(null); - emitRuntimeStatus({ - active_prompt_id: null, - cached_nodes: [], - progress_node: null, - progress_value: null, - progress_max: null, - }); - if (promptId && String(event?.type || "") !== "execution_success") { - debugJobTracking("buffer:clear", { - prompt_id: String(promptId || ""), - reason: String(event?.type || ""), - }); - executionAssetBuffer.clear(promptId); - liveAssetGate.clearPendingJob(promptId); - } - if (String(event?.type || "") === "execution_success") { - flushBufferedGenerationIndex(promptId); - } - } catch (error) { - reportError(error, "entry.execution_end"); - } - }; - api._mjrStacksUpdatedHandler = (event) => { - try { - notifyStacksUpdated(event?.detail || {}); - } catch (error) { - reportError(error, "entry.stacks_updated"); - } - }; - api._mjrRuntimeStatusHandler = (event) => { - try { - const detail = event?.detail || {}; - emitRuntimeStatus({ - progress_node: detail?.node ?? null, - progress_value: Number.isFinite(Number(detail?.value)) - ? Number(detail.value) - : null, - progress_max: Number.isFinite(Number(detail?.max)) - ? Number(detail.max) - : null, - queue_remaining: Number.isFinite( - Number(detail?.exec_info?.queue_remaining), - ) - ? Number(detail.exec_info.queue_remaining) - : Number.isFinite(Number(detail?.queue_remaining)) - ? Number(detail.queue_remaining) - : ensureExecutionRuntime().queue_remaining, - }); - } catch (error) { - reportError(error, "entry.runtime_status"); - } - }; - api._mjrExecutionCachedHandler = (event) => { - try { - const detail = event?.detail || {}; - emitRuntimeStatus({ - active_prompt_id: - detail?.prompt_id || - detail?.promptId || - ensureExecutionRuntime().active_prompt_id || - null, - cached_nodes: Array.isArray(detail?.nodes) ? detail.nodes.slice() : [], - }); - } catch (error) { - reportError(error, "entry.execution_cached"); - } - }; - _registerCleanableListener( - runtime, - api, - "execution_start", - api._mjrExecutionStartHandler, - ); - _registerCleanableListener( - runtime, - api, - "execution_success", - api._mjrExecutionEndHandler, - ); - _registerCleanableListener( - runtime, - api, - "execution_error", - api._mjrExecutionEndHandler, - ); - _registerCleanableListener( - runtime, - api, - "execution_interrupted", - api._mjrExecutionEndHandler, - ); - _registerCleanableListener( - runtime, - api, - "mjr.stacks.updated", - api._mjrStacksUpdatedHandler, - ); - _registerCleanableListener(runtime, api, "progress", api._mjrRuntimeStatusHandler); - _registerCleanableListener(runtime, api, "status", api._mjrRuntimeStatusHandler); - _registerCleanableListener( - runtime, - api, - EVENTS.RUNTIME_STATUS, - api._mjrRuntimeStatusHandler, - ); - _registerCleanableListener( - runtime, - api, - "execution_cached", - api._mjrExecutionCachedHandler, - ); - - // Listen for backend push - upsert asset directly to grid - api._mjrAssetAddedHandler = (event) => { - try { - const grid = getActiveGridContainer(); - if (grid && 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, detail); - if (handled) { - // Signal that we handled this event so handleCountersUpdate - // skips its own reload within the next window. - try { - window.__mjrLastAssetUpsert = Date.now(); - } catch (e) { - console.debug?.(e); - } - } - } - // No fallback reload here — let handleCountersUpdate's - // queuedReload() handle it (preserves scroll anchor). - } - } catch (error) { - reportError(error, "entry.asset_added"); - } - }; - _registerCleanableListener( - runtime, - api, - "mjr-asset-added", - api._mjrAssetAddedHandler, - ); - - api._mjrAssetUpdatedHandler = (event) => { - try { - const grid = getActiveGridContainer(); - if (grid && 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(); - const filepath = String(detail?.filepath || "").trim(); - const hasRenderableFields = !!kind && (!!filename || !!filepath); - let existsInGrid = false; - if (assetId) { - try { - const escapedId = _safeCssEscape(assetId); - existsInGrid = !!grid.querySelector( - `.mjr-asset-card[data-mjr-asset-id="${escapedId}"]`, - ); - } catch (e) { - console.debug?.(e); - } - } - - const canUpsert = hasRenderableFields || existsInGrid; - const handled = canUpsert ? !!upsertAsset(grid, detail) : false; - if (handled) { - try { - window.__mjrLastAssetUpsert = Date.now(); - } catch (e) { - console.debug?.(e); - } - return; - } - - // Non-renderable partial updates for unseen assets — let - // handleCountersUpdate's queuedReload() pick up changes - // (preserves scroll anchor and avoids reload storms). - } - } catch (error) { - reportError(error, "entry.asset_updated"); - } - }; - _registerCleanableListener( - runtime, - api, - "mjr-asset-updated", - api._mjrAssetUpdatedHandler, - ); - - // Listen for scan-complete events to avoid "Unknown message type" warnings - api._mjrScanCompleteHandler = (event) => { - try { - const detail = event?.detail || {}; - window.dispatchEvent(new CustomEvent(EVENTS.SCAN_COMPLETE, { detail })); - } catch (e) { - console.debug?.(e); - } - }; - _registerCleanableListener( - runtime, - api, - EVENTS.SCAN_COMPLETE, - api._mjrScanCompleteHandler, - ); - - api._mjrScanProgressHandler = (event) => { - try { - const detail = event?.detail || {}; - window.dispatchEvent(new CustomEvent(EVENTS.SCAN_PROGRESS, { detail })); - } catch (e) { - console.debug?.(e); - } - }; - _registerCleanableListener( - runtime, - api, - EVENTS.SCAN_PROGRESS, - api._mjrScanProgressHandler, - ); - - api._mjrAssetIndexingHandler = (event) => { - try { - const detail = event?.detail || {}; - window.dispatchEvent(new CustomEvent(EVENTS.ASSET_INDEXING, { detail })); - } catch (e) { - console.debug?.(e); - } - }; - _registerCleanableListener( - runtime, - api, - EVENTS.ASSET_INDEXING, - api._mjrAssetIndexingHandler, - ); - - api._mjrAssetIndexedHandler = (event) => { - try { - const detail = event?.detail || {}; - window.dispatchEvent(new CustomEvent(EVENTS.ASSET_INDEXED, { detail })); - } catch (e) { - console.debug?.(e); - } - }; - _registerCleanableListener( - runtime, - api, - EVENTS.ASSET_INDEXED, - api._mjrAssetIndexedHandler, - ); - - api._mjrEnrichmentStatusHandler = (event) => { - try { - const detail = event?.detail || {}; - const queued = Number(detail?.queued); - const queueLeft = Number(detail?.queue_left); - const queueLen = Number.isFinite(queued) - ? Math.max(0, Math.floor(queued)) - : Number.isFinite(queueLeft) - ? Math.max(0, Math.floor(queueLeft)) - : 0; - const active = !!detail?.active || queueLen > 0; - const prev = getEnrichmentState().active; - setEnrichmentState(active, queueLen); - window.dispatchEvent(new CustomEvent(EVENTS.ENRICHMENT_STATUS, { detail })); - // Do not force a full grid reset on enrichment state flips. - // Panel-level listeners handle refresh with scroll/selection anchor preservation. - if (prev && !active) { - comfyToast( - t("toast.enrichmentComplete", "Metadata enrichment complete"), - "success", - 2600, - ); - } - } catch (e) { - console.debug?.(e); - } - }; - _registerCleanableListener( - runtime, - api, - EVENTS.ENRICHMENT_STATUS, - api._mjrEnrichmentStatusHandler, - ); - - api._mjrDbRestoreStatusHandler = (event) => { - try { - const detail = event?.detail || {}; - const step = String(detail?.step || ""); - const level = String(detail?.level || "info"); - const op = String(detail?.operation || ""); - window.dispatchEvent(new CustomEvent(EVENTS.DB_RESTORE_STATUS, { detail })); - const isDelete = op === "delete_db"; - const isReset = op === "reset_index"; - if (isDelete && !["started", "done", "failed"].includes(step)) { - return; - } - const map = { - started: isDelete - ? t( - "toast.dbDeleteTriggered", - "Deleting database and rebuilding...", - ) - : isReset - ? t( - "toast.resetTriggered", - "Reset triggered: Reindexing all files...", - ) - : t("toast.dbRestoreStarted", "DB restore started"), - stopping_workers: t( - "toast.dbRestoreStopping", - "Stopping running workers", - ), - resetting_db: t( - "toast.dbRestoreResetting", - "Unlocking and resetting database", - ), - delete_db: t( - "toast.dbDeleteTriggered", - "Deleting database and rebuilding...", - ), - recreate_db: t("toast.dbRestoreReplacing", "Recreating database"), - replacing_files: t( - "toast.dbRestoreReplacing", - "Replacing database files", - ), - restarting_scan: t("toast.dbRestoreRescan", "Restarting scan"), - done: isDelete - ? t( - "toast.dbDeleteSuccess", - "Database deleted and rebuilt. Files are being reindexed.", - ) - : isReset - ? t( - "toast.resetStarted", - "Index reset started. Files will be reindexed in the background.", - ) - : t("toast.dbRestoreSuccess", "Database backup restored"), - failed: String( - detail?.message || - (isDelete - ? t("toast.dbDeleteFailed", "Failed to delete database") - : isReset - ? t("toast.resetFailed", "Failed to reset index") - : t( - "toast.dbRestoreFailed", - "Failed to restore DB backup", - )), - ), - }; - const msg = map[step] || String(detail?.message || ""); - if (!msg) return; - const tone = - level === "error" || step === "failed" - ? "error" - : step === "done" - ? "success" - : "info"; - comfyToast(msg, tone, step === "done" || step === "failed" ? 2800 : 1800); - } catch (e) { - console.debug?.(e); - } - }; - _registerCleanableListener( - runtime, - api, - EVENTS.DB_RESTORE_STATUS, - api._mjrDbRestoreStatusHandler, - ); - - const assetsDeletedHandler = (event) => { - try { - const ids = Array.isArray(event?.detail?.ids) - ? event.detail.ids.map((x) => String(x || "")).filter(Boolean) - : []; - if (!ids.length) return; - const grid = getActiveGridContainer(); - if (grid) removeAssetsFromGrid(grid, ids); - } catch (e) { - console.debug?.(e); - } - }; - _registerCleanableListener( - runtime, - window, - EVENTS.ASSETS_DELETED, - assetsDeletedHandler, - ); - try { - if (runtime && typeof runtime === "object") { - runtime.api = api; - runtime.assetsDeletedHandler = assetsDeletedHandler; - } - } catch (error) { - reportError(error, "entry.runtime_store"); - } - triggerStartupScan(); - console.debug("[Majoor] Real-time listener registered"); - } else { - console.warn("📂 Majoor [⚠️] API not available, real-time updates disabled"); - } - })().catch((e) => reportError(e, "entry.api_setup")); - - // Register sidebar tab if ( - registerSidebarTabCompat(runtimeApp, { - id: SIDEBAR_TAB_ID, - icon: "pi pi-folder", - title: t("manager.title"), - label: t("manager.sidebarLabel"), - tooltip: t("tooltip.sidebarTab"), - type: "custom", - render: (el) => { - void renderAssetsManager(el, { useComfyThemeUI: UI_FLAGS.useComfyThemeUI }); - }, - destroy: (el) => { - try { - el?._eventCleanup?.(); - } catch (e) { - console.debug?.(e); - } - try { - el?._mjrPanelState?._mjrDispose?.(); - } catch (e) { - console.debug?.(e); - } - try { - el?._mjrPopoverManager?.dispose?.(); - } catch (e) { - console.debug?.(e); - } - try { - el?.replaceChildren?.(); - } catch (e) { - console.debug?.(e); - } - }, + registerAssetsSidebar(runtimeApp, { + sidebarTabId: SIDEBAR_TAB_ID, + useComfyThemeUI: UI_FLAGS.useComfyThemeUI, }) ) { console.debug("[Majoor] Sidebar tab registered"); } else { console.warn( - "📂 Majoor Assets Manager: extensionManager.registerSidebarTab is unavailable", + "Majoor Assets Manager: extensionManager.registerSidebarTab is unavailable", ); } try { - registerBottomPanelTabCompat(runtimeApp, getGeneratedFeedBottomPanelTab()); + registerGeneratedBottomPanel(runtimeApp); } catch (e) { console.debug?.("[Majoor] Bottom panel tab registration unavailable", e); } - // Export metrics for debugging (accessible via console) - if (typeof window !== "undefined") { - window.MajoorDebug = { - exportMetrics: () => window.MajoorMetrics?.exportMetrics?.(), - getMetrics: () => window.MajoorMetrics?.getMetricsReport?.(), - resetMetrics: () => window.MajoorMetrics?.resetMetrics?.(), - }; - console.debug?.( - "[Majoor] Debug commands available: window.MajoorDebug.exportMetrics(), window.MajoorDebug.getMetrics(), window.MajoorDebug.resetMetrics()", - ); - if (NODE_STREAM_FEATURE_ENABLED) { - // Expose NodeStream API for third-party adapter registration (lazy-resolved). - const _nsApi = - (fn) => - async (...args) => { - if (!_nodeStreamMod) { - _nodeStreamMod = - await import("./features/viewer/nodeStream/NodeStreamController.js"); - } - return _nodeStreamMod[fn](...args); - }; - window.MajoorNodeStream = { - registerAdapter: _nsApi("registerAdapter"), - listAdapters: _nsApi("listAdapters"), - async createAdapter(config) { - const { createAdapter } = - await import("./features/viewer/nodeStream/adapters/BaseAdapter.js"); - return createAdapter(config); - }, - async getKnownNodeSets() { - const { getKnownNodeSets } = - await import("./features/viewer/nodeStream/adapters/KnownNodesAdapter.js"); - return getKnownNodeSets(); - }, - }; - console.debug?.( - "[Majoor] NodeStream API: window.MajoorNodeStream.registerAdapter(adapter), .createAdapter(config), .listAdapters()", - ); - } else { - try { - delete window.MajoorNodeStream; - } catch { - window.MajoorNodeStream = undefined; + // Warm the asset index quickly after startup so the first panel open feels instant, + // while still avoiding scans during active execution. + void triggerStartupScan({ delayMs: 200, idleOnly: true }).catch((e) => + reportError(e, "entry.startup_scan"), + ); + + exposeDebugApis({ + resolveNodeStreamModule: async () => { + if (!nodeStreamModule) { + nodeStreamModule = await import( + "./features/viewer/nodeStream/NodeStreamController.js" + ); } - } - } + return nodeStreamModule; + }, + }); }, /** * ComfyUI v1.42+ hook: fires when individual node outputs are updated during execution. - * @param {Map} nodeOutputs - Map of nodeId → outputs object + * @param {Map} nodeOutputs */ onNodeOutputsUpdated(nodeOutputs) { try { - const files = []; - for (const [, outputs] of nodeOutputs instanceof Map - ? nodeOutputs.entries() - : Object.entries(nodeOutputs || {})) { - for (const items of Object.values(outputs || {})) { - if (!Array.isArray(items)) continue; - for (const item of items) { - if (item?.filename && item?.type) files.push(item); - } - } - } - if (!files.length) return; - debugJobTracking("node-outputs:updated", { - prompt_id: resolveGenerationPromptId(), - file_count: files.length, - files: files.map((item) => String(item?.filename || item?.filepath || "")), - }); - bufferGenerationIndex(files, { - prompt_id: resolveGenerationPromptId(), - }); + executionRuntime.handleNodeOutputsUpdated(nodeOutputs); } catch (e) { console.debug?.("[Majoor] onNodeOutputsUpdated error", e); } }, getNodeMenuItems(node) { - if (!isMajoorTrackableNode(node)) return []; - return [ - nodeMenuEntry("View in Assets Manager", () => openAssetsManagerPanel(app)), - nodeMenuEntry("Open in Floating Viewer", () => { - try { - window.dispatchEvent(new Event(EVENTS.MFV_OPEN)); - } catch (e) { - console.debug?.(e); - } - }), - nodeMenuEntry("Index Output", () => { - try { - triggerRefreshGrid(); - comfyToast(t("toast.rescanningFile", "Rescanning file…"), "info", 1800); - } catch (e) { - console.debug?.(e); - } - }), - ]; + return getMajoorNodeMenuItems(node, app, { sidebarTabId: SIDEBAR_TAB_ID }); }, - bottomPanelTabs: [getGeneratedFeedBottomPanelTab()], + bottomPanelTabs: buildBottomPanelTabs(), }); diff --git a/js/features/bottomPanel/GeneratedFeedTab.js b/js/features/bottomPanel/GeneratedFeedTab.js index 0b6675fa..995f28f1 100644 --- a/js/features/bottomPanel/GeneratedFeedTab.js +++ b/js/features/bottomPanel/GeneratedFeedTab.js @@ -12,6 +12,7 @@ import { APP_CONFIG } from "../../app/config.js"; const FEED_FETCH_LIMIT = 240; const FEED_PAGE_SIZE = 120; +const FEED_PUSH_RENDER_DEBOUNCE_MS = 80; function _getRenderedCards(grid) { try { @@ -122,6 +123,9 @@ const FEED_STATE = { }; function _getAssetGroupKey(asset) { + if (!APP_CONFIG.EXECUTION_GROUPING_ENABLED) { + return _assetKey(asset); + } const stackId = String(asset?.stack_id || "").trim(); if (stackId) return `stack:${stackId}`; const jobId = String(asset?.job_id || "").trim(); @@ -179,6 +183,15 @@ function _storeHostAsset(host, asset) { return normalized; } +function _scheduleHostRender(host) { + if (!host?.grid || !host.loaded) return; + if (host._renderTimer) return; + host._renderTimer = setTimeout(() => { + host._renderTimer = null; + void _renderHostAssets(host); + }, FEED_PUSH_RENDER_DEBOUNCE_MS); +} + function _openGroupViewer(groupedAsset, startIndex = 0) { const assets = Array.isArray(groupedAsset?._mjrFeedGroupAssets) ? groupedAsset._mjrFeedGroupAssets.filter(Boolean) @@ -560,7 +573,7 @@ export function pushGeneratedAsset(asset) { _storeHostAsset(host, normalized); if (!host.loaded) continue; host.empty.style.display = "none"; - void _renderHostAssets(host); + _scheduleHostRender(host); } catch (e) { console.debug?.(e); } @@ -612,6 +625,14 @@ export function disposeGeneratedFeedTab(host) { } catch (e) { console.debug?.(e); } + try { + if (resolvedHost._renderTimer) { + clearTimeout(resolvedHost._renderTimer); + resolvedHost._renderTimer = null; + } + } catch (e) { + console.debug?.(e); + } try { disposeGrid(resolvedHost.grid); } catch (e) { diff --git a/js/features/grid/GridView.js b/js/features/grid/GridView.js index 208657ef..569ea63f 100644 --- a/js/features/grid/GridView.js +++ b/js/features/grid/GridView.js @@ -1,7 +1,9 @@ /** GridView facade preserving public exports. */ export { + buildGridSnapshotKey, createGridContainer, + hydrateGridFromSnapshot, loadAssets, prepareGridForScopeSwitch, loadAssetsFromList, diff --git a/js/features/grid/GridView_impl.js b/js/features/grid/GridView_impl.js index c843ef14..6e3839e3 100644 --- a/js/features/grid/GridView_impl.js +++ b/js/features/grid/GridView_impl.js @@ -112,6 +112,9 @@ const RT_HYDRATE_STATE = new WeakMap(); // gridContainer -> { queue, inflight, s const UPSERT_BATCH_DEBOUNCE_MS = 200; // Debounce window for batch flush const UPSERT_BATCH_MAX_SIZE = 50; // Max items before forced flush const UPSERT_BATCH_STATE = new WeakMap(); // gridContainer -> { pending: Map, timer: number|null } +const GRID_SNAPSHOT_CACHE = new Map(); // snapshotKey -> { assets, title, at } +const GRID_SNAPSHOT_CACHE_MAX = 8; +const GRID_SNAPSHOT_ASSET_LIMIT = 160; // Temporary debug logging for grid + infinite scroll. // Enable at runtime from the browser console: @@ -133,6 +136,81 @@ const gridDebug = (...args) => { } }; +function _normalizeSnapshotPart(value, fallback = "") { + try { + return String(value ?? fallback).trim(); + } catch { + return String(fallback); + } +} + +export function buildGridSnapshotKey(parts = {}) { + return JSON.stringify({ + scope: _normalizeSnapshotPart(parts.scope || "output", "output"), + query: _normalizeSnapshotPart(parts.query || "*", "*"), + customRootId: _normalizeSnapshotPart(parts.customRootId || ""), + subfolder: _normalizeSnapshotPart(parts.subfolder || ""), + collectionId: _normalizeSnapshotPart(parts.collectionId || ""), + viewScope: _normalizeSnapshotPart(parts.viewScope || ""), + }); +} + +function _snapshotKeyFromGrid(gridContainer, state) { + try { + return buildGridSnapshotKey({ + scope: gridContainer?.dataset?.mjrScope || state?.scope || "output", + query: gridContainer?.dataset?.mjrQuery || state?.query || "*", + customRootId: gridContainer?.dataset?.mjrCustomRootId || "", + subfolder: gridContainer?.dataset?.mjrSubfolder || "", + collectionId: state?.collectionId || "", + viewScope: state?.viewScope || "", + }); + } catch (e) { + console.debug?.(e); + return ""; + } +} + +function _rememberGridSnapshot(gridContainer, state, title = "") { + try { + const assets = Array.isArray(state?.assets) ? state.assets : []; + if (!assets.length) return; + const key = _snapshotKeyFromGrid(gridContainer, state); + if (!key) return; + GRID_SNAPSHOT_CACHE.delete(key); + GRID_SNAPSHOT_CACHE.set(key, { + assets: assets.slice(0, GRID_SNAPSHOT_ASSET_LIMIT).map((asset) => ({ ...asset })), + title: String(title || "").trim(), + at: Date.now(), + }); + while (GRID_SNAPSHOT_CACHE.size > GRID_SNAPSHOT_CACHE_MAX) { + const oldestKey = GRID_SNAPSHOT_CACHE.keys().next().value; + if (!oldestKey) break; + GRID_SNAPSHOT_CACHE.delete(oldestKey); + } + } catch (e) { + console.debug?.(e); + } +} + +export async function hydrateGridFromSnapshot(gridContainer, parts = {}, options = {}) { + try { + const key = buildGridSnapshotKey(parts); + const snapshot = GRID_SNAPSHOT_CACHE.get(key); + if (!snapshot || !Array.isArray(snapshot.assets) || !snapshot.assets.length) return false; + const state = getOrCreateState(gridContainer); + if (Array.isArray(state?.assets) && state.assets.length) return false; + await loadAssetsFromList(gridContainer, snapshot.assets, { + title: snapshot.title || options.title || "Cached", + reset: true, + }); + return true; + } catch (e) { + console.debug?.(e); + return false; + } +} + function _getRenderedCards(gridContainer) { if (!gridContainer) return []; try { @@ -1273,6 +1351,7 @@ export async function loadAssets(gridContainer, query = "*", options = {}) { setGridMessage(gridContainer, "No assets found", { clear: true }); } else { startInfiniteScroll(gridContainer, state); + _rememberGridSnapshot(gridContainer, state, state.query); } } return { ok: true, count: state.offset, total: state.total || 0 }; @@ -1458,6 +1537,8 @@ export async function loadAssetsFromList(gridContainer, assets, options = {}) { state.virtualGrid = null; } setGridMessage(gridContainer, "No assets found", { clear: true }); + } else { + _rememberGridSnapshot(gridContainer, state, title); } } diff --git a/js/features/panel/AssetsManagerPanel.js b/js/features/panel/AssetsManagerPanel.js index 5ea27a77..459dd3fb 100644 --- a/js/features/panel/AssetsManagerPanel.js +++ b/js/features/panel/AssetsManagerPanel.js @@ -11,6 +11,7 @@ import { } from "../status/StatusDot.js"; import { createGridContainer, + hydrateGridFromSnapshot, loadAssets, loadAssetsFromList, prepareGridForScopeSwitch, @@ -2063,6 +2064,25 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } { signal: panelLifecycleAC?.signal }, ); + try { + await hydrateGridFromSnapshot( + gridContainer, + { + scope: state.scope || "output", + query: String(normalizeQuery(state.searchQuery || "*") || "*"), + customRootId: state.customRootId || "", + subfolder: state.currentFolderRelativePath || "", + collectionId: state.collectionId || "", + viewScope: state.viewScope || "", + }, + { + title: state.collectionName || state.searchQuery || "Cached", + }, + ); + } catch (e) { + console.debug?.(e); + } + const initialLoadPromise = gridController.reloadGrid(); try { await refreshDuplicateAlerts(); diff --git a/js/features/panel/controllers/gridController.js b/js/features/panel/controllers/gridController.js index be7e08fb..5b3a57ae 100644 --- a/js/features/panel/controllers/gridController.js +++ b/js/features/panel/controllers/gridController.js @@ -2,6 +2,7 @@ import { vectorSearch, hybridSearch, vectorStats } from "../../../api/client.js" import { comfyToast } from "../../../app/toast.js"; import { t } from "../../../app/i18n.js"; import { loadMajoorSettings } from "../../../app/settings.js"; +import { APP_CONFIG } from "../../../app/config.js"; export function createGridController({ gridContainer, @@ -332,7 +333,10 @@ export function createGridController({ gridContainer.dataset.mjrFilterDateExact = state.dateExactFilter || ""; gridContainer.dataset.mjrSort = state.sort || "mtime_desc"; gridContainer.dataset.mjrGroupStacks = - state.scope === "output" || state.scope === "all" ? "1" : "0"; + APP_CONFIG.EXECUTION_GROUPING_ENABLED && + (state.scope === "output" || state.scope === "all") + ? "1" + : "0"; // Keep selection durable across re-renders by persisting it in the dataset // (GridView re-applies selection as cards are created). diff --git a/js/features/runtime/entryDebugApi.js b/js/features/runtime/entryDebugApi.js new file mode 100644 index 00000000..cccb3aac --- /dev/null +++ b/js/features/runtime/entryDebugApi.js @@ -0,0 +1,50 @@ +import { NODE_STREAM_FEATURE_ENABLED } from "../viewer/nodeStream/nodeStreamFeatureFlag.js"; + +export function exposeDebugApis({ resolveNodeStreamModule }) { + if (typeof window === "undefined") return; + + window.MajoorDebug = { + exportMetrics: () => window.MajoorMetrics?.exportMetrics?.(), + getMetrics: () => window.MajoorMetrics?.getMetricsReport?.(), + resetMetrics: () => window.MajoorMetrics?.resetMetrics?.(), + }; + console.debug?.( + "[Majoor] Debug commands available: window.MajoorDebug.exportMetrics(), window.MajoorDebug.getMetrics(), window.MajoorDebug.resetMetrics()", + ); + + if (!NODE_STREAM_FEATURE_ENABLED) { + try { + delete window.MajoorNodeStream; + } catch { + window.MajoorNodeStream = undefined; + } + return; + } + + const nodeStreamApi = + (fn) => + async (...args) => { + const mod = await resolveNodeStreamModule(); + return mod?.[fn]?.(...args); + }; + + window.MajoorNodeStream = { + registerAdapter: nodeStreamApi("registerAdapter"), + listAdapters: nodeStreamApi("listAdapters"), + async createAdapter(config) { + const { createAdapter } = await import( + "../viewer/nodeStream/adapters/BaseAdapter.js" + ); + return createAdapter(config); + }, + async getKnownNodeSets() { + const { getKnownNodeSets } = await import( + "../viewer/nodeStream/adapters/KnownNodesAdapter.js" + ); + return getKnownNodeSets(); + }, + }; + console.debug?.( + "[Majoor] NodeStream API: window.MajoorNodeStream.registerAdapter(adapter), .createAdapter(config), .listAdapters()", + ); +} diff --git a/js/features/runtime/entryRuntimeLifecycle.js b/js/features/runtime/entryRuntimeLifecycle.js new file mode 100644 index 00000000..56e8b9d8 --- /dev/null +++ b/js/features/runtime/entryRuntimeLifecycle.js @@ -0,0 +1,186 @@ +import { EVENTS } from "../../app/events.js"; + +export const ENTRY_RUNTIME_KEY = "__MJR_ENTRY_RUNTIME__"; + +export function removeApiHandlers(api, reportError) { + if (!api) return; + try { + if (api._mjrAssetUpdateReloadTimer) { + clearTimeout(api._mjrAssetUpdateReloadTimer); + api._mjrAssetUpdateReloadTimer = null; + } + if (api._mjrExecutedHandler) { + api.removeEventListener("executed", api._mjrExecutedHandler); + } + if (api._mjrAssetAddedHandler) { + api.removeEventListener("mjr-asset-added", api._mjrAssetAddedHandler); + } + if (api._mjrAssetUpdatedHandler) { + api.removeEventListener("mjr-asset-updated", api._mjrAssetUpdatedHandler); + } + if (api._mjrScanCompleteHandler) { + api.removeEventListener(EVENTS.SCAN_COMPLETE, api._mjrScanCompleteHandler); + } + if (api._mjrScanProgressHandler) { + api.removeEventListener(EVENTS.SCAN_PROGRESS, api._mjrScanProgressHandler); + } + if (api._mjrAssetIndexingHandler) { + api.removeEventListener(EVENTS.ASSET_INDEXING, api._mjrAssetIndexingHandler); + } + if (api._mjrAssetIndexedHandler) { + api.removeEventListener(EVENTS.ASSET_INDEXED, api._mjrAssetIndexedHandler); + } + if (api._mjrExecutionStartHandler) { + api.removeEventListener("execution_start", api._mjrExecutionStartHandler); + } + if (api._mjrExecutionEndHandler) { + api.removeEventListener("execution_success", api._mjrExecutionEndHandler); + api.removeEventListener("execution_error", api._mjrExecutionEndHandler); + api.removeEventListener("execution_interrupted", api._mjrExecutionEndHandler); + } + if (api._mjrStacksUpdatedHandler) { + api.removeEventListener("mjr.stacks.updated", api._mjrStacksUpdatedHandler); + } + if (api._mjrEnrichmentStatusHandler) { + api.removeEventListener(EVENTS.ENRICHMENT_STATUS, api._mjrEnrichmentStatusHandler); + } + if (api._mjrDbRestoreStatusHandler) { + api.removeEventListener(EVENTS.DB_RESTORE_STATUS, api._mjrDbRestoreStatusHandler); + } + if (api._mjrRuntimeStatusHandler) { + api.removeEventListener("progress", api._mjrRuntimeStatusHandler); + api.removeEventListener("status", api._mjrRuntimeStatusHandler); + api.removeEventListener(EVENTS.RUNTIME_STATUS, api._mjrRuntimeStatusHandler); + api.removeEventListener("execution_cached", api._mjrExecutionCachedHandler); + } + } catch (error) { + reportError?.(error, "entry.removeApiHandlers"); + } +} + +export function removeRuntimeWindowHandlers(runtime, reportError) { + try { + const handler = runtime?.assetsDeletedHandler; + if (handler && typeof window !== "undefined") { + window.removeEventListener(EVENTS.ASSETS_DELETED, handler); + } + } catch (error) { + reportError?.(error, "entry.removeRuntimeWindowHandlers"); + } +} + +export function cleanupEntryRuntime(runtime, { reportError } = {}) { + if (!runtime || typeof runtime !== "object") return; + try { + const cleanups = Array.isArray(runtime._listenerCleanupFns) + ? runtime._listenerCleanupFns + : []; + for (const cleanup of cleanups) { + try { + cleanup?.(); + } catch (e) { + console.debug?.(e); + } + } + runtime._listenerCleanupFns = []; + } catch (e) { + console.warn("[MJR teardown]", e); + } + try { + const controllers = Array.isArray(runtime._cleanupControllers) + ? runtime._cleanupControllers + : []; + for (const controller of controllers) { + try { + controller?.abort?.(); + } catch (e) { + console.debug?.(e); + } + } + runtime._cleanupControllers = []; + } catch (e) { + console.warn("[MJR teardown]", e); + } + try { + removeApiHandlers(runtime.api || null, reportError); + removeRuntimeWindowHandlers(runtime, reportError); + } catch (e) { + console.warn("[MJR teardown]", e); + } +} + +export function registerCleanableListener(runtime, target, event, handler, options = undefined) { + if (!runtime || !target?.addEventListener || typeof handler !== "function") return null; + const listenerOptions = options && typeof options === "object" ? { ...options } : options; + if (typeof AbortController !== "undefined") { + try { + const controller = new AbortController(); + const nextOptions = + listenerOptions && typeof listenerOptions === "object" + ? { ...listenerOptions, signal: controller.signal } + : { signal: controller.signal }; + target.addEventListener(event, handler, nextOptions); + runtime._cleanupControllers = Array.isArray(runtime._cleanupControllers) + ? runtime._cleanupControllers + : []; + runtime._cleanupControllers.push(controller); + return controller; + } catch (e) { + console.debug?.(e); + } + } + target.addEventListener(event, handler, listenerOptions); + runtime._listenerCleanupFns = Array.isArray(runtime._listenerCleanupFns) + ? runtime._listenerCleanupFns + : []; + runtime._listenerCleanupFns.push(() => { + try { + target.removeEventListener(event, handler, listenerOptions); + } catch (e) { + console.debug?.(e); + } + }); + return null; +} + +export function installEntryRuntimeController({ + cleanupEntryRuntimeFn = cleanupEntryRuntime, + teardownLiveStreamTracker, + teardownNodeStream, + teardownFloatingViewerManager, + reportError, +}) { + try { + if (typeof window !== "undefined") { + try { + const prev = window[ENTRY_RUNTIME_KEY]; + cleanupEntryRuntimeFn(prev, { reportError }); + } catch (e) { + console.warn("[MJR teardown]", e); + } + try { + teardownLiveStreamTracker?.(window.__MJR_RUNTIME_APP__); + } catch (e) { + console.warn("[MJR teardown]", e); + } + try { + teardownNodeStream?.(window.__MJR_RUNTIME_APP__); + } catch (e) { + console.warn("[MJR teardown]", e); + } + try { + teardownFloatingViewerManager?.(); + } catch (e) { + console.warn("[MJR teardown]", e); + } + window[ENTRY_RUNTIME_KEY] = { + api: null, + assetsDeletedHandler: null, + _cleanupControllers: [], + _listenerCleanupFns: [], + }; + } + } catch (e) { + console.warn("[MJR teardown]", e); + } +} diff --git a/js/features/runtime/entryUiRegistration.js b/js/features/runtime/entryUiRegistration.js new file mode 100644 index 00000000..4677bdfe --- /dev/null +++ b/js/features/runtime/entryUiRegistration.js @@ -0,0 +1,151 @@ +import { + activateSidebarTabCompat, + registerBottomPanelTabCompat, + registerCommandCompat, + registerSidebarTabCompat, +} from "../../app/comfyApiBridge.js"; +import { EVENTS } from "../../app/events.js"; +import { comfyToast } from "../../app/toast.js"; +import { t } from "../../app/i18n.js"; +import { renderAssetsManager } from "../panel/AssetsManagerPanel.js"; +import { getGeneratedFeedBottomPanelTab } from "../bottomPanel/GeneratedFeedTab.js"; + +export function openAssetsManagerPanel(runtimeApp, sidebarTabId) { + const opened = activateSidebarTabCompat(runtimeApp, sidebarTabId); + if (!opened) { + try { + window.dispatchEvent(new Event(EVENTS.OPEN_ASSETS_MANAGER)); + } catch (e) { + console.debug?.(e); + } + } + return opened; +} + +export function triggerRefreshGrid() { + try { + window.dispatchEvent( + new CustomEvent(EVENTS.RELOAD_GRID, { detail: { reason: "command" } }), + ); + } catch (e) { + console.debug?.(e); + } +} + +export function buildNativeCommands(runtimeApp, { sidebarTabId, triggerStartupScan }) { + return [ + { + id: "mjr.openAssetsManager", + label: t("manager.title", "Assets Manager"), + icon: "pi pi-folder-open", + function: () => openAssetsManagerPanel(runtimeApp, sidebarTabId), + }, + { + id: "mjr.scanAssets", + label: t("command.scanAssets", "Scan assets"), + icon: "pi pi-refresh", + function: () => triggerStartupScan(), + }, + { + id: "mjr.toggleFloatingViewer", + label: t("command.toggleFloatingViewer", "Toggle floating viewer"), + icon: "pi pi-images", + function: () => window.dispatchEvent(new Event(EVENTS.MFV_TOGGLE)), + }, + { + id: "mjr.refreshAssetsGrid", + label: t("command.refreshAssetsGrid", "Refresh assets grid"), + icon: "pi pi-sync", + function: () => triggerRefreshGrid(), + }, + ]; +} + +export function registerNativeCommands(runtimeApp, options) { + for (const command of buildNativeCommands(runtimeApp, options)) { + registerCommandCompat(runtimeApp, command); + } +} + +export function registerAssetsSidebar(runtimeApp, { sidebarTabId, useComfyThemeUI }) { + return registerSidebarTabCompat(runtimeApp, { + id: sidebarTabId, + icon: "pi pi-folder", + title: t("manager.title"), + label: t("manager.sidebarLabel"), + tooltip: t("tooltip.sidebarTab"), + type: "custom", + render: (el) => { + void renderAssetsManager(el, { useComfyThemeUI }); + }, + destroy: (el) => { + try { + el?._eventCleanup?.(); + } catch (e) { + console.debug?.(e); + } + try { + el?._mjrPanelState?._mjrDispose?.(); + } catch (e) { + console.debug?.(e); + } + try { + el?._mjrPopoverManager?.dispose?.(); + } catch (e) { + console.debug?.(e); + } + try { + el?.replaceChildren?.(); + } catch (e) { + console.debug?.(e); + } + }, + }); +} + +export function registerGeneratedBottomPanel(runtimeApp) { + return registerBottomPanelTabCompat(runtimeApp, getGeneratedFeedBottomPanelTab()); +} + +export function buildBottomPanelTabs() { + return [getGeneratedFeedBottomPanelTab()]; +} + +export function isMajoorTrackableNode(node) { + const comfyClass = String( + node?.comfyClass || node?.type || node?.constructor?.type || "", + ).trim(); + if (!comfyClass) return false; + return /save|load|preview/i.test(comfyClass); +} + +function nodeMenuEntry(label, callback) { + return { + content: label, + callback, + }; +} + +export function getMajoorNodeMenuItems(node, runtimeApp, { sidebarTabId }) { + if (!isMajoorTrackableNode(node)) return []; + return [ + nodeMenuEntry("View in Assets Manager", () => + openAssetsManagerPanel(runtimeApp, sidebarTabId), + ), + nodeMenuEntry("Open in Floating Viewer", () => { + try { + window.dispatchEvent(new Event(EVENTS.MFV_OPEN)); + } catch (e) { + console.debug?.(e); + } + }), + nodeMenuEntry("Index Output", () => { + try { + triggerRefreshGrid(); + comfyToast(t("toast.rescanningFile", "Rescanning file…"), "info", 1800); + } catch (e) { + console.debug?.(e); + } + }), + ]; +} diff --git a/js/features/runtime/registerRealtimeListeners.js b/js/features/runtime/registerRealtimeListeners.js new file mode 100644 index 00000000..6cc9d708 --- /dev/null +++ b/js/features/runtime/registerRealtimeListeners.js @@ -0,0 +1,305 @@ +import { EVENTS } from "../../app/events.js"; + +export async function registerRealtimeListeners({ + api, + runtime, + executionRuntime, + appRef, + liveStreamModule, + ensureExecutionRuntime, + emitRuntimeStatus, + getActiveGridContainer, + pushGeneratedAsset, + upsertAsset, + removeAssetsFromGrid, + getEnrichmentState, + setEnrichmentState, + comfyToast, + t, + triggerStartupScan, + reportError, + registerCleanableListener, +}) { + api._mjrExecutedHandler = (event) => { + try { + executionRuntime.handleExecutedEvent(event, { appRef }); + } catch (error) { + reportError(error, "entry.executed"); + } + }; + registerCleanableListener(runtime, api, "executed", api._mjrExecutedHandler); + + api._mjrExecutionStartHandler = (event) => { + try { + executionRuntime.handleExecutionStart(event, { + setCurrentJobId: (jobId) => liveStreamModule?.setCurrentJobId(jobId), + }); + } catch (error) { + reportError(error, "entry.execution_start"); + } + }; + api._mjrExecutionEndHandler = (event) => { + try { + executionRuntime.handleExecutionEnd(event, { + setCurrentJobId: (jobId) => liveStreamModule?.setCurrentJobId(jobId), + }); + } catch (error) { + reportError(error, "entry.execution_end"); + } + }; + api._mjrStacksUpdatedHandler = (event) => { + try { + executionRuntime.notifyStacksUpdated(event?.detail || {}); + } catch (error) { + reportError(error, "entry.stacks_updated"); + } + }; + api._mjrRuntimeStatusHandler = (event) => { + try { + const detail = event?.detail || {}; + emitRuntimeStatus({ + progress_node: detail?.node ?? null, + progress_value: Number.isFinite(Number(detail?.value)) ? Number(detail.value) : null, + progress_max: Number.isFinite(Number(detail?.max)) ? Number(detail.max) : null, + queue_remaining: Number.isFinite(Number(detail?.exec_info?.queue_remaining)) + ? Number(detail.exec_info.queue_remaining) + : Number.isFinite(Number(detail?.queue_remaining)) + ? Number(detail.queue_remaining) + : ensureExecutionRuntime().queue_remaining, + }); + } catch (error) { + reportError(error, "entry.runtime_status"); + } + }; + api._mjrExecutionCachedHandler = (event) => { + try { + const detail = event?.detail || {}; + emitRuntimeStatus({ + active_prompt_id: + detail?.prompt_id || detail?.promptId || ensureExecutionRuntime().active_prompt_id || null, + cached_nodes: Array.isArray(detail?.nodes) ? detail.nodes.slice() : [], + }); + } catch (error) { + reportError(error, "entry.execution_cached"); + } + }; + registerCleanableListener(runtime, api, "execution_start", api._mjrExecutionStartHandler); + registerCleanableListener(runtime, api, "execution_success", api._mjrExecutionEndHandler); + registerCleanableListener(runtime, api, "execution_error", api._mjrExecutionEndHandler); + registerCleanableListener(runtime, api, "execution_interrupted", api._mjrExecutionEndHandler); + registerCleanableListener(runtime, api, "mjr.stacks.updated", api._mjrStacksUpdatedHandler); + registerCleanableListener(runtime, api, "progress", api._mjrRuntimeStatusHandler); + registerCleanableListener(runtime, api, "status", api._mjrRuntimeStatusHandler); + registerCleanableListener(runtime, api, EVENTS.RUNTIME_STATUS, api._mjrRuntimeStatusHandler); + registerCleanableListener(runtime, api, "execution_cached", api._mjrExecutionCachedHandler); + + api._mjrAssetAddedHandler = (event) => { + try { + const grid = getActiveGridContainer(); + if (grid && event?.detail) { + const liveEvent = executionRuntime.prepareLiveAssetEvent(event.detail); + if (liveEvent?.defer) return; + const detail = liveEvent?.detail || event.detail; + const renderable = executionRuntime.isRenderableLiveAsset(detail); + if (renderable) { + pushGeneratedAsset(detail); + } + const scope = grid.dataset?.mjrScope || "output"; + if (scope !== "output" && scope !== "all") return; + const query = grid.dataset?.mjrQuery || "*"; + const canDirectUpsert = String(query).trim() === "*"; + if (canDirectUpsert && renderable) { + const handled = !!upsertAsset(grid, detail); + if (handled) { + try { + window.__mjrLastAssetUpsert = Date.now(); + } catch (e) { + console.debug?.(e); + } + } + } + } + } catch (error) { + reportError(error, "entry.asset_added"); + } + }; + registerCleanableListener(runtime, api, "mjr-asset-added", api._mjrAssetAddedHandler); + + api._mjrAssetUpdatedHandler = (event) => { + try { + const grid = getActiveGridContainer(); + if (grid && event?.detail) { + const liveEvent = executionRuntime.prepareLiveAssetEvent(event.detail); + if (liveEvent?.defer) return; + const detail = liveEvent?.detail || event.detail; + if (executionRuntime.isRenderableLiveAsset(detail)) { + pushGeneratedAsset(detail); + } + const assetId = String(detail?.id || "").trim(); + const kind = String(detail?.kind || "").trim(); + const filename = String(detail?.filename || "").trim(); + const filepath = String(detail?.filepath || "").trim(); + const hasRenderableFields = !!kind && (!!filename || !!filepath); + let existsInGrid = false; + if (assetId) { + try { + existsInGrid = !!grid.querySelector( + `.mjr-asset-card[data-mjr-asset-id="${assetId.replace(/([!"#$%&'()*+,./:;<=>?@[\\\]^`{|}~])/g, "\\$1")}"]`, + ); + } catch (e) { + console.debug?.(e); + } + } + const canUpsert = hasRenderableFields || existsInGrid; + const handled = canUpsert ? !!upsertAsset(grid, detail) : false; + if (handled) { + try { + window.__mjrLastAssetUpsert = Date.now(); + } catch (e) { + console.debug?.(e); + } + } + } + } catch (error) { + reportError(error, "entry.asset_updated"); + } + }; + registerCleanableListener(runtime, api, "mjr-asset-updated", api._mjrAssetUpdatedHandler); + + api._mjrScanCompleteHandler = (event) => { + try { + const detail = event?.detail || {}; + window.dispatchEvent(new CustomEvent(EVENTS.SCAN_COMPLETE, { detail })); + } catch (e) { + console.debug?.(e); + } + }; + registerCleanableListener(runtime, api, EVENTS.SCAN_COMPLETE, api._mjrScanCompleteHandler); + + api._mjrScanProgressHandler = (event) => { + try { + const detail = event?.detail || {}; + window.dispatchEvent(new CustomEvent(EVENTS.SCAN_PROGRESS, { detail })); + } catch (e) { + console.debug?.(e); + } + }; + registerCleanableListener(runtime, api, EVENTS.SCAN_PROGRESS, api._mjrScanProgressHandler); + + api._mjrAssetIndexingHandler = (event) => { + try { + const detail = event?.detail || {}; + window.dispatchEvent(new CustomEvent(EVENTS.ASSET_INDEXING, { detail })); + } catch (e) { + console.debug?.(e); + } + }; + registerCleanableListener(runtime, api, EVENTS.ASSET_INDEXING, api._mjrAssetIndexingHandler); + + api._mjrAssetIndexedHandler = (event) => { + try { + const detail = event?.detail || {}; + window.dispatchEvent(new CustomEvent(EVENTS.ASSET_INDEXED, { detail })); + } catch (e) { + console.debug?.(e); + } + }; + registerCleanableListener(runtime, api, EVENTS.ASSET_INDEXED, api._mjrAssetIndexedHandler); + + api._mjrEnrichmentStatusHandler = (event) => { + try { + const detail = event?.detail || {}; + const queued = Number(detail?.queued); + const queueLeft = Number(detail?.queue_left); + const queueLen = Number.isFinite(queued) + ? Math.max(0, Math.floor(queued)) + : Number.isFinite(queueLeft) + ? Math.max(0, Math.floor(queueLeft)) + : 0; + const active = !!detail?.active || queueLen > 0; + const prev = getEnrichmentState().active; + setEnrichmentState(active, queueLen); + window.dispatchEvent(new CustomEvent(EVENTS.ENRICHMENT_STATUS, { detail })); + if (prev && !active) { + comfyToast(t("toast.enrichmentComplete", "Metadata enrichment complete"), "success", 2600); + } + } catch (e) { + console.debug?.(e); + } + }; + registerCleanableListener(runtime, api, EVENTS.ENRICHMENT_STATUS, api._mjrEnrichmentStatusHandler); + + api._mjrDbRestoreStatusHandler = (event) => { + try { + const detail = event?.detail || {}; + const step = String(detail?.step || ""); + const level = String(detail?.level || "info"); + const op = String(detail?.operation || ""); + window.dispatchEvent(new CustomEvent(EVENTS.DB_RESTORE_STATUS, { detail })); + const isDelete = op === "delete_db"; + const isReset = op === "reset_index"; + if (isDelete && !["started", "done", "failed"].includes(step)) { + return; + } + const map = { + started: isDelete + ? t("toast.dbDeleteTriggered", "Deleting database and rebuilding...") + : isReset + ? t("toast.resetTriggered", "Reset triggered: Reindexing all files...") + : t("toast.dbRestoreStarted", "DB restore started"), + stopping_workers: t("toast.dbRestoreStopping", "Stopping running workers"), + resetting_db: t("toast.dbRestoreResetting", "Unlocking and resetting database"), + delete_db: t("toast.dbDeleteTriggered", "Deleting database and rebuilding..."), + recreate_db: t("toast.dbRestoreReplacing", "Recreating database"), + replacing_files: t("toast.dbRestoreReplacing", "Replacing database files"), + restarting_scan: t("toast.dbRestoreRescan", "Restarting scan"), + done: isDelete + ? t("toast.dbDeleteSuccess", "Database deleted and rebuilt. Files are being reindexed.") + : isReset + ? t("toast.resetStarted", "Index reset started. Files will be reindexed in the background.") + : t("toast.dbRestoreSuccess", "Database backup restored"), + failed: String( + detail?.message || + (isDelete + ? t("toast.dbDeleteFailed", "Failed to delete database") + : isReset + ? t("toast.resetFailed", "Failed to reset index") + : t("toast.dbRestoreFailed", "Failed to restore DB backup")), + ), + }; + const msg = map[step] || String(detail?.message || ""); + if (!msg) return; + const tone = + level === "error" || step === "failed" + ? "error" + : step === "done" + ? "success" + : "info"; + comfyToast(msg, tone, step === "done" || step === "failed" ? 2800 : 1800); + } catch (e) { + console.debug?.(e); + } + }; + registerCleanableListener(runtime, api, EVENTS.DB_RESTORE_STATUS, api._mjrDbRestoreStatusHandler); + + const assetsDeletedHandler = (event) => { + try { + const ids = Array.isArray(event?.detail?.ids) + ? event.detail.ids.map((x) => String(x || "")).filter(Boolean) + : []; + if (!ids.length) return; + const grid = getActiveGridContainer(); + if (grid) removeAssetsFromGrid(grid, ids); + } catch (e) { + console.debug?.(e); + } + }; + registerCleanableListener(runtime, window, EVENTS.ASSETS_DELETED, assetsDeletedHandler); + + if (runtime && typeof runtime === "object") { + runtime.api = api; + runtime.assetsDeletedHandler = assetsDeletedHandler; + } + triggerStartupScan({ delayMs: 2500, idleOnly: true }); + console.debug("[Majoor] Real-time listener registered"); +} diff --git a/js/features/stacks/executionRuntimeController.js b/js/features/stacks/executionRuntimeController.js new file mode 100644 index 00000000..9adbe5b6 --- /dev/null +++ b/js/features/stacks/executionRuntimeController.js @@ -0,0 +1,558 @@ +import { APP_CONFIG } from "../../app/config.js"; +import { EVENTS } from "../../app/events.js"; +import { createExecutionAssetBuffer } from "./executionAssetBuffer.js"; +import { createJobFinalizeQueue } from "./finalizeQueue.js"; +import { createLiveAssetGate } from "./liveAssetGate.js"; +import { createTTLCache } from "../../utils/ttlCache.js"; + +const DEDUPE_TTL_MS = 2000; +const RECENT_FILES_MAX = 2000; +const INDEX_RETRYABLE_CODES = new Set([ + "DB_MAINTENANCE", + "TIMEOUT", + "NETWORK_ERROR", + "SERVICE_UNAVAILABLE", +]); +const INDEX_RETRY_MAX_ATTEMPTS = 8; +const INDEX_RETRY_BASE_DELAY_MS = 1200; +const INDEX_RETRY_MAX_DELAY_MS = 15_000; +const EXECUTION_START_TTL_MS = 10 * 60 * 1000; +const EXECUTION_STARTS_MAX = 500; +const ORPHAN_GENERATION_RETRY_MAX_ATTEMPTS = 6; + +export function createExecutionRuntimeController({ + post, + ENDPOINTS, + reportError, + extractOutputFiles, + ensureExecutionRuntime, + emitRuntimeStatus, + refreshGeneratedFeedHosts, + getActiveGridContainer, +}) { + const recentFiles = createTTLCache({ ttlMs: DEDUPE_TTL_MS, maxSize: RECENT_FILES_MAX }); + const executionStarts = createTTLCache({ + ttlMs: EXECUTION_START_TTL_MS, + maxSize: EXECUTION_STARTS_MAX, + }); + const orphanGenerationEntries = new Map(); + const executionAssetBuffer = createExecutionAssetBuffer(); + const liveAssetGate = createLiveAssetGate(); + let lastStacksUpdateSignature = ""; + let lastStacksUpdateAt = 0; + + function isJobTrackingDebugEnabled() { + try { + if (typeof window === "undefined") return false; + if (window.__MJR_DEBUG_JOB_TRACKING__ === true) return true; + return String(window.localStorage?.getItem("mjr.debug.jobTracking") || "").trim() === "1"; + } catch (e) { + console.debug?.(e); + return false; + } + } + + function debugJobTracking(stage, detail = {}) { + if (!isJobTrackingDebugEnabled()) return; + try { + console.debug("[Majoor][job-tracking]", stage, detail); + } catch (e) { + console.debug?.(e); + } + } + + function isExecutionGroupingEnabled() { + return !!APP_CONFIG.EXECUTION_GROUPING_ENABLED; + } + + function getExecutionRuntimePolicy() { + const groupingEnabled = isExecutionGroupingEnabled(); + return { + groupingEnabled, + useExecutionBuffer: groupingEnabled, + usePostExecutionScan: !groupingEnabled, + allowLiveStackFinalize: groupingEnabled, + }; + } + + function notifyStacksUpdated(detail = {}) { + debugJobTracking("stacks-updated", { + targeted_job_id: String(detail?.targeted_job_id || ""), + stacks: Array.isArray(detail?.stacks) + ? detail.stacks.map((item) => ({ + job_id: String(item?.job_id || ""), + stack_id: String(item?.stack_id || ""), + asset_count: Number(item?.asset_count || 0) || 0, + })) + : [], + }); + try { + const targeted = String(detail?.targeted_job_id || "").trim(); + 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) liveAssetGate.clearPendingJob(jobId); + } + } catch (e) { + console.debug?.(e); + } + try { + const stacks = Array.isArray(detail?.stacks) ? detail.stacks : []; + const signature = JSON.stringify({ + targeted_job_id: String(detail?.targeted_job_id || ""), + stacks: stacks.map((item) => ({ + job_id: String(item?.job_id || ""), + stack_id: String(item?.stack_id || ""), + asset_count: Number(item?.asset_count || 0) || 0, + })), + }); + const now = Date.now(); + if ( + signature && + signature === lastStacksUpdateSignature && + now - lastStacksUpdateAt < 1500 + ) { + return; + } + lastStacksUpdateSignature = signature; + lastStacksUpdateAt = now; + } catch (e) { + console.debug?.(e); + } + try { + window.dispatchEvent( + new CustomEvent("mjr:stacks-updated", { detail: { ...(detail || {}) } }), + ); + } catch (e) { + console.debug?.(e); + } + try { + refreshGeneratedFeedHosts(); + } catch (e) { + console.debug?.(e); + } + try { + const grid = getActiveGridContainer(); + const scope = String(grid?.dataset?.mjrScope || "") + .trim() + .toLowerCase(); + if (grid && (scope === "output" || scope === "all")) { + window.dispatchEvent( + new CustomEvent(EVENTS.RELOAD_GRID, { + detail: { reason: "stacks-updated", ...(detail || {}) }, + }), + ); + } + } catch (e) { + console.debug?.(e); + } + } + + const stackFinalizeQueue = createJobFinalizeQueue({ + defaultDelayMs: 900, + async postJob(targetJobId) { + debugJobTracking("auto-stack:start", { job_id: targetJobId }); + try { + const res = await post(ENDPOINTS.STACKS_AUTO_STACK, { + mode: "job_id", + job_id: targetJobId, + }); + if (!res?.ok) { + liveAssetGate.clearPendingJob(targetJobId); + reportError( + new Error(String(res?.error || "Auto stack failed")), + "executionRuntime.execution_end.auto_stack", + ); + return; + } + debugJobTracking("auto-stack:done", { + job_id: targetJobId, + stacks: Array.isArray(res?.data?.stacks) ? res.data.stacks.length : 0, + }); + notifyStacksUpdated(res?.data || { targeted_job_id: targetJobId }); + } catch (error) { + reportError(error, "executionRuntime.execution_end.auto_stack"); + } + }, + }); + + function rememberExecutionStart(promptId, timestampMs) { + if (!promptId) return; + const ts = Number(timestampMs) || Date.now(); + executionStarts.set(String(promptId), ts, { at: ts }); + executionStarts.prune(); + } + + function dedupeFiles(files) { + recentFiles.prune(); + const fresh = []; + for (const f of files) { + const key = `${f.type || ""}|${f.subfolder || ""}|${f.filename || ""}`.toLowerCase(); + if (!recentFiles.has(key)) { + recentFiles.set(key, true); + fresh.push(f); + } + } + return fresh; + } + + function indexRetryDelayMs(attempt) { + const a = Math.max(1, Number(attempt) || 1); + const exp = Math.min(6, a - 1); + return Math.min(INDEX_RETRY_MAX_DELAY_MS, INDEX_RETRY_BASE_DELAY_MS * 2 ** exp); + } + + function shouldRetryIndexResponse(res) { + const code = String(res?.code || "") + .trim() + .toUpperCase(); + return INDEX_RETRYABLE_CODES.has(code); + } + + function scheduleGenerationIndex(files, attempt = 1, meta = {}) { + const payloadFiles = Array.isArray(files) ? files : []; + if (!payloadFiles.length) return; + const payload = { files: payloadFiles, origin: "generation" }; + if (meta?.prompt_id || meta?.promptId) { + payload.prompt_id = String(meta.prompt_id || meta.promptId); + } + debugJobTracking("index:direct", { + prompt_id: String(payload.prompt_id || ""), + attempt, + files: payloadFiles.map((item) => String(item?.filename || item?.filepath || "")), + }); + post(ENDPOINTS.INDEX_FILES, payload) + .then((res) => { + if (res?.ok) return; + if (shouldRetryIndexResponse(res) && attempt < INDEX_RETRY_MAX_ATTEMPTS) { + const delayMs = indexRetryDelayMs(attempt); + setTimeout( + () => scheduleGenerationIndex(payloadFiles, attempt + 1, meta), + delayMs, + ); + return; + } + try { + const code = String(res?.code || "INDEX_FAILED"); + const msg = String(res?.error || "Indexing generated files failed"); + reportError(new Error(`${code}: ${msg}`), "executionRuntime.executed.index"); + } catch (e) { + console.debug?.(e); + } + }) + .catch((error) => reportError(error, "executionRuntime.executed.index")); + } + + function resolveGenerationPromptId(meta = {}) { + const explicitPromptId = String(meta?.prompt_id || meta?.promptId || "").trim(); + if (explicitPromptId) return explicitPromptId; + try { + return String(ensureExecutionRuntime()?.active_prompt_id || "").trim(); + } catch (e) { + console.debug?.(e); + return ""; + } + } + + function orphanGenerationKey(files) { + if (!Array.isArray(files) || !files.length) return ""; + return files + .map((item) => String(item?.filepath || item?.filename || "")) + .filter(Boolean) + .sort() + .join("|"); + } + + function enqueueOrphanGenerationIndex(files, meta = {}, attempt = 1) { + const payloadFiles = Array.isArray(files) ? files : []; + if (!payloadFiles.length) return; + const key = orphanGenerationKey(payloadFiles); + if (!key) { + scheduleGenerationIndex(payloadFiles, 1, meta); + return; + } + const existing = orphanGenerationEntries.get(key); + if (existing?.timer) clearTimeout(existing.timer); + const nextAttempt = Math.max(1, Number(attempt) || 1); + const delayMs = Math.min(2000, 250 * nextAttempt); + const timer = setTimeout(() => { + orphanGenerationEntries.delete(key); + bufferGenerationIndex(payloadFiles, meta, nextAttempt + 1); + }, delayMs); + orphanGenerationEntries.set(key, { + files: payloadFiles, + meta, + attempt: nextAttempt, + timer, + }); + } + + function flushOrphanGenerationEntries(promptId) { + const normalizedPromptId = String(promptId || "").trim(); + if (!normalizedPromptId || orphanGenerationEntries.size <= 0) return; + for (const [key, entry] of Array.from(orphanGenerationEntries.entries())) { + try { + if (entry?.timer) clearTimeout(entry.timer); + } catch (e) { + console.debug?.(e); + } + orphanGenerationEntries.delete(key); + executionAssetBuffer.add(normalizedPromptId, entry?.files || []); + } + } + + function clearOrphanGenerationEntries() { + for (const [key, entry] of Array.from(orphanGenerationEntries.entries())) { + try { + if (entry?.timer) clearTimeout(entry.timer); + } catch (e) { + console.debug?.(e); + } + orphanGenerationEntries.delete(key); + } + } + + function bufferGenerationIndex(files, meta = {}, attempt = 1) { + const promptId = resolveGenerationPromptId(meta); + debugJobTracking("buffer:add", { + prompt_id: promptId, + explicit_prompt_id: String(meta?.prompt_id || meta?.promptId || ""), + attempt, + file_count: Array.isArray(files) ? files.length : 0, + files: Array.isArray(files) + ? files.map((item) => String(item?.filename || item?.filepath || "")) + : [], + }); + if (!promptId) { + if (attempt < ORPHAN_GENERATION_RETRY_MAX_ATTEMPTS) { + enqueueOrphanGenerationIndex(files, meta, attempt); + return; + } + scheduleGenerationIndex(files, 1, meta); + return; + } + executionAssetBuffer.add(promptId, files); + } + + function scheduleFinalizeExecutionStacks(jobId, delayMs = 900) { + if (!isExecutionGroupingEnabled()) return; + debugJobTracking("finalize:schedule", { job_id: String(jobId || ""), delay_ms: delayMs }); + stackFinalizeQueue.schedule(jobId, delayMs); + } + + function schedulePostExecutionScan(delayMs = 1200) { + const waitMs = Math.max(0, Number(delayMs) || 0); + setTimeout(() => { + post(ENDPOINTS.SCAN, { + recursive: true, + incremental: true, + fast: true, + background_metadata: true, + }).catch((error) => reportError(error, "executionRuntime.execution_end.scan")); + }, waitMs); + } + + function flushBufferedGenerationIndex(jobId) { + const policy = getExecutionRuntimePolicy(); + const promptId = String(jobId || "").trim(); + if (!promptId) return; + const files = executionAssetBuffer.take(promptId); + debugJobTracking("buffer:flush", { + prompt_id: promptId, + file_count: files.length, + files: files.map((item) => String(item?.filename || item?.filepath || "")), + }); + if (policy.allowLiveStackFinalize) { + liveAssetGate.markPendingJob(promptId, { files, expectedCount: files.length }); + } + if (!files.length) { + if (policy.allowLiveStackFinalize) { + scheduleFinalizeExecutionStacks(promptId); + } + return; + } + post(ENDPOINTS.INDEX_FILES, { + files, + origin: "generation", + prompt_id: promptId, + }) + .then((res) => { + if (!res?.ok) { + if (policy.allowLiveStackFinalize) { + liveAssetGate.clearPendingJob(promptId); + } + reportError( + new Error(String(res?.error || "Indexing generated files failed")), + "executionRuntime.executed.index", + ); + return; + } + debugJobTracking("index:flush:done", { + prompt_id: promptId, + file_count: files.length, + }); + if (policy.allowLiveStackFinalize) { + scheduleFinalizeExecutionStacks(promptId, 1800); + } + }) + .catch((error) => { + if (policy.allowLiveStackFinalize) { + liveAssetGate.clearPendingJob(promptId); + } + reportError(error, "executionRuntime.executed.index"); + }); + } + + function prepareLiveAssetEvent(detail) { + if (!isExecutionGroupingEnabled()) { + return { detail, jobId: "", defer: false }; + } + try { + return liveAssetGate.prepare(detail); + } catch (e) { + console.debug?.(e); + return { detail, jobId: "", defer: false }; + } + } + + function isRenderableLiveAsset(detail) { + const kind = String(detail?.kind || "").trim(); + const filename = String(detail?.filename || "").trim(); + const filepath = String(detail?.filepath || "").trim(); + return !!kind && (!!filename || !!filepath); + } + + function handleExecutedEvent(event, { appRef }) { + const policy = getExecutionRuntimePolicy(); + if (!policy.useExecutionBuffer) return; + const outputFiles = dedupeFiles(extractOutputFiles(event?.detail?.output)); + if (!outputFiles.length) return; + const promptId = event?.detail?.prompt_id || event?.detail?.promptId; + const sourceNodeId = String(event?.detail?.node || event?.detail?.display_node || "").trim(); + const startTs = promptId ? executionStarts.get(String(promptId)) : null; + const genTimeMs = startTs ? Math.max(0, Date.now() - startTs) : 0; + let sourceNodeType = ""; + try { + if (sourceNodeId && typeof appRef !== "undefined" && appRef?.graph) { + const graphNode = appRef.graph.getNodeById(Number(sourceNodeId)); + if (graphNode) { + sourceNodeType = String(graphNode.type || graphNode.comfyClass || "").trim(); + } + } + } catch (e) { + console.debug?.(e); + } + const files = outputFiles.map((f) => ({ + ...f, + ...(genTimeMs > 0 ? { generation_time_ms: genTimeMs } : {}), + ...(sourceNodeId ? { source_node_id: sourceNodeId } : {}), + ...(sourceNodeType ? { source_node_type: sourceNodeType } : {}), + })); + debugJobTracking("executed:event", { + prompt_id: String(promptId || ""), + source_node_id: sourceNodeId, + source_node_type: sourceNodeType, + file_count: files.length, + files: files.map((item) => String(item?.filename || item?.filepath || "")), + }); + try { + window.dispatchEvent( + new CustomEvent(EVENTS.NEW_GENERATION_OUTPUT, { detail: { files } }), + ); + } catch (e) { + console.debug?.(e); + } + bufferGenerationIndex(files, { prompt_id: promptId }); + } + + function handleExecutionStart(event, { setCurrentJobId }) { + const promptId = event?.detail?.prompt_id || event?.detail?.promptId; + const ts = event?.detail?.timestamp; + debugJobTracking("execution:start", { + prompt_id: String(promptId || ""), + timestamp: Number(ts) || null, + }); + rememberExecutionStart(promptId, ts); + setCurrentJobId?.(promptId || null); + emitRuntimeStatus({ + active_prompt_id: promptId || null, + cached_nodes: [], + progress_node: null, + progress_value: null, + progress_max: null, + }); + if (promptId) { + flushOrphanGenerationEntries(promptId); + } + } + + function handleExecutionEnd(event, { setCurrentJobId }) { + const policy = getExecutionRuntimePolicy(); + const promptId = event?.detail?.prompt_id || event?.detail?.promptId; + debugJobTracking("execution:end", { + type: String(event?.type || ""), + prompt_id: String(promptId || ""), + }); + if (promptId) { + executionStarts.delete(String(promptId)); + } + executionStarts.prune(); + setCurrentJobId?.(null); + emitRuntimeStatus({ + active_prompt_id: null, + cached_nodes: [], + progress_node: null, + progress_value: null, + progress_max: null, + }); + if (promptId && String(event?.type || "") !== "execution_success") { + debugJobTracking("buffer:clear", { + prompt_id: String(promptId || ""), + reason: String(event?.type || ""), + }); + executionAssetBuffer.clear(promptId); + if (policy.allowLiveStackFinalize) { + liveAssetGate.clearPendingJob(promptId); + } + clearOrphanGenerationEntries(); + } + if (String(event?.type || "") === "execution_success") { + if (policy.usePostExecutionScan) { + schedulePostExecutionScan(1800); + } else { + if (promptId) { + flushOrphanGenerationEntries(promptId); + } + flushBufferedGenerationIndex(promptId); + } + } + } + + function handleNodeOutputsUpdated(nodeOutputs) { + if (!isExecutionGroupingEnabled()) return; + const files = []; + for (const [, outputs] of nodeOutputs instanceof Map ? nodeOutputs.entries() : []) { + const found = extractOutputFiles(outputs); + if (found?.length) { + for (const f of found) files.push(f); + } + } + if (!files.length || !isExecutionGroupingEnabled()) return; + debugJobTracking("node-outputs:updated", { + prompt_id: resolveGenerationPromptId(), + file_count: files.length, + }); + bufferGenerationIndex(files, { prompt_id: resolveGenerationPromptId() }); + } + + return { + notifyStacksUpdated, + prepareLiveAssetEvent, + isRenderableLiveAsset, + handleExecutedEvent, + handleExecutionStart, + handleExecutionEnd, + handleNodeOutputsUpdated, + }; +} diff --git a/mjr_am_backend/config.py b/mjr_am_backend/config.py index 6c65ca8f..800db530 100644 --- a/mjr_am_backend/config.py +++ b/mjr_am_backend/config.py @@ -315,11 +315,23 @@ INDEX_DEDUPE_MAX = _env_int(5000, "MJR_AM_INDEX_DEDUPE_MAX", "MAJOOR_INDEX_DEDUP # File watcher (watches for manual file additions in output/custom directories) # Disable with MJR_ENABLE_WATCHER=0 WATCHER_ENABLED = _env_bool(True, "MJR_AM_ENABLE_WATCHER", "MJR_ENABLE_WATCHER") +WATCHER_START_ON_BOOT = _env_bool( + False, + "MJR_AM_WATCHER_START_ON_BOOT", + "MJR_WATCHER_START_ON_BOOT", +) # 3000ms absorbs file-write bursts from generators that write temp + final files. WATCHER_DEFAULT_DEBOUNCE_MS = _env_int(3000, "MJR_AM_WATCHER_DEBOUNCE_MS", "MJR_WATCHER_DEBOUNCE_MS", min_value=0, max_value=120_000) WATCHER_DEFAULT_DEDUPE_TTL_MS = _env_int(3000, "MJR_AM_WATCHER_DEDUPE_TTL_MS", "MJR_WATCHER_DEDUPE_TTL_MS", min_value=0, max_value=120_000) WATCHER_DEBOUNCE_MS = WATCHER_DEFAULT_DEBOUNCE_MS WATCHER_DEDUPE_TTL_MS = WATCHER_DEFAULT_DEDUPE_TTL_MS +WATCHER_GENERATED_GRACE_SECONDS = _env_float( + 15.0, + "MJR_AM_WATCHER_GENERATED_GRACE_SECONDS", + "MJR_WATCHER_GENERATED_GRACE_SECONDS", + min_value=0.0, + max_value=300.0, +) # Backward-compatible aliases: # - MAJOOR_WATCHER_MAX_PENDING_FILES # - MAJOOR_WATCHER_MIN_FILE_SIZE @@ -441,10 +453,15 @@ VECTOR_SEARCH_ENABLED = _env_bool( "MAJOOR_ENABLE_VECTOR_SEARCH", ) VECTOR_INDEX_ON_SCAN = _env_bool( - True, + False, "MJR_AM_VECTOR_INDEX_ON_SCAN", "MAJOOR_VECTOR_INDEX_ON_SCAN", ) +EXECUTION_GROUPING_ENABLED = _env_bool( + True, + "MJR_AM_EXECUTION_GROUPING_ENABLED", + "MAJOOR_EXECUTION_GROUPING_ENABLED", +) def is_vector_search_enabled() -> bool: @@ -465,6 +482,15 @@ def is_vector_index_on_scan_enabled() -> bool: "MAJOOR_VECTOR_INDEX_ON_SCAN", ) + +def is_execution_grouping_enabled() -> bool: + """Return runtime toggle for job_id / stack_id execution grouping logic.""" + return _env_bool( + EXECUTION_GROUPING_ENABLED, + "MJR_AM_EXECUTION_GROUPING_ENABLED", + "MAJOOR_EXECUTION_GROUPING_ENABLED", + ) + # Primary multimodal model name (default: SigLIP2 SO400M). VECTOR_MODEL_NAME = str( _env_raw( @@ -514,7 +540,7 @@ VECTOR_EMBEDDING_DIM = _env_int(1152, "MJR_AM_VECTOR_DIM", min_value=64, max_val # Pre-warm the text-query model path and Faiss index during startup prewarm. VECTOR_PREWARM_ON_STARTUP = _env_bool( - True, + False, "MJR_AM_VECTOR_PREWARM_ON_STARTUP", "MAJOOR_VECTOR_PREWARM_ON_STARTUP", ) diff --git a/mjr_am_backend/deps.py b/mjr_am_backend/deps.py index 2b341c06..43793483 100644 --- a/mjr_am_backend/deps.py +++ b/mjr_am_backend/deps.py @@ -19,12 +19,14 @@ from .config import ( INDEX_DB, VECTORS_DB, WATCHER_ENABLED, + WATCHER_START_ON_BOOT, initialize_directories, is_vector_search_enabled, ) from .features.duplicates import DuplicatesService from .features.health import HealthService from .features.index import IndexService +from .features.index.vector_runtime import ensure_vector_runtime from .features.index.watcher import OutputWatcher from .features.index.watcher_scope import build_watch_paths, load_watcher_scope from .features.metadata import MetadataService @@ -127,13 +129,7 @@ def _build_services_dict( # ── Vector / multimodal search (opt-in) ─────────────────────────── if is_vector_search_enabled(): try: - from .features.index.vector_searcher import VectorSearcher - from .features.index.vector_service import VectorService - - vs = VectorService() - services["vector_service"] = vs - services["vector_searcher"] = VectorSearcher(db, vs) - log_success(logger, "SigLIP2/X-CLIP vector search enabled") + log_success(logger, "SigLIP2/X-CLIP vector search enabled (lazy init)") except Exception as exc: logger.warning("Vector search disabled: %s", exc) return services @@ -155,7 +151,7 @@ def _attach_rating_tags_sync_worker(services: dict, exiftool: ExifTool) -> None: async def _attach_watcher_if_enabled(services: dict, index_service: IndexService) -> None: - if not WATCHER_ENABLED: + if not WATCHER_ENABLED or not WATCHER_START_ON_BOOT: return try: watcher = await _create_watcher(index_service) @@ -233,11 +229,13 @@ async def build_services(db_path: str | None = None) -> Result[dict]: index_service, settings_service, ) - # Attach vector services to IndexService for automatic background embedding. - if "vector_service" in services and callable(getattr(index_service, "set_vector_services", None)): - index_service.set_vector_services( - services["vector_service"], - services.get("vector_searcher"), + if callable(getattr(index_service, "set_vector_services_resolver", None)): + index_service.set_vector_services_resolver( + lambda: ensure_vector_runtime( + services, + logger=logger, + reason="automatic-indexing", + ) ) services["watcher_scope"] = await _load_watcher_scope_or_default(db) services["watcher_scope_by_user"] = {} diff --git a/mjr_am_backend/features/index/service.py b/mjr_am_backend/features/index/service.py index 9d2050d9..d55a5cd1 100644 --- a/mjr_am_backend/features/index/service.py +++ b/mjr_am_backend/features/index/service.py @@ -27,6 +27,7 @@ from .scan_batch_utils import compute_state_hash, normalize_filepath_str from .scanner import IndexScanner from .searcher import IndexSearcher from .updater import AssetUpdater +from .vector_runtime import ensure_vector_runtime if TYPE_CHECKING: from .vector_searcher import VectorSearcher @@ -131,6 +132,9 @@ async def _update_path_keyed_index_tables( def _build_runtime_vector_services(db: Sqlite) -> tuple[VectorService, VectorSearcher]: + """ + Backward-compatible hook kept for tests and narrow monkeypatching. + """ from .vector_searcher import VectorSearcher as _VS from .vector_service import VectorService as _VSvc @@ -159,6 +163,7 @@ class IndexService: self._has_tags_text_column = has_tags_text_column self._vector_service: VectorService | None = None self._vector_searcher: VectorSearcher | None = None + self._vector_services_resolver = None logger.debug("asset_metadata.tags_text column available: %s", self._has_tags_text_column) # Initialize specialized components @@ -179,6 +184,10 @@ class IndexService: self._vector_searcher = vector_searcher self._scanner.set_vector_services(vector_service, vector_searcher) + def set_vector_services_resolver(self, resolver) -> None: + """Attach a shared lazy resolver so all runtime paths reuse one vector stack.""" + self._vector_services_resolver = resolver + def _ensure_vector_services(self) -> None: if self._vector_service is not None: self._scanner.set_vector_services(self._vector_service, self._vector_searcher) @@ -190,8 +199,26 @@ class IndexService: except Exception as exc: logger.debug("Runtime vector service initialization skipped: %s", exc) return - self.set_vector_services(vector_service, vector_searcher) - logger.info("Runtime vector services initialized for automatic indexing") + if vector_service is not None: + self.set_vector_services(vector_service, vector_searcher) + + async def _ensure_vector_services_async(self) -> None: + if self._vector_service is not None: + self._scanner.set_vector_services(self._vector_service, self._vector_searcher) + return + if not is_vector_search_enabled(): + return + resolver = self._vector_services_resolver + if callable(resolver): + try: + vector_service, vector_searcher = await resolver() + except Exception as exc: + logger.debug("Runtime vector service initialization skipped: %s", exc) + return + if vector_service is not None: + self.set_vector_services(vector_service, vector_searcher) + return + self._ensure_vector_services() # ==================== Scanning Operations ==================== @@ -220,7 +247,7 @@ class IndexService: Returns: Result with scan statistics """ - self._ensure_vector_services() + await self._ensure_vector_services_async() self._enricher.begin_scan_pause() try: result = await self._scanner.scan_directory( diff --git a/mjr_am_backend/features/index/vector_runtime.py b/mjr_am_backend/features/index/vector_runtime.py new file mode 100644 index 00000000..b4cfc110 --- /dev/null +++ b/mjr_am_backend/features/index/vector_runtime.py @@ -0,0 +1,66 @@ +""" +Shared lazy vector-service initialization. + +Centralizes runtime creation of VectorService/VectorSearcher so route handlers +and indexing flows do not each build their own instances with divergent state. +""" + +from __future__ import annotations + +import asyncio +from typing import Any + +from ...config import is_vector_search_enabled + + +def _get_vector_runtime_lock(services: dict[str, Any]) -> asyncio.Lock: + lock = services.get("_vector_runtime_lock") + if isinstance(lock, asyncio.Lock): + return lock + lock = asyncio.Lock() + services["_vector_runtime_lock"] = lock + return lock + + +def _build_vector_runtime(db: Any) -> tuple[Any, Any]: + from .vector_searcher import VectorSearcher + from .vector_service import VectorService + + vector_service = VectorService() + vector_searcher = VectorSearcher(db, vector_service) + return vector_service, vector_searcher + + +async def ensure_vector_runtime( + services: dict[str, Any] | None, + *, + logger: Any = None, + reason: str = "runtime", +) -> tuple[Any | None, Any | None]: + if not is_vector_search_enabled() or not isinstance(services, dict): + return None, None + + vector_service = services.get("vector_service") + vector_searcher = services.get("vector_searcher") + if vector_service is not None and vector_searcher is not None: + return vector_service, vector_searcher + + db = services.get("db") + if db is None: + return None, None + + lock = _get_vector_runtime_lock(services) + async with lock: + vector_service = services.get("vector_service") + vector_searcher = services.get("vector_searcher") + if vector_service is not None and vector_searcher is not None: + return vector_service, vector_searcher + vector_service, vector_searcher = _build_vector_runtime(db) + services["vector_service"] = vector_service + services["vector_searcher"] = vector_searcher + if logger is not None: + try: + logger.info("Vector services initialized lazily (%s)", str(reason or "runtime")) + except Exception: + pass + return vector_service, vector_searcher diff --git a/mjr_am_backend/features/index/watcher.py b/mjr_am_backend/features/index/watcher.py index 3e145f7e..98a5cd6c 100644 --- a/mjr_am_backend/features/index/watcher.py +++ b/mjr_am_backend/features/index/watcher.py @@ -30,6 +30,7 @@ from ...config import ( WATCHER_DEBOUNCE_MS, WATCHER_DEDUPE_TTL_MS, WATCHER_FLUSH_MAX_FILES, + WATCHER_GENERATED_GRACE_SECONDS, WATCHER_MAX_FILE_SIZE_BYTES, WATCHER_MAX_FLUSH_CONCURRENCY, WATCHER_MIN_FILE_SIZE_BYTES, @@ -428,11 +429,15 @@ class DebouncedWatchHandler(FileSystemEventHandler): def _filter_flush_candidates(self, candidates: list[str]) -> list[str]: files: list[str] = [] + deferred_recent: list[str] = [] for f in candidates: if _is_recent_generated(f): continue if not self._is_supported_flush_file(f): continue + if self._should_defer_recent_file(f): + deferred_recent.append(f) + continue size = self._flush_file_size_if_eligible(f) if size is None: continue @@ -441,8 +446,36 @@ class DebouncedWatchHandler(FileSystemEventHandler): logger.debug("Watcher skipping oversized file %s (%d bytes > %d)", f, size, max_size) continue files.append(f) + if deferred_recent: + self._requeue_deferred_recent_files(deferred_recent) return files + @staticmethod + def _should_defer_recent_file(filepath: str) -> bool: + grace_s = max(0.0, float(WATCHER_GENERATED_GRACE_SECONDS or 0.0)) + if grace_s <= 0.0: + return False + try: + stat = os.stat(filepath) + except OSError: + return False + now = time.time() + latest_ts = max( + float(getattr(stat, "st_mtime", 0.0) or 0.0), + float(getattr(stat, "st_ctime", 0.0) or 0.0), + ) + if latest_ts <= 0.0: + return False + return (now - latest_ts) < grace_s + + def _requeue_deferred_recent_files(self, files: list[str]) -> None: + now = time.time() + with self._lock: + for f in files: + if f not in self._pending and f not in self._overflow: + self._overflow[f] = now + self._schedule_flush() + @staticmethod def _is_supported_flush_file(filepath: str) -> bool: try: diff --git a/mjr_am_backend/features/runtime/bootstrap.py b/mjr_am_backend/features/runtime/bootstrap.py index e347b734..b1412295 100644 --- a/mjr_am_backend/features/runtime/bootstrap.py +++ b/mjr_am_backend/features/runtime/bootstrap.py @@ -21,6 +21,10 @@ async def apply_startup_settings( "Vector search setting restore failed: %s", "apply_vector_search_override_on_startup", ), + ( + "Execution grouping setting restore failed: %s", + "apply_execution_grouping_override_on_startup", + ), ( "HuggingFace token restore failed: %s", "apply_huggingface_token_on_startup", diff --git a/mjr_am_backend/routes/handlers/__init__.py b/mjr_am_backend/routes/handlers/__init__.py index 415183a4..8d65c8dd 100644 --- a/mjr_am_backend/routes/handlers/__init__.py +++ b/mjr_am_backend/routes/handlers/__init__.py @@ -20,6 +20,7 @@ from .vector_search import register_vector_search_routes from .vendor import register_vendor_routes from .version import register_version_routes from .viewer import register_viewer_routes +from .plugins import register_plugin_routes __all__ = [ "register_health_routes", @@ -42,4 +43,5 @@ __all__ = [ "register_audit_routes", "register_stacks_routes", "register_vendor_routes", + "register_plugin_routes", ] diff --git a/mjr_am_backend/routes/handlers/health.py b/mjr_am_backend/routes/handlers/health.py index 64890b9f..ea23c166 100644 --- a/mjr_am_backend/routes/handlers/health.py +++ b/mjr_am_backend/routes/handlers/health.py @@ -188,6 +188,15 @@ def _extract_vector_search_payload(body: dict) -> object | None: return None +def _extract_execution_grouping_payload(body: dict) -> object | None: + if "enabled" in body: + return body.get("enabled") + prefs = body.get("prefs") if isinstance(body.get("prefs"), dict) else {} + if isinstance(prefs, dict) and "enabled" in prefs: + return prefs.get("enabled") + return None + + def _extract_huggingface_token_payload(body: dict) -> str | None: if "token" in body: return str(body.get("token") or "") @@ -814,6 +823,67 @@ def register_health_routes(routes: web.RouteTableDef) -> None: ) return _json_response(response_result) + @routes.get("/mjr/am/settings/execution-grouping") + async def get_execution_grouping_settings(request): + svc, error_result = await _require_services() + if error_result: + return _json_response(error_result) + + settings_service = svc.get("settings") + if not settings_service: + return _json_response(Result.Err("SERVICE_UNAVAILABLE", "Settings service unavailable")) + + enabled = await settings_service.get_execution_grouping_enabled() + return _json_response(Result.Ok({"prefs": {"enabled": bool(enabled)}})) + + @routes.post("/mjr/am/settings/execution-grouping") + async def update_execution_grouping_settings(request): + csrf = _csrf_error(request) + if csrf: + return _json_response(Result.Err(ErrorCode.CSRF, csrf)) + auth = _require_write_access(request) + if not auth.ok: + return _json_response(auth) + + svc, error_result = await _require_services() + if error_result: + return _json_response(error_result) + + settings_service = svc.get("settings") + if not settings_service: + return _json_response(Result.Err(ErrorCode.SERVICE_UNAVAILABLE, "Settings service unavailable")) + + body_res = await _read_json(request) + if not body_res.ok: + return _json_response(body_res) + body = body_res.data or {} + + enabled = _extract_execution_grouping_payload(body) + if enabled is None: + return _json_response(Result.Err("INVALID_INPUT", "Missing execution-grouping enabled value")) + + result = await settings_service.set_execution_grouping_enabled(enabled) + if not result.ok: + await _audit_settings_write( + svc, + request, + "settings_execution_grouping", + "settings:execution_grouping", + result, + enabled=enabled, + ) + return _json_response(result) + response_result = Result.Ok({"prefs": {"enabled": bool(result.data)}}) + await _audit_settings_write( + svc, + request, + "settings_execution_grouping", + "settings:execution_grouping", + response_result, + enabled=enabled, + ) + return _json_response(response_result) + @routes.get("/mjr/am/settings/huggingface") async def get_huggingface_settings(request): svc, error_result = await _require_services() diff --git a/mjr_am_backend/routes/handlers/plugins.py b/mjr_am_backend/routes/handlers/plugins.py new file mode 100644 index 00000000..ce618409 --- /dev/null +++ b/mjr_am_backend/routes/handlers/plugins.py @@ -0,0 +1,67 @@ +""" +Plugin management HTTP handlers. + +Exposes endpoints: +- GET /mjr/am/plugins/list +- POST /mjr/am/plugins/{name}/enable +- POST /mjr/am/plugins/reload +""" +from __future__ import annotations + +from aiohttp import web + +from mjr_am_backend.shared import Result +from ..core import _require_services, _json_response + + +def register_plugin_routes(routes: web.RouteTableDef) -> None: + @routes.get("/mjr/am/plugins/list") + async def list_plugins(request: web.Request): + svc, err = await _require_services() + if err: + return _json_response(err) + + metadata = svc.get("metadata") if isinstance(svc, dict) else None + if not metadata or not hasattr(metadata, "plugin_manager"): + return _json_response(Result.Err("NOT_AVAILABLE", "Plugin manager not available")) + + try: + plugins = metadata.plugin_manager.list_plugins() + return _json_response(Result.Ok(plugins)) + except Exception as exc: + return _json_response(Result.Err("PLUGIN_ERROR", f"Failed to list plugins: {exc}")) + + @routes.post("/mjr/am/plugins/{name}/enable") + async def enable_plugin(request: web.Request): + svc, err = await _require_services() + if err: + return _json_response(err) + + metadata = svc.get("metadata") if isinstance(svc, dict) else None + if not metadata or not hasattr(metadata, "plugin_manager"): + return _json_response(Result.Err("NOT_AVAILABLE", "Plugin manager not available")) + + name = request.match_info.get("name", "") + try: + ok = metadata.plugin_manager.enable_plugin(name) + if not ok: + return _json_response(Result.Err("NOT_FOUND", f"Plugin not found or could not be enabled: {name}")) + return _json_response(Result.Ok({"enabled": True})) + except Exception as exc: + return _json_response(Result.Err("PLUGIN_ERROR", f"Failed to enable plugin: {exc}")) + + @routes.post("/mjr/am/plugins/reload") + async def reload_plugins(request: web.Request): + svc, err = await _require_services() + if err: + return _json_response(err) + + metadata = svc.get("metadata") if isinstance(svc, dict) else None + if not metadata or not hasattr(metadata, "plugin_manager"): + return _json_response(Result.Err("NOT_AVAILABLE", "Plugin manager not available")) + + try: + count = await metadata.plugin_manager.reload() + return _json_response(Result.Ok({"reloaded": count})) + except Exception as exc: + return _json_response(Result.Err("PLUGIN_ERROR", f"Failed to reload plugins: {exc}")) diff --git a/mjr_am_backend/routes/handlers/scan.py b/mjr_am_backend/routes/handlers/scan.py index 38a4b1b4..5ead22d5 100644 --- a/mjr_am_backend/routes/handlers/scan.py +++ b/mjr_am_backend/routes/handlers/scan.py @@ -25,6 +25,7 @@ from mjr_am_backend.config import ( SCAN_DEFAULT_BACKGROUND_METADATA, SCAN_DEFAULT_FAST, TO_THREAD_TIMEOUT_S, + is_execution_grouping_enabled, ) from mjr_am_backend.custom_roots import resolve_custom_root from mjr_am_backend.features.index.metadata_helpers import MetadataHelpers @@ -111,6 +112,33 @@ def _send_prompt_event(event: str, payload: Any) -> None: pass +def _normalize_job_id(value: Any) -> str: + return str(value or "").strip() + + +def _normalize_stack_id(value: Any) -> int | None: + try: + stack_id = int(value) + except Exception: + return None + return stack_id if stack_id > 0 else None + + +def _should_preserve_existing_job(existing_job_id: Any, candidate_job_id: Any) -> bool: + existing = _normalize_job_id(existing_job_id) + candidate = _normalize_job_id(candidate_job_id) + if not existing or not candidate: + return False + return existing != candidate + + +def _resolve_stack_assignment(existing_stack_id: Any, candidate_stack_id: Any) -> int | None: + candidate = _normalize_stack_id(candidate_stack_id) + if candidate is not None: + return candidate + return _normalize_stack_id(existing_stack_id) + + def _scan_enqueued_from_kickoff(result: Any) -> bool: # Backward-compatible for monkeypatched tests that return None. if result is None: @@ -803,34 +831,14 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: gen_time_by_id[int(asset_id)] = gt except Exception: logger.debug("Failed to upsert generation time for asset %s", asset_id) - if prompt_lookup or prompt_lookup_by_path: + if is_execution_grouping_enabled() and (prompt_lookup or prompt_lookup_by_path): try: - # Build job_id → stack_id lookup from existing stacks - all_prompt_ids = set( - pid for pid in - (list(prompt_lookup_by_path.values()) + list(prompt_lookup.values())) - if pid - ) - job_stack_lookup: dict[str, int] = {} - if all_prompt_ids: - for _jstart in range(0, len(all_prompt_ids), _MAX_IN_BATCH): - _jchunk = list(all_prompt_ids)[_jstart:_jstart + _MAX_IN_BATCH] - _jph = ",".join("?" * len(_jchunk)) - _jres = await db_adapter.aquery( - f"SELECT job_id, id FROM asset_stacks WHERE job_id IN ({_jph})", - tuple(_jchunk), - ) - if _jres.ok and _jres.data: - for _jr in _jres.data: - job_stack_lookup[str(_jr["job_id"])] = int(_jr["id"]) - prompt_affected_stack_ids: set[int] = set() - if prompt_lookup_by_path: for path_value, row_prompt_id in prompt_lookup_by_path.items(): if not row_prompt_id: continue q_res = await db_adapter.aquery( - "SELECT id, stack_id FROM assets WHERE filepath = ? OR filepath = ? COLLATE NOCASE", + "SELECT id, job_id, stack_id FROM assets WHERE filepath = ? OR filepath = ? COLLATE NOCASE", (path_value, path_value), ) if not (q_res.ok and q_res.data): @@ -840,22 +848,26 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: asset_id = int(row.get("id") or 0) if not asset_id: continue - old_stack_id = int(row.get("stack_id") or 0) - if old_stack_id and old_stack_id != new_stack_id: - prompt_affected_stack_ids.add(old_stack_id) + existing_job_id = row.get("job_id") + if _should_preserve_existing_job(existing_job_id, row_prompt_id): + logger.debug( + "Skipping prompt correlation overwrite for asset %s: existing job_id=%s candidate=%s", + asset_id, + existing_job_id, + row_prompt_id, + ) + continue await db_adapter.aexecute( - "UPDATE assets SET job_id = ?, stack_id = COALESCE(?, stack_id) WHERE id = ?", - (row_prompt_id, new_stack_id, asset_id), + "UPDATE assets SET job_id = ? WHERE id = ?", + (row_prompt_id, asset_id), ) - if new_stack_id: - prompt_affected_stack_ids.add(new_stack_id) for start in range(0, len(fnames_for_prompt), _MAX_IN_BATCH): chunk = fnames_for_prompt[start:start + _MAX_IN_BATCH] if not chunk: continue placeholders = ",".join("?" * len(chunk)) q_res = await db_adapter.aquery( - f"SELECT id, filename, subfolder, source, root_id, stack_id FROM assets WHERE lower(filename) IN ({placeholders})", + f"SELECT id, filename, subfolder, source, root_id, job_id, stack_id FROM assets WHERE lower(filename) IN ({placeholders})", tuple(chunk), ) if not (q_res.ok and q_res.data): @@ -873,29 +885,19 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: asset_id = int(row.get("id") or 0) if not asset_id: continue - new_stack_id = job_stack_lookup.get(row_prompt_id) - old_stack_id = int(row.get("stack_id") or 0) - if old_stack_id and old_stack_id != new_stack_id: - prompt_affected_stack_ids.add(old_stack_id) + existing_job_id = row.get("job_id") + if _should_preserve_existing_job(existing_job_id, row_prompt_id): + logger.debug( + "Skipping filename prompt correlation overwrite for asset %s: existing job_id=%s candidate=%s", + asset_id, + existing_job_id, + row_prompt_id, + ) + continue await db_adapter.aexecute( - "UPDATE assets SET job_id = ?, stack_id = COALESCE(?, stack_id) WHERE id = ?", - (row_prompt_id, new_stack_id, asset_id), + "UPDATE assets SET job_id = ? WHERE id = ?", + (row_prompt_id, asset_id), ) - if new_stack_id: - prompt_affected_stack_ids.add(new_stack_id) - # Refresh counts for all stacks affected by prompt correlation - for stack_id in prompt_affected_stack_ids: - await db_adapter.aexecute( - "UPDATE asset_stacks SET " - "asset_count = (SELECT COUNT(*) FROM assets WHERE stack_id = ?), " - "cover_asset_id = COALESCE(" - " (SELECT id FROM assets WHERE stack_id = ? AND kind = 'image' ORDER BY size DESC, mtime DESC LIMIT 1), " - " (SELECT id FROM assets WHERE stack_id = ? ORDER BY mtime DESC LIMIT 1)" - "), " - "updated_at = CURRENT_TIMESTAMP " - "WHERE id = ?", - (stack_id, stack_id, stack_id, stack_id), - ) sibling_rows: list[dict[str, Any]] = [] if all_fnames: @@ -911,7 +913,7 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: ) if q_res.ok and q_res.data: sibling_rows.extend(q_res.data) - sibling_job_lookup: dict[tuple[str, str, str, str], tuple[str, Any]] = {} + sibling_job_lookup: dict[tuple[str, str, str, str], str] = {} for row in sibling_rows: row_job_id = str(row.get("job_id") or "").strip() if not row_job_id: @@ -923,9 +925,8 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: row.get("source") or "output", row.get("root_id") or "", ), - (row_job_id, row.get("stack_id")), + row_job_id, ) - affected_stack_ids: set[int] = set() for row in sibling_rows: filename_value = row.get("filename") or "" if _filename_ext_upper(filename_value) != "PNG": @@ -948,28 +949,19 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: asset_id = int(row.get("id") or 0) if not asset_id: continue - row_job_id, row_stack_id = sibling_job + row_job_id = sibling_job + existing_job_id = row.get("job_id") + if _should_preserve_existing_job(existing_job_id, row_job_id): + logger.debug( + "Skipping sibling prompt correlation overwrite for asset %s: existing job_id=%s candidate=%s", + asset_id, + existing_job_id, + row_job_id, + ) + continue await db_adapter.aexecute( - "UPDATE assets SET job_id = ?, stack_id = COALESCE(?, stack_id) WHERE id = ?", - (row_job_id, row_stack_id, asset_id), - ) - try: - stack_id_int = int(row_stack_id or 0) - except Exception: - stack_id_int = 0 - if stack_id_int > 0: - affected_stack_ids.add(stack_id_int) - for stack_id in affected_stack_ids: - await db_adapter.aexecute( - "UPDATE asset_stacks SET " - "asset_count = (SELECT COUNT(*) FROM assets WHERE stack_id = ?), " - "cover_asset_id = COALESCE(" - " (SELECT id FROM assets WHERE stack_id = ? AND kind = 'image' ORDER BY size DESC, mtime DESC LIMIT 1), " - " (SELECT id FROM assets WHERE stack_id = ? ORDER BY mtime DESC LIMIT 1)" - "), " - "updated_at = CURRENT_TIMESTAMP " - "WHERE id = ?", - (stack_id, stack_id, stack_id, stack_id), + "UPDATE assets SET job_id = ? WHERE id = ?", + (row_job_id, asset_id), ) except Exception as ex: logger.debug("Failed to assign prompt/job correlation: %s", ex) @@ -1029,10 +1021,13 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: if not item_id or item_id not in gen_time_by_id: continue payload = dict(item) + if not is_execution_grouping_enabled(): + payload.pop("job_id", None) + payload.pop("stack_id", None) if item_id in gen_time_by_id: payload["generation_time_ms"] = gen_time_by_id[item_id] row_prompt = str(item.get("job_id") or prompt_id or "") - if row_prompt: + if is_execution_grouping_enabled() and row_prompt: payload["job_id"] = row_prompt payloads_by_id[item_id] = payload except Exception: diff --git a/mjr_am_backend/routes/handlers/stacks.py b/mjr_am_backend/routes/handlers/stacks.py index d51a7fca..c6b822fd 100644 --- a/mjr_am_backend/routes/handlers/stacks.py +++ b/mjr_am_backend/routes/handlers/stacks.py @@ -13,6 +13,7 @@ Routes: """ from aiohttp import web +from mjr_am_backend.config import is_execution_grouping_enabled from mjr_am_backend.shared import Result, get_logger from ..core import ( @@ -203,6 +204,16 @@ def register_stacks_routes(routes: web.RouteTableDef) -> None: auth = _require_write_access(request) if not auth.ok: return _json_response(auth) + if not is_execution_grouping_enabled(): + return _json_response( + Result.Ok( + { + "disabled": True, + "targeted_job_id": None, + "stacks": [], + } + ) + ) services, error_result = await _require_services() if error_result is not None or not services: return _json_response(error_result or Result.Err("SERVICE_UNAVAILABLE", "Backend not ready"), status=503) diff --git a/mjr_am_backend/routes/handlers/vector_search.py b/mjr_am_backend/routes/handlers/vector_search.py index 525ebaa3..5464d524 100644 --- a/mjr_am_backend/routes/handlers/vector_search.py +++ b/mjr_am_backend/routes/handlers/vector_search.py @@ -21,6 +21,7 @@ from ...config import ( is_vector_search_enabled, ) from ...features.index.searcher import _build_filter_clauses +from ...features.index.vector_runtime import ensure_vector_runtime from ...shared import Result, get_logger from ..core import _json_response, _require_services, safe_error_message from ..core.security import _check_rate_limit @@ -330,22 +331,12 @@ def _require_vector_services(services: dict[str, Any]): vs = services.get("vector_searcher") if vs is None: - db = services.get("db") - if db is not None: - try: - from ...features.index.vector_searcher import VectorSearcher - from ...features.index.vector_service import VectorService - - vector_service = services.get("vector_service") - if vector_service is None: - vector_service = VectorService() - services["vector_service"] = vector_service - vs = VectorSearcher(db, vector_service) - services["vector_searcher"] = vs - logger.info("Vector services initialized lazily") - except Exception as exc: - logger.warning("Lazy vector service init failed: %s", exc) - vs = None + return None, _json_response( + Result.Err( + "SERVICE_UNAVAILABLE", + "Vector search is unavailable (dependencies/model not ready).", + ), + ) if vs is None: return None, _json_response( @@ -357,6 +348,36 @@ def _require_vector_services(services: dict[str, Any]): return vs, None +async def _require_vector_services_async(services: dict[str, Any]): + if not is_vector_search_enabled(): + return None, _json_response( + Result.Err( + "SERVICE_UNAVAILABLE", + "Vector search is disabled. Enable it in Majoor settings (AI toggle) or set MJR_AM_ENABLE_VECTOR_SEARCH=1.", + ), + ) + existing = services.get("vector_searcher") if isinstance(services, dict) else None + if existing is not None: + return existing, None + try: + _vector_service, vector_searcher = await ensure_vector_runtime( + services, + logger=logger, + reason="vector-routes", + ) + except Exception as exc: + logger.warning("Lazy vector service init failed: %s", exc) + vector_searcher = None + if vector_searcher is None: + return None, _json_response( + Result.Err( + "SERVICE_UNAVAILABLE", + "Vector search is unavailable (dependencies/model not ready).", + ), + ) + return vector_searcher, None + + async def _hydrate_vector_results( services: dict[str, Any], result: Result, @@ -471,7 +492,7 @@ def register_vector_search_routes(routes: web.RouteTableDef) -> None: return _json_response(err) services_dict = _services_dict(services) - searcher, verr = _require_vector_services(services_dict) + searcher, verr = await _require_vector_services_async(services_dict) if verr or searcher is None: return verr or _json_response( Result.Err("SERVICE_UNAVAILABLE", "Vector search is unavailable.") @@ -547,7 +568,7 @@ def register_vector_search_routes(routes: web.RouteTableDef) -> None: return _json_response(err) services_dict = _services_dict(services) - searcher, verr = _require_vector_services(services_dict) + searcher, verr = await _require_vector_services_async(services_dict) if verr or searcher is None: return verr or _json_response( Result.Err("SERVICE_UNAVAILABLE", "Vector search is unavailable.") @@ -607,7 +628,7 @@ def register_vector_search_routes(routes: web.RouteTableDef) -> None: return _json_response(err) services_dict = _services_dict(services) - searcher, verr = _require_vector_services(services_dict) + searcher, verr = await _require_vector_services_async(services_dict) if verr or searcher is None: return verr or _json_response( Result.Err("SERVICE_UNAVAILABLE", "Vector search is unavailable.") @@ -761,7 +782,7 @@ def register_vector_search_routes(routes: web.RouteTableDef) -> None: return _json_response(err) services_dict = _services_dict(services) - searcher, verr = _require_vector_services(services_dict) + searcher, verr = await _require_vector_services_async(services_dict) if verr or searcher is None: return verr or _json_response( Result.Err("SERVICE_UNAVAILABLE", "Vector search is unavailable.") @@ -837,7 +858,7 @@ def register_vector_search_routes(routes: web.RouteTableDef) -> None: return _json_response(err) services_dict = _services_dict(services) - searcher, verr = _require_vector_services(services_dict) + searcher, verr = await _require_vector_services_async(services_dict) if verr or searcher is None: return verr or _json_response( Result.Err("SERVICE_UNAVAILABLE", "Vector search is unavailable.") diff --git a/mjr_am_backend/routes/registry.py b/mjr_am_backend/routes/registry.py index 9e3650ca..c5aa326b 100644 --- a/mjr_am_backend/routes/registry.py +++ b/mjr_am_backend/routes/registry.py @@ -38,6 +38,7 @@ from .handlers import ( register_stacks_routes, register_vector_search_routes, register_vendor_routes, + register_plugin_routes, register_version_routes, register_viewer_routes, ) @@ -357,6 +358,13 @@ def register_all_routes() -> web.RouteTableDef: logger.info(" POST /mjr/am/vector/suggest-collections (Added)") except Exception as e: logger.error(f"Failed to register vector search routes: {e}") + try: + register_plugin_routes(routes) + logger.info(" GET /mjr/am/plugins/list (Added)") + logger.info(" POST /mjr/am/plugins/{name}/enable (Added)") + logger.info(" POST /mjr/am/plugins/reload (Added)") + except Exception as e: + logger.error(f"Failed to register plugin routes: {e}") try: register_hybrid_search_routes(routes) logger.info(" GET /mjr/am/search/hybrid (Added)") diff --git a/mjr_am_backend/settings.py b/mjr_am_backend/settings.py index cf2035f3..927a96d8 100644 --- a/mjr_am_backend/settings.py +++ b/mjr_am_backend/settings.py @@ -12,7 +12,12 @@ import time from collections.abc import Mapping from typing import Any -from .config import MEDIA_PROBE_BACKEND, OUTPUT_ROOT, is_vector_search_enabled +from .config import ( + MEDIA_PROBE_BACKEND, + OUTPUT_ROOT, + is_execution_grouping_enabled, + is_vector_search_enabled, +) from .shared import Result, get_logger from .utils import env_bool, parse_bool @@ -23,6 +28,7 @@ _OUTPUT_DIRECTORY_KEY = "output_directory_override" _METADATA_FALLBACK_IMAGE_KEY = "metadata_fallback_image" _METADATA_FALLBACK_MEDIA_KEY = "metadata_fallback_media" _VECTOR_SEARCH_ENABLED_KEY = "vector_search_enabled" +_EXECUTION_GROUPING_ENABLED_KEY = "execution_grouping_enabled" _HUGGINGFACE_TOKEN_KEY = "huggingface_token" _AI_VERBOSE_LOGS_KEY = "ai_verbose_logs" _SETTINGS_VERSION_KEY = "__settings_version" @@ -85,6 +91,7 @@ class AppSettings: self._default_metadata_fallback_image = True self._default_metadata_fallback_media = True self._default_vector_search_enabled = bool(is_vector_search_enabled()) + self._default_execution_grouping_enabled = bool(is_execution_grouping_enabled()) self._default_ai_verbose_logs = self._env_ai_verbose_logs_enabled() self._runtime_api_token: str = "" self._runtime_api_token_hash: str = "" @@ -752,6 +759,21 @@ class AppSettings: return None return parse_bool(cached, self._default_vector_search_enabled) + def _cached_execution_grouping_pref(self, current_version: int) -> bool | None: + cached = self._cache.get(_EXECUTION_GROUPING_ENABLED_KEY) + if cached is None: + return None + try: + ts = float(self._cache_at.get(_EXECUTION_GROUPING_ENABLED_KEY) or 0.0) + except Exception: + ts = 0.0 + cached_ver = int(self._cache_version.get(_EXECUTION_GROUPING_ENABLED_KEY) or 0) + if cached_ver != int(current_version or 0): + return None + if not ts or (time.monotonic() - ts) >= self._cache_ttl_s: + return None + return parse_bool(cached, self._default_execution_grouping_enabled) + async def set_vector_search_enabled(self, enabled: Any) -> Result[bool]: """Persist vector-search enable preference and apply runtime env vars.""" normalized = parse_bool(enabled, self._default_vector_search_enabled) @@ -772,6 +794,57 @@ class AppSettings: self._cache_version[_VECTOR_SEARCH_ENABLED_KEY] = current_version return Result.Ok(normalized) + async def get_execution_grouping_enabled(self) -> bool: + """Return persisted execution-grouping enable preference.""" + async with self._lock: + current_version = await self._get_settings_version() + cached = self._cached_execution_grouping_pref(current_version) + if cached is not None: + return cached + raw = await self._read_setting(_EXECUTION_GROUPING_ENABLED_KEY) + enabled = ( + parse_bool(raw, self._default_execution_grouping_enabled) + if raw is not None + else self._default_execution_grouping_enabled + ) + self._cache[_EXECUTION_GROUPING_ENABLED_KEY] = "1" if enabled else "0" + self._cache_at[_EXECUTION_GROUPING_ENABLED_KEY] = time.monotonic() + self._cache_version[_EXECUTION_GROUPING_ENABLED_KEY] = int(current_version or 0) + return enabled + + @staticmethod + def _set_execution_grouping_env_vars(enabled: bool) -> None: + normalized = "1" if parse_bool(enabled, True) else "0" + try: + os.environ["MJR_AM_EXECUTION_GROUPING_ENABLED"] = normalized + os.environ["MAJOOR_EXECUTION_GROUPING_ENABLED"] = normalized + except Exception: + pass + + async def set_execution_grouping_enabled(self, enabled: Any) -> Result[bool]: + """Persist execution-grouping enable preference and apply runtime env vars.""" + normalized = parse_bool(enabled, self._default_execution_grouping_enabled) + async with self._lock: + res = await self._write_setting( + _EXECUTION_GROUPING_ENABLED_KEY, "1" if normalized else "0" + ) + if not res.ok: + return Result.Err( + "DB_ERROR", res.error or "Failed to persist execution_grouping_enabled" + ) + self._set_execution_grouping_env_vars(normalized) + bump = await self._bump_settings_version_locked() + if not bump.ok: + try: + logger.warning("Failed to bump settings version: %s", bump.error) + except Exception: + pass + current_version = int(bump.data or await self._get_settings_version() or 0) + self._cache[_EXECUTION_GROUPING_ENABLED_KEY] = "1" if normalized else "0" + self._cache_at[_EXECUTION_GROUPING_ENABLED_KEY] = time.monotonic() + self._cache_version[_EXECUTION_GROUPING_ENABLED_KEY] = current_version + return Result.Ok(normalized) + async def get_huggingface_token_info(self) -> dict[str, Any]: async with self._lock: token = self._env_huggingface_token() @@ -1092,6 +1165,29 @@ class AppSettings: except Exception as exc: logger.warning("Failed to restore vector search setting on startup: %s", exc) + async def apply_execution_grouping_override_on_startup(self) -> None: + """Restore execution-grouping enabled preference into environment on startup.""" + try: + async with self._lock: + raw = await self._read_setting(_EXECUTION_GROUPING_ENABLED_KEY) + enabled = ( + parse_bool(raw, self._default_execution_grouping_enabled) + if raw is not None + else self._default_execution_grouping_enabled + ) + self._set_execution_grouping_env_vars(enabled) + self._cache[_EXECUTION_GROUPING_ENABLED_KEY] = "1" if enabled else "0" + self._cache_at[_EXECUTION_GROUPING_ENABLED_KEY] = time.monotonic() + self._cache_version[_EXECUTION_GROUPING_ENABLED_KEY] = int( + await self._get_settings_version() or 0 + ) + logger.info( + "Restored execution grouping setting on startup: %s", + "enabled" if enabled else "disabled", + ) + except Exception as exc: + logger.warning("Failed to restore execution grouping setting on startup: %s", exc) + async def apply_huggingface_token_on_startup(self) -> None: try: async with self._lock: diff --git a/tests/features/test_scan_routes_more.py b/tests/features/test_scan_routes_more.py index d9a6de83..d8be1f4f 100644 --- a/tests/features/test_scan_routes_more.py +++ b/tests/features/test_scan_routes_more.py @@ -17,6 +17,19 @@ def _app(): return app +def test_should_preserve_existing_job_blocks_conflicting_job_ids() -> None: + assert scan_mod._should_preserve_existing_job("core-job-1", "mjr-job-2") is True + assert scan_mod._should_preserve_existing_job("core-job-1", "core-job-1") is False + assert scan_mod._should_preserve_existing_job("", "mjr-job-2") is False + + +def test_resolve_stack_assignment_prefers_candidate_then_existing() -> None: + assert scan_mod._resolve_stack_assignment(12, 34) == 34 + assert scan_mod._resolve_stack_assignment(12, None) == 12 + assert scan_mod._resolve_stack_assignment(None, "9") == 9 + assert scan_mod._resolve_stack_assignment(None, None) is None + + @pytest.mark.asyncio async def test_index_files_rejects_invalid_files_list(monkeypatch) -> None: async def _require_services():