Corrige les conflits de merge dans health.py : fusionne les fonctions et supprime les marqueurs de conflit.

This commit is contained in:
MajoorWaldi
2026-03-29 15:37:11 +02:00
35 changed files with 2197 additions and 1414 deletions
+4
View File
@@ -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
+2 -2
View File
@@ -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*
---
+8
View File
@@ -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);
}
+1
View File
@@ -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",
+30 -1
View File
@@ -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 {
+1
View File
@@ -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,
+3 -9
View File
@@ -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;
+33
View File
@@ -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);
}
}
+45 -1
View File
@@ -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: ", "")),
+137 -1283
View File
File diff suppressed because it is too large Load Diff
+22 -1
View File
@@ -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) {
+2
View File
@@ -1,7 +1,9 @@
/** GridView facade preserving public exports. */
export {
buildGridSnapshotKey,
createGridContainer,
hydrateGridFromSnapshot,
loadAssets,
prepareGridForScopeSwitch,
loadAssetsFromList,
+81
View File
@@ -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<id, asset>, 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);
}
}
+20
View File
@@ -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();
@@ -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).
+50
View File
@@ -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()",
);
}
@@ -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);
}
}
+151
View File
@@ -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);
}
}),
];
}
@@ -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");
}
@@ -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,
};
}
+28 -2
View File
@@ -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",
)
+11 -13
View File
@@ -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"] = {}
+30 -3
View File
@@ -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(
@@ -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
+33
View File
@@ -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:
@@ -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",
@@ -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",
]
+70
View File
@@ -188,12 +188,21 @@ def _extract_vector_search_payload(body: dict) -> object | None:
return None
def _str_token_from_body(source: dict, *keys: str) -> str | None:
for key in keys:
if key in source:
return str(source.get(key) or "")
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:
result = _str_token_from_body(body, "token", "huggingface_token")
@@ -817,6 +826,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()
+67
View File
@@ -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}"))
+71 -76
View File
@@ -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:
+11
View File
@@ -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)
+42 -21
View File
@@ -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
@@ -419,22 +420,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(
@@ -446,6 +437,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,
@@ -517,7 +538,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.")
@@ -593,7 +614,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.")
@@ -653,7 +674,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.")
@@ -807,7 +828,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.")
@@ -883,7 +904,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.")
+8
View File
@@ -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)")
+97 -1
View File
@@ -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()
@@ -1087,6 +1160,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:
+13
View File
@@ -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():