diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index 6769dcd5..e0534ab2 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -48,7 +48,10 @@ jobs: pip install .[dev] - name: Run canonical quality gate - run: python scripts/run_quality_gate.py --python-only --skip-tests + shell: bash + run: | + set -o pipefail + python scripts/run_quality_gate.py --python-only --skip-tests 2>&1 | tee quality-gate.log - name: Install JS dependencies for audit run: npm ci --ignore-scripts diff --git a/js/entry.js b/js/entry.js index 7a918a21..ee2874fd 100644 --- a/js/entry.js +++ b/js/entry.js @@ -299,6 +299,8 @@ const INDEX_RETRY_MAX_DELAY_MS = 15_000; 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 }); +let _stackFinalizeTimer = null; +let _stackFinalizeInFlight = null; function rememberExecutionStart(promptId, timestampMs) { if (!promptId) return; @@ -364,6 +366,26 @@ function scheduleGenerationIndex(files, attempt = 1, meta = {}) { .catch((error) => reportError(error, "entry.executed.index")); } +function scheduleFinalizeExecutionStacks(delayMs = 900) { + try { + if (_stackFinalizeTimer) clearTimeout(_stackFinalizeTimer); + } catch (e) { console.debug?.(e); } + _stackFinalizeTimer = setTimeout(() => { + _stackFinalizeTimer = null; + if (_stackFinalizeInFlight) return; + _stackFinalizeInFlight = post(ENDPOINTS.STACKS_AUTO_STACK, { mode: "job_id" }) + .then((res) => { + if (!res?.ok) { + reportError(new Error(String(res?.error || "Auto stack failed")), "entry.execution_end.auto_stack"); + } + }) + .catch((error) => reportError(error, "entry.execution_end.auto_stack")) + .finally(() => { + _stackFinalizeInFlight = null; + }); + }, Math.max(0, Number(delayMs) || 0)); +} + app.registerExtension({ name: "Majoor.AssetsManager", settings: buildMajoorSettings(app), @@ -519,6 +541,9 @@ app.registerExtension({ progress_value: null, progress_max: null, }); + if (String(event?.type || "") === "execution_success") { + scheduleFinalizeExecutionStacks(); + } } catch (error) { reportError(error, "entry.execution_end"); } diff --git a/js/features/bottomPanel/GeneratedFeedTab.js b/js/features/bottomPanel/GeneratedFeedTab.js index a230e39b..585acb63 100644 --- a/js/features/bottomPanel/GeneratedFeedTab.js +++ b/js/features/bottomPanel/GeneratedFeedTab.js @@ -106,8 +106,6 @@ const FEED_STATE = { function _getAssetGroupKey(asset) { const stackId = String(asset?.stack_id || "").trim(); if (stackId) return `stack:${stackId}`; - const jobId = String(asset?.job_id || "").trim(); - if (jobId) return `job:${jobId}`; return _assetKey(asset); } diff --git a/js/features/grid/StackGroupCards.js b/js/features/grid/StackGroupCards.js index 8c3951df..8d61da49 100644 --- a/js/features/grid/StackGroupCards.js +++ b/js/features/grid/StackGroupCards.js @@ -1,7 +1,7 @@ import { createAssetCard } from "../../components/Card.js"; import { getViewerInstance } from "../../components/Viewer.js"; import { get } from "../../api/client.js"; -import { buildStackByJobURL, buildStackMembersURL } from "../../api/endpoints.js"; +import { buildStackMembersURL } from "../../api/endpoints.js"; import { createPopoverManager } from "../panel/views/popoverManager.js"; function _isGroupEnabled(gridContainer) { @@ -11,8 +11,6 @@ function _isGroupEnabled(gridContainer) { function _getGroupKey(asset) { const stackId = String(asset?.stack_id || "").trim(); if (stackId) return `stack:${stackId}`; - const jobId = String(asset?.job_id || "").trim(); - if (jobId) return `job:${jobId}`; return ""; } @@ -40,15 +38,6 @@ async function _fetchGroupMembers(asset) { if (res?.ok && Array.isArray(res?.data)) return res.data; throw new Error(String(res?.error || "Failed to load stack members")); } - - const jobId = String(asset?.job_id || "").trim(); - if (!jobId) return [asset].filter(Boolean); - - const stackRes = await get(buildStackByJobURL(jobId), { timeoutMs: 30_000 }); - if (stackRes?.ok && stackRes?.data?.id != null) { - const membersRes = await get(buildStackMembersURL(stackRes.data.id), { timeoutMs: 30_000 }); - if (membersRes?.ok && Array.isArray(membersRes?.data)) return membersRes.data; - } return [asset].filter(Boolean); } diff --git a/mjr_am_backend/adapters/db/sqlite_facade.py b/mjr_am_backend/adapters/db/sqlite_facade.py index cc16f093..486a008c 100644 --- a/mjr_am_backend/adapters/db/sqlite_facade.py +++ b/mjr_am_backend/adapters/db/sqlite_facade.py @@ -1734,6 +1734,10 @@ class Sqlite: def _schedule_delete_on_reboot(path: Path) -> bool: return tx_schedule_delete_on_reboot(path) + def _has_async_runtime_state(self) -> bool: + """Return True when the instance has the async DB runtime needed for schema queries.""" + return all(hasattr(self, attr) for attr in ("_loop_thread", "_pool", "_tx_conns")) + async def areset(self) -> Result[bool]: """ Aggressive reset: @@ -1774,9 +1778,12 @@ class Sqlite: if not schema_res.ok: return schema_res - columns_res = await ensure_columns_exist(self) - if not columns_res.ok: - return columns_res + if self._has_async_runtime_state(): + columns_res = await ensure_columns_exist(self) + if not columns_res.ok: + return columns_res + else: + logger.debug("Skipping ensure_columns_exist during reset for partially constructed Sqlite instance") indexes_res = await ensure_indexes_and_triggers(self) if not indexes_res.ok: diff --git a/mjr_am_backend/features/index/searcher.py b/mjr_am_backend/features/index/searcher.py index 1283073e..d3f81929 100644 --- a/mjr_am_backend/features/index/searcher.py +++ b/mjr_am_backend/features/index/searcher.py @@ -470,12 +470,9 @@ def _group_assets_by_stack(assets: list[dict[str, Any]]) -> list[dict[str, Any]] if not isinstance(asset, dict): continue stack_id = asset.get("stack_id") - job_id = str(asset.get("job_id") or "").strip() stack_id_int = _safe_positive_int(stack_id) if stack_id_int is not None: key = f"stack:{stack_id_int}" - elif job_id: - key = f"job:{job_id}" else: key = f"asset:{_safe_positive_int(asset.get('id')) or 0}" if key in seen: diff --git a/mjr_am_backend/features/index/service.py b/mjr_am_backend/features/index/service.py index a6aaad42..9d2050d9 100644 --- a/mjr_am_backend/features/index/service.py +++ b/mjr_am_backend/features/index/service.py @@ -17,6 +17,7 @@ from typing import TYPE_CHECKING, Any from mjr_am_shared.scan_throttle import mark_directory_indexed from ...adapters.db.sqlite import Sqlite +from ...config import BATCH_ASSET_PUSH_LIMIT, is_vector_search_enabled from ...shared import Result, get_logger from ...utils import sanitize_for_json from ..metadata import MetadataService @@ -26,7 +27,6 @@ from .scan_batch_utils import compute_state_hash, normalize_filepath_str from .scanner import IndexScanner from .searcher import IndexSearcher from .updater import AssetUpdater -from ...config import BATCH_ASSET_PUSH_LIMIT, is_vector_search_enabled if TYPE_CHECKING: from .vector_searcher import VectorSearcher diff --git a/mjr_am_backend/routes/handlers/scan.py b/mjr_am_backend/routes/handlers/scan.py index 26bbe205..049a6bab 100644 --- a/mjr_am_backend/routes/handlers/scan.py +++ b/mjr_am_backend/routes/handlers/scan.py @@ -2,9 +2,6 @@ Directory scanning and file indexing endpoints. """ import asyncio -from contextlib import suppress -import os -import shutil import time from pathlib import Path from typing import Any @@ -30,29 +27,27 @@ from mjr_am_backend.config import ( TO_THREAD_TIMEOUT_S, ) from mjr_am_backend.custom_roots import resolve_custom_root +from mjr_am_backend.features.index.metadata_helpers import MetadataHelpers from mjr_am_backend.features.index.scan_route_service import ( build_reset_scan_summary, - parse_scan_request_payload, parse_reset_index_payload, + parse_scan_request_payload, prepare_scan_route_context, resolve_reset_index_target_roots, + run_all_scope_scan, run_reset_clear_phase, run_reset_reindex, - run_all_scope_scan, ) -from mjr_am_backend.features.index.metadata_helpers import MetadataHelpers from mjr_am_backend.shared import Result, get_logger, sanitize_error_message from mjr_am_backend.utils import parse_bool, sanitize_for_json from ..core import ( - audit_log_write, _check_rate_limit, _csrf_error, _is_path_allowed, _is_path_allowed_custom, _is_within_root, _json_response, - _normalize_path, _read_json, _require_operation_enabled, _require_services, @@ -62,80 +57,49 @@ from ..core import ( safe_error_message, ) from .db_maintenance import ( + _restart_watcher_if_needed, is_db_maintenance_active, is_vector_backfill_active, request_vector_backfill_priority_window, set_db_maintenance_active, - _restart_watcher_if_needed, -) -from .scan_staging import ( - StageDestination, - _resolve_stage_destination, - register_staging_routes, -) -from mjr_am_backend.features.watcher_settings import get_watcher_settings, update_watcher_settings -from mjr_am_backend.features.index.watcher_scope import build_watch_paths - -from .scan_upload import ( - _index_uploaded_input_best_effort, - register_upload_routes, -) -from .scan_watcher import ( - _build_watcher_callbacks, - _delay_for_recent_generated_marker, - _filter_recent_generated_files, - _start_watcher_for_scope, - _stop_watcher_if_running, - _watcher_scope_config, - register_watcher_routes, ) from .filesystem import _kickoff_background_scan -from .scan_consistency import ( - _collect_missing_asset_rows, - _delete_missing_asset_rows, - _missing_asset_row, - _query_consistency_sample, - _run_consistency_check, -) +from .scan_consistency import _run_consistency_check from .scan_helpers import ( _DB_CONSISTENCY_COOLDOWN_SECONDS, - _DB_CONSISTENCY_SAMPLE, - _FILE_COMPARE_CHUNK_BYTES, - _INDEX_SEMAPHORE, - _MAX_CONCURRENT_INDEX, - _MAX_FILENAME_LEN, - _MAX_RENAME_ATTEMPTS, - _MAX_UPLOAD_SIZE, - _UPLOAD_READ_CHUNK_BYTES, - _add_env_upload_extensions, - _allowed_upload_exts, - _cleanup_temp_upload_file, - _default_allowed_upload_exts, _emit_maintenance_status, - _file_sizes_equal, - _files_equal_content, - _files_exist_and_are_regular, _is_db_malformed_result, - _normalize_upload_extension, - _paths_are_same_file, - _read_upload_file_field, - _refresh_watcher_runtime_settings, - _resolve_input_directory, _resolve_scan_root, _runtime_output_root, - _schedule_index_task, - _stream_content_equal, - _unique_upload_destination, - _upload_skip_index, - _validate_upload_filename, - _watcher_settings_from_body, - _write_multipart_file_atomic, - _write_upload_chunks, +) +from .scan_staging import register_staging_routes +from .scan_upload import register_upload_routes +from .scan_watcher import ( + _stop_watcher_if_running, + register_watcher_routes, ) logger = get_logger(__name__) +def _get_prompt_server(): + try: + from .. import registry as registry_mod + + return getattr(registry_mod, "PromptServer", None) + except Exception: + return None + + +def _send_prompt_event(event: str, payload: Any) -> None: + try: + prompt_server = _get_prompt_server() + if prompt_server is not None: + prompt_server.instance.send_sync(event, payload) + except Exception: + pass + + def _scan_enqueued_from_kickoff(result: Any) -> bool: # Backward-compatible for monkeypatched tests that return None. if result is None: @@ -615,7 +579,7 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: return _json_response(result) try: - PromptServer.instance.send_sync("mjr.scan.progress", sanitize_for_json({ + _send_prompt_event("mjr.scan.progress", sanitize_for_json({ "status": "started", "scope": "index-files", "origin": origin, @@ -624,7 +588,7 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: })) for item in files: if isinstance(item, dict): - PromptServer.instance.send_sync("mjr.asset.indexing", sanitize_for_json({ + _send_prompt_event("mjr.asset.indexing", sanitize_for_json({ "filename": item.get("filename") or item.get("name"), "subfolder": item.get("subfolder") or "", "type": item.get("type") or "output", @@ -770,11 +734,8 @@ 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) - assigned_stack_by_id: dict[int, int] = {} if prompt_lookup: try: - from ...features.stacks.service import StacksService - stack_service = StacksService(db_adapter) for start in range(0, len(fnames_for_prompt), _MAX_IN_BATCH): chunk = fnames_for_prompt[start:start + _MAX_IN_BATCH] if not chunk: @@ -803,18 +764,14 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: "UPDATE assets SET job_id = ? WHERE id = ?", (row_prompt_id, asset_id), ) - stack_res = await stack_service.assign_asset_by_job_id(asset_id, row_prompt_id) - if stack_res.ok and stack_res.data is not None: - assigned_stack_by_id[asset_id] = int(stack_res.data) except Exception as ex: - logger.debug("Failed to assign prompt/job/stack correlation: %s", ex) + logger.debug("Failed to assign prompt/job correlation: %s", ex) - if gen_time_by_id or assigned_stack_by_id: + if gen_time_by_id: try: - from ..registry import PromptServer payloads_by_id: dict[int, dict[str, Any]] = {} index_service = svc.get("index") if isinstance(svc, dict) else None - target_ids = list(dict.fromkeys(list(gen_time_by_id.keys()) + list(assigned_stack_by_id.keys()))) + target_ids = list(dict.fromkeys(list(gen_time_by_id.keys()))) if index_service and hasattr(index_service, "get_assets_batch"): try: batch_res = await index_service.get_assets_batch(target_ids) @@ -831,8 +788,6 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: payload = dict(item) if item_id in gen_time_by_id: payload["generation_time_ms"] = gen_time_by_id[item_id] - if item_id in assigned_stack_by_id: - payload["stack_id"] = assigned_stack_by_id[item_id] row_prompt = str(item.get("job_id") or prompt_id or "") if row_prompt: payload["job_id"] = row_prompt @@ -846,12 +801,12 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: payload = payloads_by_id.get(int(item_id)) if not isinstance(payload, dict): continue - PromptServer.instance.send_sync( + _send_prompt_event( "mjr-asset-updated", sanitize_for_json(payload), ) try: - PromptServer.instance.send_sync( + _send_prompt_event( "mjr.asset.indexed", sanitize_for_json(payload), ) @@ -877,14 +832,14 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: logger.debug("Failed to attach scan stats: %s", exc) try: - PromptServer.instance.send_sync("mjr.scan.progress", sanitize_for_json({ + _send_prompt_event("mjr.scan.progress", sanitize_for_json({ "status": "completed", "scope": "index-files", "origin": origin, "prompt_id": prompt_id or None, "stats": total_stats, })) - PromptServer.instance.send_sync("mjr.runtime.status", sanitize_for_json({ + _send_prompt_event("mjr.runtime.status", sanitize_for_json({ "active_prompt_id": prompt_id or None, "origin": origin, "stats": total_stats, diff --git a/tests/features/test_index_service.py b/tests/features/test_index_service.py index a8e1b4ca..d90c86e0 100644 --- a/tests/features/test_index_service.py +++ b/tests/features/test_index_service.py @@ -2,7 +2,6 @@ from pathlib import Path from unittest.mock import AsyncMock import pytest - from mjr_am_backend.features.index import service as isvc from mjr_am_backend.shared import Result