"""
Background scheduler for resumable document folder scan jobs.

Per file_path schedule (file_schedules table): 30-minute cycle by default —
20 minutes of batch processing (active), then 10 minutes rest, repeating until
the job completes, fails, or is cancelled.
"""
from __future__ import annotations

import logging
import os
import sys
import threading
import time
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Set, Tuple

import json
from sqlalchemy import text
from app.models.file import FileSchedule
from app.services.file_schedule_service import (
    FileScheduleService,
    _ACTIVE_FILE_SCHEDULE_STATUSES,
)
from app.utils.file_schedule_utils import (
    advance_file_schedule_if_due,
    build_folder_scan_step_user_message,
    ensure_file_schedule_row,
    folder_scan_schedule_seconds,
    remaining_rest_after_upload_overflow,
    scheduler_message_from_schedule_row,
    scheduler_status_message,
    seconds_until_phase_end,
)

_file_schedule_svc = FileScheduleService()

_logger = logging.getLogger(__name__)

_SCHEDULE_LOCK = threading.Lock()
_SCHEDULED: Dict[str, Dict[str, Any]] = {}
_CANCEL_EVENTS: Dict[str, threading.Event] = {}
_STOP_REQUESTED_PATHS: Set[Tuple[str, str]] = set()
# Tracks jobs currently ingesting one file (used for deferred in-memory cleanup only).
_INGEST_IN_PROGRESS: Set[str] = set()
_HTTP_UPLOAD_REFS: Dict[Tuple[str, str], int] = {}
_PENDING_STOPS: Dict[Tuple[str, str], Dict[str, Any]] = {}

_GLOBAL_THREAD_LOCK = threading.Lock()
_GLOBAL_THREAD: Optional[threading.Thread] = None
_GLOBAL_CANCEL_EVENT = threading.Event()
_PATH_SCHEDULE_ENQUEUE_LOCK = threading.Lock()

_ACTIVE_SCHEDULER_STATUSES = frozenset(
    {"starting", "running", "sleeping", "resting", "cancelling"}
)

# In-memory states that block scheduling the same path again (excludes ``cancelling``).
_UPLOAD_BLOCKING_SCHEDULER_STATUSES = frozenset(
    {"starting", "running", "sleeping", "resting"}
)

_INVALID_JOB_ID_LITERALS = frozenset(
    {"string", "null", "none", "undefined", "job_id", "uuid", "id"}
)


def _now_iso() -> str:
    return datetime.now().isoformat()


def _normalize_scan_path(file_path: str) -> str:
    return os.path.normpath(os.path.abspath(str(file_path or "").strip()))


def _normalize_job_id_for_stop(job_id: Optional[str]) -> str:
    """Ignore Swagger placeholder values like ``string``."""
    jid = str(job_id or "").strip()
    if not jid or jid.lower() in _INVALID_JOB_ID_LITERALS:
        return ""
    return jid


def _mark_path_stop_requested(agent_id: str, file_path: str) -> None:
    path_norm = _normalize_scan_path(file_path)
    aid = str(agent_id or "").strip()
    if not path_norm or not aid:
        return
    with _SCHEDULE_LOCK:
        _STOP_REQUESTED_PATHS.add((aid, path_norm))


def _clear_path_stop_requested(agent_id: str, file_path: str) -> None:
    """Allow a fresh upload/schedule on this path after a prior stop or completion."""
    path_norm = _normalize_scan_path(file_path)
    aid = str(agent_id or "").strip()
    if not path_norm or not aid:
        return
    with _SCHEDULE_LOCK:
        _STOP_REQUESTED_PATHS.discard((aid, path_norm))
        related = [
            key
            for key in _STOP_REQUESTED_PATHS
            if key[0] == aid
            and (
                _paths_related(path_norm, key[1])
                or _paths_related(key[1], path_norm)
            )
        ]
        for key in related:
            _STOP_REQUESTED_PATHS.discard(key)


def _is_path_stop_requested_unlocked(agent_id: str, file_path: str) -> bool:
    """Caller must hold ``_SCHEDULE_LOCK``."""
    path_norm = _normalize_scan_path(file_path)
    aid = str(agent_id or "").strip()
    if not path_norm or not aid:
        return False
    if (aid, path_norm) in _STOP_REQUESTED_PATHS:
        return True
    return any(
        _paths_related(path_norm, stopped_path)
        for stopped_aid, stopped_path in _STOP_REQUESTED_PATHS
        if stopped_aid == aid
    )


def _is_path_stop_requested(agent_id: str, file_path: str) -> bool:
    with _SCHEDULE_LOCK:
        return _is_path_stop_requested_unlocked(agent_id, file_path)


def _path_key(agent_id: str, file_path: str) -> Tuple[str, str]:
    return (str(agent_id or "").strip(), _normalize_scan_path(file_path))


def register_folder_http_upload(agent_id: str, file_path: str) -> None:
    """Mark an in-flight upload-folder HTTP request (multipart save to disk)."""
    key = _path_key(agent_id, file_path)
    if not key[0] or not key[1]:
        return
    with _SCHEDULE_LOCK:
        _HTTP_UPLOAD_REFS[key] = int(_HTTP_UPLOAD_REFS.get(key, 0)) + 1


def unregister_folder_http_upload(agent_id: str, file_path: str) -> None:
    """End upload-folder HTTP tracking; apply a deferred stop-schedule if one was queued."""
    key = _path_key(agent_id, file_path)
    if not key[0] or not key[1]:
        return
    pending: Optional[Dict[str, Any]] = None
    with _SCHEDULE_LOCK:
        refs = int(_HTTP_UPLOAD_REFS.get(key, 0)) - 1
        if refs <= 0:
            _HTTP_UPLOAD_REFS.pop(key, None)
            pending = _PENDING_STOPS.pop(key, None)
        else:
            _HTTP_UPLOAD_REFS[key] = refs
    if pending:
        stop_folder_upload_scheduler(
            pending["agent_id"],
            job_id=pending.get("job_id"),
            file_path=pending.get("file_path"),
        )


def is_folder_http_upload_in_progress(agent_id: str, file_path: str) -> bool:
    key = _path_key(agent_id, file_path)
    if not key[0] or not key[1]:
        return False
    with _SCHEDULE_LOCK:
        return int(_HTTP_UPLOAD_REFS.get(key, 0)) > 0


def folder_scan_ingest_begin(job_id: str) -> None:
    """Scheduler is ingesting one file; defer mid-file cancellation."""
    jid = str(job_id or "").strip()
    if not jid:
        return
    with _SCHEDULE_LOCK:
        _INGEST_IN_PROGRESS.add(jid)


def folder_scan_ingest_end(job_id: str) -> None:
    """Finished ingesting one file; cancellation may take effect before the next file."""
    jid = str(job_id or "").strip()
    if not jid:
        return
    with _SCHEDULE_LOCK:
        _INGEST_IN_PROGRESS.discard(jid)


def find_scheduled_job_for_file_path(
    file_path: str,
    agent_id: str,
) -> Optional[Dict[str, Any]]:
    """In-memory scheduler entry for this path+agent (exact or ancestor/descendant)."""
    path_norm = _normalize_scan_path(file_path)
    aid = str(agent_id or "").strip()
    if not path_norm or not aid:
        return None
    with _SCHEDULE_LOCK:
        for entry in _SCHEDULED.values():
            if entry.get("agent_id") != aid:
                continue
            entry_path = str(entry.get("file_path") or "")
            if not _paths_related(entry_path, path_norm):
                continue
            return {k: v for k, v in entry.items() if k != "thread"}
    return None


def _collect_job_ids_to_stop(
    agent_id: str,
    *,
    job_id: Optional[str] = None,
    file_path: Optional[str] = None,
) -> List[str]:
    """Resolve in-memory and DB job ids that match agent + optional path/job filter."""
    aid = str(agent_id or "").strip()
    jid_filter = _normalize_job_id_for_stop(job_id)
    path_norm = _normalize_scan_path(file_path) if file_path else ""
    found: List[str] = []
    seen: Set[str] = set()

    def _add(candidate: Optional[str]) -> None:
        c = str(candidate or "").strip()
        if not c or c in seen:
            return
        seen.add(c)
        found.append(c)

    with _SCHEDULE_LOCK:
        for jid, entry in _SCHEDULED.items():
            if entry.get("agent_id") != aid:
                continue
            entry_path = str(entry.get("file_path") or "")
            if jid_filter and jid == jid_filter:
                _add(jid)
            elif path_norm and _paths_related(entry_path, path_norm):
                _add(jid)
            elif not jid_filter and not path_norm:
                _add(jid)

    if path_norm:
        from app.services.document_db_service import DocumentDbService
        from app.services.document_service import FILE_SYS_METADATA_DB

        svc = DocumentDbService(db_file=FILE_SYS_METADATA_DB)
        try:
            active = svc.get_active_folder_scan_job_for_root(path_norm)
            if active and active.get("id"):
                _add(str(active["id"]))
            elif jid_filter:
                row = svc.get_folder_scan_job(jid_filter)
                if row and row.get("id"):
                    _add(str(row["id"]))
        finally:
            svc.close()

    if jid_filter and jid_filter not in seen:
        from app.services.document_db_service import DocumentDbService
        from app.services.document_service import FILE_SYS_METADATA_DB

        svc = DocumentDbService(db_file=FILE_SYS_METADATA_DB)
        try:
            row = svc.get_folder_scan_job(jid_filter)
            if row and row.get("id"):
                _add(str(row["id"]))
        finally:
            svc.close()

    if jid_filter and jid_filter not in seen:
        with _SCHEDULE_LOCK:
            if jid_filter in _SCHEDULED:
                _add(jid_filter)

    return found


def is_file_path_scheduler_running(file_path: str, agent_id: str) -> bool:
    existing = find_scheduled_job_for_file_path(file_path, agent_id)
    if not existing:
        return False
    return (existing.get("status") or "").lower() in _ACTIVE_SCHEDULER_STATUSES


def announce_scheduler_remaining_on_file_upload(
    file_path: str,
    agent_id: str,
    filename: str,
) -> Optional[str]:
    """
    After each file in a folder scan batch: refresh in-memory scheduler_message
    and log countdown from the active window (e.g. 20 minutes → remaining).
    """
    path_norm = _normalize_scan_path(file_path)
    aid = str(agent_id or "").strip()
    if not path_norm or not aid:
        return None
    job = find_scheduled_job_for_file_path(path_norm, aid)
    if not job or not job.get("schedule_enabled", True):
        return None
    if (job.get("status") or "").lower() not in _ACTIVE_SCHEDULER_STATUSES:
        return None
    jid = str(job.get("job_id") or "")
    if jid and _get_folder_scan_job_status(jid) == "completed":
        return None
    row = _file_schedule_svc.get_file_schedule(file_path=path_norm, agent_id=aid)
    if not row or not int(row.get("schedule_enabled") or 0):
        return None
    phase = (row.get("phase") or "active").lower()
    if phase != "active":
        return None
    msg = scheduler_message_from_schedule_row(row)
    job_id = str(job.get("job_id") or "")
    with _SCHEDULE_LOCK:
        entry = _SCHEDULED.get(job_id)
        if entry:
            entry["scheduler_message"] = msg
    line = f"[SCHEDULER] {filename}: {msg}"
    print(line, flush=True)
    print(line, file=sys.stderr, flush=True)
    return msg


def _announce_scheduler_phase(
    job_id: str,
    *,
    phase: str,
    rest_minutes: int,
    active_minutes: int,
) -> None:
    """Log phase message only when active↔rest changes (not every batch)."""
    phase_norm = (phase or "active").lower()
    with _SCHEDULE_LOCK:
        entry = _SCHEDULED.get(job_id)
        if not entry:
            return
        if entry.get("announced_phase") == phase_norm:
            return
        entry["announced_phase"] = phase_norm
        row = _file_schedule_svc.get_file_schedule(
            file_path=str(entry.get("file_path") or ""),
            agent_id=str(entry.get("agent_id") or ""),
        )
        if row and int(row.get("schedule_enabled") or 0):
            entry["scheduler_message"] = scheduler_message_from_schedule_row(row)
        else:
            cycle_s, active_s, rest_s = folder_scan_schedule_seconds()
            default_remaining = rest_s if phase_norm == "rest" else active_s
            entry["scheduler_message"] = scheduler_status_message(
                phase=phase_norm,
                rest_minutes=rest_minutes,
                active_minutes=active_minutes,
                seconds_remaining=default_remaining,
            )
    msg = entry["scheduler_message"]
    print(msg, flush=True)
    print(msg, file=sys.stderr, flush=True)


def _stop_file_path_schedule(
    file_path: str,
    agent_id: str,
    *,
    job_id: Optional[str] = None,
) -> None:
    _file_schedule_svc.disable_file_schedule(
        file_path=file_path,
        agent_id=agent_id,
        job_id=job_id,
    )


def _get_folder_scan_job_status(job_id: str) -> Optional[str]:
    """Return folder_scan_jobs.status for ``job_id``, or None if missing."""
    from app.services.document_db_service import DocumentDbService
    from app.services.document_service import FILE_SYS_METADATA_DB

    jid = str(job_id or "").strip()
    if not jid:
        return None
    try:
        svc = DocumentDbService(db_file=FILE_SYS_METADATA_DB)
        try:
            row = svc.get_folder_scan_job(jid)
        finally:
            svc.close()
        if not row:
            return None
        return str(row.get("status") or "").lower()
    except Exception:
        return None


def _finalize_folder_scan_scheduler_job(
    job_id: str,
    agent_id: str,
    file_path: str,
    *,
    status: str = "completed",
    last_error: Optional[str] = None,
    cancel_ev: Optional[threading.Event] = None,
) -> None:
    """
    Tear down in-memory scheduler and disable per-path scheduling.

    Called when all files are uploaded (completed), on failure, or on user stop.
    """
    jid = str(job_id or "").strip()
    aid = str(agent_id or "").strip()
    path_norm = _normalize_scan_path(file_path)
    status_norm = (status or "completed").lower()
    if status_norm not in ("completed", "failed", "cancelled"):
        status_norm = "completed"

    if cancel_ev is not None:
        cancel_ev.set()
    if path_norm and aid:
        if status_norm == "completed":
            _clear_path_stop_requested(aid, path_norm)
        elif status_norm == "cancelled":
            _clear_path_stop_requested(aid, path_norm)
            _stop_file_path_schedule(path_norm, aid, job_id=jid or None)
        elif status_norm == "failed" and jid:
            _stop_file_path_schedule(path_norm, aid, job_id=jid)

    if jid:
        if status_norm in ("failed", "cancelled"):
            _persist_folder_scan_job_status(
                jid,
                status=status_norm,
                last_error=last_error,
            )
        _update_job_state(
            jid,
            status=status_norm,
            completed=(status_norm == "completed"),
            schedule_enabled=False,
            last_error=last_error,
        )

    with _SCHEDULE_LOCK:
        if jid:
            ev = _CANCEL_EVENTS.get(jid)
            if ev:
                ev.set()
            _CANCEL_EVENTS.pop(jid, None)
            _SCHEDULED.pop(jid, None)


def _print_then_sleep(cancel_ev: threading.Event, interval_seconds: int, phase: str = "rest") -> bool:
    """Block until interval elapses (phase message printed separately on transition)."""
    return _sleep_with_cancel(cancel_ev, interval_seconds)


def _build_document_assistant(agent_id: str):
    """Return (DocumentService instance, error_message)."""
    from app.services.agent_config import agent_has_file_datasource, get_agent_config
    from app.services.document_service import (
        FILE_EXTRACT_DISK_ROOT,
        FILE_IMAGE_DISK_ROOT,
        FILE_SYS_METADATA_DB,
        FILE_UPLOAD_DISK_ROOT,
        FILE_VECTORS_DB_DIR,
        DocumentService,
    )

    if not agent_has_file_datasource(str(agent_id)):
        return None, "File datasource is not enabled for this agent"

    agent_config = get_agent_config(agent_id=agent_id)
    if not agent_config:
        return None, f"Agent configuration not found for agent_id: {agent_id}"

    api_key = agent_config.get("api_key")
    if not api_key:
        return None, "OpenAI API key not configured for this agent (required for embeddings)"

    try:
        return (
            DocumentService(
                upload_folder=FILE_UPLOAD_DISK_ROOT,
                image_folder=FILE_IMAGE_DISK_ROOT,
                extracted_folder=FILE_EXTRACT_DISK_ROOT,
                db_file=FILE_SYS_METADATA_DB,
                chroma_path=FILE_VECTORS_DB_DIR,
                openai_api_key=str(api_key),
                claude_api_key=str(agent_config.get("claude_api_key") or ""),
                agent_id=agent_id,
            ),
            None,
        )
    except Exception as exc:
        return None, f"Error initializing document service: {exc}"


def get_scheduled_job_status(job_id: str) -> Optional[Dict[str, Any]]:
    """Return in-memory scheduler state for a job, or None if not scheduled here."""
    with _SCHEDULE_LOCK:
        entry = _SCHEDULED.get(job_id)
        if not entry:
            return None
        return {k: v for k, v in entry.items() if k != "thread"}


def list_scheduled_jobs(agent_id: Optional[str] = None) -> Dict[str, Dict[str, Any]]:
    with _SCHEDULE_LOCK:
        out = {}
        for jid, entry in _SCHEDULED.items():
            if agent_id and entry.get("agent_id") != agent_id:
                continue
            out[jid] = {k: v for k, v in entry.items() if k != "thread"}
        return out


def _paths_related(path_a: str, path_b: str) -> bool:
    a = _normalize_scan_path(path_a)
    b = _normalize_scan_path(path_b)
    if not a or not b:
        return False
    if a == b:
        return True
    sep = os.sep
    return a.startswith(b + sep) or b.startswith(a + sep)


def cancel_scheduled_folder_scans_for_paths(
    paths: list[str],
    agent_id: str,
) -> int:
    """Cancel in-process schedulers whose file_path overlaps any of ``paths``."""
    aid = str(agent_id or "").strip()
    paths_norm = [_normalize_scan_path(p) for p in paths if p and str(p).strip()]
    if not aid or not paths_norm:
        return 0
    with _SCHEDULE_LOCK:
        job_ids = [
            jid
            for jid, entry in _SCHEDULED.items()
            if entry.get("agent_id") == aid
            and any(
                _paths_related(str(entry.get("file_path") or ""), p)
                for p in paths_norm
            )
        ]
    cancelled = 0
    for jid in job_ids:
        if cancel_scheduled_folder_scan(jid):
            cancelled += 1
    return cancelled


def try_resume_background_scheduler(
    agent_id: str,
    job_id: str,
    *,
    file_path: str,
    batch_size: Optional[int] = None,
    schedule_enabled: bool = True,
) -> bool:
    """
    Re-attach a background worker when the job exists in DB but is not in memory
    (e.g. after uvicorn reload). Returns True if a worker is now scheduled.
    """
    if get_scheduled_job_status(job_id):
        return True
    from app.services.document_db_service import DocumentDbService
    from app.services.document_service import FILE_SYS_METADATA_DB

    svc = DocumentDbService(db_file=FILE_SYS_METADATA_DB)
    try:
        row = svc.get_folder_scan_job(job_id)
    finally:
        svc.close()
    if not row:
        return False
    status = (row.get("status") or "").lower()
    if status in ("completed", "failed", "cancelled"):
        return False
    bs = batch_size
    if bs is None and row.get("batch_size"):
        bs = int(row["batch_size"])
    root = file_path or row.get("root_path") or ""
    return schedule_folder_scan_job(
        agent_id,
        job_id,
        file_path=root,
        batch_size=bs,
        schedule_enabled=schedule_enabled,
        allow_existing=True,
    )


def is_job_cancelled(job_id: str) -> bool:
    if not job_id:
        return False
    try:
        sched_row = _file_schedule_svc.get_file_schedule_by_job_id(job_id)
        if sched_row and sched_row.get("status") == "cancelled":
            return True
    except Exception as e:
        _logger.error(f"Error checking cancellation status for {job_id}: {e}")
    return False


def wait_for_folder_scan_stop(
    job_id: str,
    *,
    timeout_seconds: float = 1800.0,
    poll_interval_seconds: float = 0.5,
) -> bool:
    """
    Block until ``job_id`` is no longer ingesting / holding the scheduler heartbeat.

    Returns True when stop finished, False if ``timeout_seconds`` elapsed first.
    """
    jid = str(job_id or "").strip()
    if not jid:
        return True
    deadline = time.monotonic() + max(1.0, float(timeout_seconds))
    while time.monotonic() < deadline:
        if not is_folder_scan_stop_pending(jid):
            return True
        time.sleep(poll_interval_seconds)
    return not is_folder_scan_stop_pending(jid)


def wait_for_folder_scans_stop(
    job_ids: List[str],
    *,
    timeout_seconds: float = 1800.0,
    poll_interval_seconds: float = 0.5,
) -> bool:
    """Block until none of the given jobs are still stopping."""
    ids = [str(j).strip() for j in job_ids if j and str(j).strip()]
    if not ids:
        return True
    deadline = time.monotonic() + max(1.0, float(timeout_seconds))
    while time.monotonic() < deadline:
        if not any(is_folder_scan_stop_pending(jid) for jid in ids):
            return True
        time.sleep(poll_interval_seconds)
    return not any(is_folder_scan_stop_pending(jid) for jid in ids)


def is_folder_scan_stop_pending(job_id: str) -> bool:
    """
    True while the job is still actively processing on a worker thread.

    Note: ``cancelling`` is intentionally excluded — that state means stop was
    accepted; waiting on it would deadlock because cleanup runs after the wait.
    """
    jid = str(job_id or "").strip()
    if not jid:
        return False
    with _SCHEDULE_LOCK:
        if jid in _INGEST_IN_PROGRESS:
            return True
        entry = _SCHEDULED.get(jid)
        if entry and (entry.get("status") or "").lower() in ("starting", "running"):
            return True
    heartbeat = _file_schedule_svc.get_scheduler_status()
    if heartbeat and str(heartbeat.get("current_job_id") or "").strip() == jid:
        return True
    job_status = _get_folder_scan_job_status(jid)
    if job_status in ("running", "starting"):
        return True
    return False


def is_folder_scan_cancel_requested(job_id: str) -> bool:
    """True when the user requested stop or the schedule/job row is cancelled."""
    jid = str(job_id or "").strip()
    if not jid:
        return False

    if is_job_cancelled(jid):
        return True

    memo_aid = ""
    memo_root = ""
    with _SCHEDULE_LOCK:
        ev = _CANCEL_EVENTS.get(jid)
        if ev and ev.is_set():
            return True
        entry = _SCHEDULED.get(jid)
        if entry:
            st = (entry.get("status") or "").lower()
            if st in ("cancelling", "cancelled"):
                return True
            if not entry.get("schedule_enabled", True):
                return True
            memo_root = str(entry.get("file_path") or "")
            memo_aid = str(entry.get("agent_id") or "")
            if _is_path_stop_requested_unlocked(memo_aid, memo_root):
                return True
    try:
        from app.services.document_db_service import DocumentDbService
        from app.services.document_service import FILE_SYS_METADATA_DB

        svc = DocumentDbService(db_file=FILE_SYS_METADATA_DB)
        try:
            row = svc.get_folder_scan_job(jid)
        finally:
            svc.close()
        if row:
            if (row.get("status") or "").lower() == "cancelled":
                return True
            root = _normalize_scan_path(str(row.get("root_path") or "")) or memo_root
            aid = memo_aid or str(row.get("agent_id") or "").strip()
            if aid and root and _is_path_stop_requested(aid, root):
                return True
    except Exception:
        pass
    return False


def cancel_scheduled_folder_scan(job_id: str) -> bool:
    """Request immediate cancellation; in-flight ingest aborts at the next cancel check."""
    jid = str(job_id or "").strip()
    if not jid:
        return False
    with _SCHEDULE_LOCK:
        ev = _CANCEL_EVENTS.get(jid)
        if ev is None:
            ev = threading.Event()
            _CANCEL_EVENTS[jid] = ev
        ev.set()
        entry = _SCHEDULED.get(jid)
        if entry:
            entry["schedule_enabled"] = False
            entry["status"] = "cancelling"
            entry["updated_at"] = _now_iso()
    return True


def release_folder_scan_job_after_stop(
    job_id: str,
    *,
    agent_id: str = "",
    file_path: str = "",
) -> None:
    """
    After schedule-stop: drop in-memory scheduler state when safe so the same
    path can be scheduled again. If a file is still ingesting, cleanup happens
    when that file finishes.
    """
    jid = str(job_id or "").strip()
    if not jid:
        return
    aid = str(agent_id or "").strip()
    path_norm = _normalize_scan_path(file_path)
    with _SCHEDULE_LOCK:
        if jid in _INGEST_IN_PROGRESS:
            return
        entry = _SCHEDULED.get(jid)
        if entry:
            if not aid:
                aid = str(entry.get("agent_id") or "").strip()
            if not path_norm:
                path_norm = _normalize_scan_path(str(entry.get("file_path") or ""))
    _finalize_folder_scan_scheduler_job(
        jid,
        aid,
        path_norm,
        status="cancelled",
    )


def _persist_folder_scan_job_status(
    job_id: str,
    *,
    status: str,
    last_error: Optional[str] = None,
) -> None:
    """Write terminal scheduler state to folder_scan_jobs (file_sys_metadata.db)."""
    from app.services.document_db_service import DocumentDbService
    from app.services.document_service import FILE_SYS_METADATA_DB

    jid = str(job_id or "").strip()
    if not jid:
        return
    try:
        svc = DocumentDbService(db_file=FILE_SYS_METADATA_DB)
        try:
            svc.update_folder_scan_job(
                jid,
                status=status,
                last_error=last_error,
            )
        finally:
            svc.close()
    except Exception as exc:
        _logger.warning("persist_folder_scan_job_status failed job_id=%s: %s", jid, exc)


def stop_folder_upload_scheduler(
    agent_id: str,
    *,
    job_id: Optional[str] = None,
    file_path: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Stop the in-process folder upload scheduler and disable per-path scheduling.

    If a file is still being saved (upload-folder HTTP) or ingested by the scheduler,
    stop is deferred until that file finishes, then the worker exits.
    """
    aid = str(agent_id or "").strip()
    path_in = str(file_path or "").strip() if file_path else ""
    path_norm = _normalize_scan_path(path_in) if path_in else ""
    primary_jid = _normalize_job_id_for_stop(job_id)

    if path_norm and is_folder_http_upload_in_progress(aid, path_norm):
        key = _path_key(aid, path_norm)
        with _SCHEDULE_LOCK:
            _PENDING_STOPS[key] = {
                "agent_id": aid,
                "job_id": primary_jid or None,
                "file_path": path_norm,
            }
            _mark_path_stop_requested(aid, path_norm)
        return {
            "stopped": False,
            "stop_pending": True,
            "scheduler_cancel_requested": False,
            "schedule_disabled": False,
            "job_id": primary_jid or None,
            "file_path": path_norm,
            "scheduler_status": "cancelling",
            "completed": False,
            "cancelled": False,
            "message": (
                "Stop scheduled: the current upload-folder request will finish saving "
                "files first, then the scheduler will stop."
            ),
        }

    if path_norm:
        _mark_path_stop_requested(aid, path_norm)

    job_ids = _collect_job_ids_to_stop(aid, job_id=job_id, file_path=path_norm or None)
    cancelled_in_memory = False
    resolved_paths: Set[str] = set()
    if path_norm:
        resolved_paths.add(path_norm)

    ingest_in_progress = False
    with _SCHEDULE_LOCK:
        for jid in job_ids:
            if jid in _INGEST_IN_PROGRESS:
                ingest_in_progress = True
                break

    for jid in job_ids:
        if cancel_scheduled_folder_scan(jid):
            cancelled_in_memory = True
        with _SCHEDULE_LOCK:
            entry = _SCHEDULED.get(jid)
            if entry:
                entry["schedule_enabled"] = False
                if (entry.get("status") or "").lower() not in (
                    "completed",
                    "failed",
                    "cancelled",
                ):
                    entry["status"] = "cancelling"
                entry["updated_at"] = _now_iso()
                fp = str(entry.get("file_path") or "")
                if fp:
                    resolved_paths.add(_normalize_scan_path(fp))
                    _mark_path_stop_requested(aid, fp)

    if not cancelled_in_memory and path_norm:
        cancelled_in_memory = cancel_scheduled_folder_scans_for_paths(
            [path_norm], aid
        ) > 0

    if path_norm:
        _mark_path_stop_requested(aid, path_norm)
    for jid in job_ids:
        _persist_folder_scan_job_status(
            jid,
            status="cancelled",
            last_error="Upload schedule stopped by user",
        )

    schedule_disabled = False
    if not ingest_in_progress:
        if job_ids:
            for jid in job_ids:
                _stop_file_path_schedule(path_norm or "", aid, job_id=jid)
                schedule_disabled = True
        else:
            for rp in resolved_paths:
                _stop_file_path_schedule(rp, aid)
                schedule_disabled = True
            if path_norm and path_norm not in resolved_paths:
                _stop_file_path_schedule(path_norm, aid)
                schedule_disabled = True

    if not job_ids:
        primary_jid = _normalize_job_id_for_stop(job_id)
    else:
        primary_jid = job_ids[0] if job_ids else _normalize_job_id_for_stop(job_id)
    resolved_path = path_norm
    if not resolved_path and primary_jid:
        from app.services.document_db_service import DocumentDbService
        from app.services.document_service import FILE_SYS_METADATA_DB

        svc = DocumentDbService(db_file=FILE_SYS_METADATA_DB)
        try:
            row = svc.get_folder_scan_job(primary_jid)
            if row and row.get("root_path"):
                resolved_path = _normalize_scan_path(str(row["root_path"]))
        finally:
            svc.close()
    if not resolved_path:
        bg_entry = find_scheduled_job_for_file_path(path_norm, aid) if path_norm else None
        resolved_path = str((bg_entry or {}).get("file_path") or "")

    bg = (
        get_scheduled_job_status(primary_jid)
        if primary_jid
        else (find_scheduled_job_for_file_path(resolved_path, aid) if resolved_path else None)
    )
    scheduler_status = (bg or {}).get("status") or (
        "cancelled"
        if schedule_disabled
        else (
            "cancelling"
            if cancelled_in_memory or ingest_in_progress
            else "not_running"
        )
    )

    stop_pending = bool(ingest_in_progress and cancelled_in_memory)
    return {
        "stopped": bool(schedule_disabled or (cancelled_in_memory and not ingest_in_progress)),
        "stop_pending": stop_pending,
        "scheduler_cancel_requested": bool(cancelled_in_memory),
        "schedule_disabled": schedule_disabled,
        "job_id": primary_jid or (bg or {}).get("job_id"),
        "file_path": resolved_path or (bg or {}).get("file_path"),
        "scheduler_status": scheduler_status,
        "completed": False,
        "cancelled": bool(schedule_disabled),
        "message": (
            "Stop requested: uploads will stop as soon as the current file step completes."
            if stop_pending
            else None
        ),
    }


def ensure_global_scheduler_thread_running() -> None:
    global _GLOBAL_THREAD
    with _GLOBAL_THREAD_LOCK:
        if _GLOBAL_THREAD is None or not _GLOBAL_THREAD.is_alive():
            _GLOBAL_CANCEL_EVENT.clear()
            _GLOBAL_THREAD = threading.Thread(
                target=_run_scheduled_loop,
                args=(_GLOBAL_CANCEL_EVENT,),
                name="global-folder-scan-scheduler",
                daemon=True,
            )
            _GLOBAL_THREAD.start()
            _logger.info("Global folder scan scheduler thread started.")


def _register_scheduled_job_in_memory(
    *,
    job_id: str,
    agent_id: str,
    file_path_norm: str,
    schedule_enabled: bool,
    batch_size: Optional[int],
) -> None:
    with _SCHEDULE_LOCK:
        entry = _SCHEDULED.get(job_id)
        if entry and (entry.get("status") or "").lower() in _ACTIVE_SCHEDULER_STATUSES:
            return
        _SCHEDULED[job_id] = {
            "job_id": job_id,
            "agent_id": agent_id,
            "file_path": file_path_norm,
            "status": "running",
            "schedule_enabled": schedule_enabled,
            "batch_size": batch_size,
            "batches_completed": 0,
            "last_step": None,
            "last_error": None,
            "started_at": _now_iso(),
            "updated_at": _now_iso(),
            "completed": False,
        }
        ev = _CANCEL_EVENTS.get(job_id)
        if ev is None:
            _CANCEL_EVENTS[job_id] = threading.Event()
        else:
            ev.clear()


def purge_stale_scheduler_memory(
    agent_id: str,
    file_path: Optional[str] = None,
) -> None:
    """Drop terminal/cancelling in-memory jobs so a new upload is not blocked."""
    aid = str(agent_id or "").strip()
    path_norm = _normalize_scan_path(file_path) if file_path else ""
    terminal = frozenset({"cancelling", "cancelled", "completed", "failed"})
    with _SCHEDULE_LOCK:
        for jid, entry in list(_SCHEDULED.items()):
            if entry.get("agent_id") != aid:
                continue
            entry_path = str(entry.get("file_path") or "")
            if path_norm and not _paths_related(entry_path, path_norm):
                continue
            if (entry.get("status") or "").lower() in terminal:
                _SCHEDULED.pop(jid, None)
                _CANCEL_EVENTS.pop(jid, None)


def signal_immediate_folder_scan_stop(
    agent_id: str,
    job_id: str,
    file_path: str,
) -> None:
    """In-memory stop (no DB). Frees the path for a new schedule right away."""
    jid = str(job_id or "").strip()
    aid = str(agent_id or "").strip()
    fp = str(file_path or "").strip()
    if jid:
        cancel_scheduled_folder_scan(jid)
        with _SCHEDULE_LOCK:
            _SCHEDULED.pop(jid, None)
            _CANCEL_EVENTS.pop(jid, None)
    if fp and aid:
        _clear_path_stop_requested(aid, fp)
        purge_stale_scheduler_memory(aid, fp)


def schedule_folder_scan_job(
    agent_id: str,
    job_id: str,
    *,
    file_path: str,
    batch_size: Optional[int] = None,
    schedule_enabled: bool = True,
    free_time: Optional[int] = None,
    allow_existing: bool = False,
) -> Optional[str]:
    """
    Queue folder path once as a single atomic task in file_schedules queue,
    and ensure the global heartbeat background thread is running.

    Returns job_id on success. When ``allow_existing`` is False (API uploads),
    an active schedule for the same path returns None without writing to DB.
    When ``allow_existing`` is True (resume), reuses the existing in-flight job.
    """
    job_id = str(job_id).strip()
    agent_id = str(agent_id).strip()
    file_path_norm = _normalize_scan_path(file_path)
    if not job_id or not agent_id or not file_path_norm:
        return None

    from app.utils.file_schedule_utils import is_global_scheduler_resting

    if is_global_scheduler_resting() and not allow_existing:
        print(
            f"Folder scan rejected — scheduler is on rest: {file_path_norm}",
            flush=True,
        )
        return None

    with _PATH_SCHEDULE_ENQUEUE_LOCK:
        _clear_path_stop_requested(agent_id, file_path_norm)
        purge_stale_scheduler_memory(agent_id, file_path_norm)

        existing = get_scheduled_job_status(job_id)
        if existing and existing.get("status") in _ACTIVE_SCHEDULER_STATUSES:
            return job_id

        mem_existing = find_scheduled_job_for_file_path(file_path_norm, agent_id)
        if mem_existing and (mem_existing.get("status") or "").lower() in _UPLOAD_BLOCKING_SCHEDULER_STATUSES:
            reuse_jid = str(mem_existing.get("job_id") or "").strip()
            if reuse_jid:
                # If cancellation/completion already happened in DB or via cancel event,
                # drop stale in-memory state instead of blocking a fresh upload.
                db_status = _get_folder_scan_job_status(reuse_jid)
                with _SCHEDULE_LOCK:
                    ev = _CANCEL_EVENTS.get(reuse_jid)
                    cancel_requested = bool(ev and ev.is_set())
                    schedule_enabled = bool(
                        (_SCHEDULED.get(reuse_jid) or {}).get("schedule_enabled", True)
                    )
                is_stale = cancel_requested or (not schedule_enabled) or db_status in {
                    "cancelled",
                    "completed",
                    "failed",
                }
                if is_stale:
                    with _SCHEDULE_LOCK:
                        _SCHEDULED.pop(reuse_jid, None)
                        _CANCEL_EVENTS.pop(reuse_jid, None)
                    mem_existing = None
                    reuse_jid = ""
            if reuse_jid:
 
                if not allow_existing:
                    print(
                        f"Folder scan rejected — path already active in memory: {file_path_norm}",
                        flush=True,
                    )
                    return None
                ensure_global_scheduler_thread_running()
                print(
                    f"Folder scan already in memory for {file_path_norm} "
                    f"(job_id={reuse_jid})",
                    flush=True,
                )
                return reuse_jid

    doc_assistant, err = _build_document_assistant(agent_id)
    if not doc_assistant:
        _update_job_state(job_id, status="failed", last_error=err, completed=True)
        return None

    try:
        effective_jid = ensure_file_schedule_row(
            _file_schedule_svc,
            file_path=file_path_norm,
            agent_id=agent_id,
            schedule_enabled=True,
            job_id=job_id,
        )
    except Exception as exc:
        _logger.exception(f"Error queueing folder scan task: {exc}")
        return None

    if not effective_jid:
        return None

    if effective_jid != job_id and not allow_existing:
        print(
            f"Folder scan deduped to existing job {effective_jid} for {file_path_norm}",
            flush=True,
        )

    with _PATH_SCHEDULE_ENQUEUE_LOCK:
        _register_scheduled_job_in_memory(
            job_id=effective_jid,
            agent_id=agent_id,
            file_path_norm=file_path_norm,
            schedule_enabled=schedule_enabled,
            batch_size=batch_size,
        )

    _file_schedule_svc.consolidate_active_schedules_for_path(
        file_path=file_path_norm,
        agent_id=agent_id,
        keep_job_id=effective_jid,
    )

    ensure_global_scheduler_thread_running()
    print(
        f"Folder scan job queued (job_id={effective_jid}, file_path={file_path_norm})",
        flush=True,
    )
    return effective_jid


def _sleep_with_cancel(cancel_ev: threading.Event, seconds: int) -> bool:
    """Sleep up to ``seconds``; return True if cancelled."""
    end = time.monotonic() + seconds
    while time.monotonic() < end:
        if cancel_ev.is_set():
            return True
        time.sleep(min(1.0, end - time.monotonic()))
    return cancel_ev.is_set()


def _update_job_state(job_id: str, **fields: Any) -> None:
    with _SCHEDULE_LOCK:
        entry = _SCHEDULED.get(job_id)
        if entry:
            entry.update(fields)
            entry["updated_at"] = _now_iso()


def _sync_schedule_state_to_job(
    job_id: str,
    file_path: str,
    agent_id: str,
    schedule_enabled: bool,
) -> str:
    """No-op wrapper returning current phase for backwards compatibility."""
    status = _file_schedule_svc.get_scheduler_status()
    return (status.get("phase") if status else "active").lower()


_LAST_LOGGED_REMAINING_MIN = None
_LAST_LOGGED_TASK_SNAPSHOT_AT = 0.0


def log_remaining_minutes_if_needed(phase: str, remaining_secs: float) -> None:
    global _LAST_LOGGED_REMAINING_MIN
    if phase.lower() != "active":
        _LAST_LOGGED_REMAINING_MIN = None
        return
    
    # 20 minutes active window = 1200 seconds.
    # Minutes remaining:
    remaining_min = int(remaining_secs // 60)
    
    if remaining_min != _LAST_LOGGED_REMAINING_MIN:
        _LAST_LOGGED_REMAINING_MIN = remaining_min
        msg = f"[SCHEDULER] {remaining_min} minutes remaining in the active phase."
        print(msg, flush=True)
        _logger.info(msg)


def log_all_tasks_snapshot_if_needed(all_tasks_str: Any, interval_seconds: int = 60) -> None:
    """Throttle noisy queue snapshot logs to once per interval."""
    global _LAST_LOGGED_TASK_SNAPSHOT_AT
    now_mono = time.monotonic()
    if (now_mono - _LAST_LOGGED_TASK_SNAPSHOT_AT) < max(1, int(interval_seconds)):
        return
    _LAST_LOGGED_TASK_SNAPSHOT_AT = now_mono
    msg = f"[SCHEDULER DEBUG] All tasks in DB: {all_tasks_str}"
    print(msg, flush=True)
    _logger.info(msg)



def _run_scheduled_loop(cancel_ev: threading.Event) -> None:
    _logger.info("Entering global scheduler thread loop")

    while not cancel_ev.is_set():
        try:
            # 1. Heartbeat Check
            status = _file_schedule_svc.get_scheduler_status()
            if not status:
                _sleep_with_cancel(cancel_ev, 5)
                continue

            current_job_id = status.get("current_job_id")
            phase = (status.get("phase") or "active").lower()
            active_s = int(status.get("active_seconds") or 1200)
            rest_s = int(status.get("rest_seconds") or 600)
            phase_started_at_str = status.get("phase_started_at")
            next_phase_at_str = status.get("next_phase_at")

            current_time = datetime.now()
            next_phase_at = datetime.fromisoformat(next_phase_at_str) if next_phase_at_str else current_time

            if next_phase_at_str and phase == "active":
                remaining_secs = (next_phase_at - current_time).total_seconds()
                log_remaining_minutes_if_needed(phase, remaining_secs)

            # Cancellation Check:
            if current_job_id and is_job_cancelled(current_job_id):
                _logger.info(f"[SCHEDULER] Job {current_job_id} has been cancelled. Resetting system to idle.")
                _file_schedule_svc.update_scheduler_status(
                    phase="active",
                    phase_started_at=current_time.isoformat(),
                    next_phase_at=(current_time + timedelta(seconds=active_s)).isoformat(),
                    clear_job_id=True
                )
                _finalize_folder_scan_scheduler_job(
                    current_job_id,
                    "",
                    "",
                    status="cancelled"
                )
                continue

            # 2. Rest State Logic
            if phase == 'rest':
                if current_time >= next_phase_at:
                    new_phase = 'active'
                    new_next_phase_at = current_time + timedelta(seconds=active_s)
                    _file_schedule_svc.update_scheduler_status(
                        phase=new_phase,
                        phase_started_at=current_time.isoformat(),
                        next_phase_at=new_next_phase_at.isoformat()
                    )
                    _logger.info(f"[SCHEDULER] Advanced phase from rest to active. Next phase at {new_next_phase_at.isoformat()}")
                    print(f"[SCHEDULER] Advanced phase from rest to active. Next phase at {new_next_phase_at.isoformat()}", flush=True)
                    continue
                else:
                    # execute a short sleep interval while continuously listening for cancellation flags. Do not process files.
                    for _ in range(5):
                        if cancel_ev.is_set():
                            break
                        latest_status = _file_schedule_svc.get_scheduler_status()
                        if latest_status and latest_status.get("current_job_id") and is_job_cancelled(latest_status.get("current_job_id")):
                            break
                        time.sleep(1)
                    continue

            # 3. Active State Logic
            if phase == 'active':
                if current_time >= next_phase_at:
                    new_phase = 'rest'
                    new_next_phase_at = current_time + timedelta(seconds=rest_s)
                    _file_schedule_svc.update_scheduler_status(
                        phase=new_phase,
                        phase_started_at=current_time.isoformat(),
                        next_phase_at=new_next_phase_at.isoformat()
                    )
                    _logger.info(f"[SCHEDULER] Active window expired smoothly. Transitioned to rest phase. Next phase at {new_next_phase_at.isoformat()}")
                    print(f"[SCHEDULER] Active window expired smoothly. Transitioned to rest phase. Next phase at {new_next_phase_at.isoformat()}", flush=True)
                    continue

            # 4. Task Execution: select the oldest row where status == 'pending'
            from app.core.database import get_db_session
            from sqlalchemy import select

            pending_task = None
            task_id = None
            task_file_path = None
            task_agent_id = None
            task_job_id = None

            with get_db_session() as db:
                # Debug print
                all_tasks = db.query(FileSchedule).all()
                all_tasks_str = [{t.id: (t.job_id, t.status)} for t in all_tasks]
                log_all_tasks_snapshot_if_needed(all_tasks_str, interval_seconds=60)

                q = select(FileSchedule).where(FileSchedule.status == "pending").order_by(FileSchedule.id.asc())
                pending_task_row = db.execute(q).scalars().first()
                if pending_task_row:
                    pending_task_row.status = "processing"
                    task_id = pending_task_row.id
                    task_file_path = pending_task_row.file_path
                    task_agent_id = pending_task_row.agent_id
                    task_job_id = pending_task_row.job_id
                    db.commit()

            if task_id is None:
                _sleep_with_cancel(cancel_ev, 2)
                continue

            # Map its job_id to scheduler_status.current_job_id
            _file_schedule_svc.update_scheduler_status(
                phase=phase,
                phase_started_at=phase_started_at_str,
                next_phase_at=next_phase_at_str,
                current_job_id=task_job_id
            )

            _logger.info(f"[SCHEDULER] Found pending task id={task_id} job_id={task_job_id} path={task_file_path}")
            doc_assistant, err = _build_document_assistant(task_agent_id)
            if not doc_assistant:
                _logger.error(f"[SCHEDULER] Failed to build document assistant for agent {task_agent_id}: {err}")
                with get_db_session() as db:
                    t = db.query(FileSchedule).filter(FileSchedule.id == task_id).first()
                    if t:
                        t.status = "failed"
                        db.commit()
                _file_schedule_svc.update_scheduler_status(
                    phase=phase,
                    phase_started_at=phase_started_at_str,
                    next_phase_at=next_phase_at_str,
                    clear_job_id=True
                )
                continue

            success = False
            error_message = None
            rest_paused = False
            res_stats = {"uploaded": 0, "skipped": 0, "total_chunks": 0}

            try:
                # Ingestion logic wrapper
                if os.path.isdir(task_file_path):
                    # Directory task - process synchronously using scan_folder_path
                    max_file_size = doc_assistant.DEFAULT_MAX_FILE_SIZE
                    if task_job_id:
                        scan_job = doc_assistant.db_service.get_folder_scan_job(task_job_id)
                        if scan_job and scan_job.get("max_file_size"):
                            max_file_size = int(scan_job["max_file_size"])

                    _logger.info(f"[SCHEDULER] Scanning folder path: {task_file_path}")
                    res = doc_assistant.scan_folder_path(
                        folder_path=task_file_path,
                        max_file_size=max_file_size,
                        job_id=task_job_id
                    )
                    if res.get("cancelled"):
                        success = False
                        error_message = res.get("message") or "Job cancelled by user"
                    elif res.get("rest_paused"):
                        success = False
                        rest_paused = True
                        error_message = res.get("message") or "Scheduler rest phase"
                        res_stats = {
                            "uploaded": int(res.get("uploaded") or 0),
                            "skipped": int(res.get("skipped") or 0),
                            "total_chunks": int(res.get("total_chunks") or 0),
                        }
                    elif res.get("success"):
                        success = True
                        res_stats = {
                            "uploaded": res.get("uploaded", 0),
                            "skipped": res.get("skipped", 0),
                            "total_chunks": res.get("total_chunks", 0)
                        }
                    else:
                        error_message = res.get("message") or "Folder scan failed"
                        success = False
                else:
                    # Single file task
                    filename = os.path.basename(task_file_path)
                    file_ext = filename.split(".")[-1].lower()
                    parent_dir = os.path.dirname(task_file_path)
                    root_folder_name = os.path.basename(parent_dir)

                    doc_assistant.db_service.save_folder(root_folder_name, root_folder_name, "", parent_dir)

                    content, _ = doc_assistant.process_file_from_path(task_file_path, file_ext)
                    if content and not content.startswith("Error") and not content.startswith("Unsupported"):
                        file_id = doc_assistant.get_file_hash(content, filename, "")
                        doc_assistant.db_service.save_file(
                            file_id, filename, file_ext, task_file_path,
                            content, root_folder_name, "",
                            "", "", doc_assistant.agent_id
                        )
                        if doc_assistant.vector_db:
                            doc_meta = {
                                "filename": filename,
                                "file_type": file_ext,
                                "folder_path": root_folder_name,
                                "relative_path": "",
                                "upload_date": datetime.now().isoformat(),
                                "agent_id": doc_assistant._ingestion_agent_id,
                            }
                            doc_assistant._vector_index_document(
                                file_id, content, doc_meta, file_path=task_file_path, file_ext=file_ext
                            )
                        success = True
                        res_stats = {"uploaded": 1, "skipped": 0, "total_chunks": 0}
                    else:
                        error_message = content or "File content error"
                        success = False
            except Exception as e:
                _logger.exception(f"Error processing scheduled task {task_id}: {e}")
                error_message = str(e)
                success = False

            scan_cancelled = bool(
                (not success and error_message and "cancel" in error_message.lower())
                or is_job_cancelled(task_job_id)
            )
            if rest_paused:
                _logger.info(
                    f"[SCHEDULER] Job {task_job_id} paused for rest; re-queueing as pending."
                )
                uploaded_n = int(res_stats.get("uploaded") or 0)
                if uploaded_n == 0 and os.path.isdir(task_file_path):
                    try:
                        root_name = os.path.basename(
                            os.path.abspath(task_file_path)
                        )
                        doc_assistant.db_service.delete_folder(root_name)
                    except Exception as exc:
                        _logger.warning(
                            "Could not remove empty folder after rest pause: %s",
                            exc,
                        )
                with get_db_session() as db:
                    t = db.query(FileSchedule).filter(FileSchedule.id == task_id).first()
                    if t:
                        t.status = "pending"
                        db.commit()
                if task_job_id:
                    try:
                        doc_assistant.db_service.update_folder_scan_job(
                            task_job_id,
                            status="pending",
                            last_error=error_message,
                        )
                    except Exception:
                        pass
                _file_schedule_svc.update_scheduler_status(
                    phase=phase,
                    phase_started_at=phase_started_at_str,
                    next_phase_at=next_phase_at_str,
                    clear_job_id=True,
                )
                continue

            if scan_cancelled:
                _logger.info(f"[SCHEDULER] Job {task_job_id} was cancelled mid-run. Cleaning up.")
                _file_schedule_svc.update_scheduler_status(
                    phase="active",
                    phase_started_at=datetime.now().isoformat(),
                    next_phase_at=(datetime.now() + timedelta(seconds=active_s)).isoformat(),
                    clear_job_id=True
                )
                with get_db_session() as db:
                    t = db.query(FileSchedule).filter(FileSchedule.id == task_id).first()
                    if t:
                        t.status = "cancelled"
                        db.commit()
                _finalize_folder_scan_scheduler_job(
                    task_job_id,
                    task_agent_id,
                    task_file_path,
                    status="cancelled",
                    last_error="Upload schedule stopped by user"
                )
                continue

            uploaded_n = int(res_stats.get("uploaded") or 0)
            skipped_n = int(res_stats.get("skipped") or 0)
            if uploaded_n > 0 or (skipped_n > 0 and success):
                status_to_set = "completed" if success else "failed"
            elif success and uploaded_n == 0 and skipped_n == 0:
                status_to_set = "completed"
            else:
                status_to_set = "failed"
            with get_db_session() as db:
                t = db.query(FileSchedule).filter(FileSchedule.id == task_id).first()
                if t:
                    t.status = status_to_set
                    db.commit()

            if task_job_id:
                try:
                    stats_json = json.dumps(res_stats)
                    doc_assistant.db_service.update_folder_scan_job(
                        task_job_id,
                        status=status_to_set,
                        stats_json=stats_json,
                        last_error=error_message if not success else None
                    )

                    _finalize_folder_scan_scheduler_job(
                        task_job_id,
                        task_agent_id,
                        task_file_path,
                        status=status_to_set,
                        last_error=error_message
                    )
                except Exception as ex:
                    _logger.warning(f"Failed to update folder scan job status/stats for job {task_job_id}: {ex}")

            # Mid-Task Overflow Management
            current_time = datetime.now()
            status = _file_schedule_svc.get_scheduler_status()
            if status:
                next_phase_at_str = status.get("next_phase_at")
                next_phase_at = datetime.fromisoformat(next_phase_at_str) if next_phase_at_str else current_time
                active_s = int(status.get("active_seconds") or 1200)

                if current_time >= next_phase_at:
                    remaining_rest, rest_started, rest_next_at = (
                        remaining_rest_after_upload_overflow(status, as_of=current_time)
                    )
                    _file_schedule_svc.update_scheduler_status(
                        phase="rest",
                        phase_started_at=rest_started.isoformat(),
                        next_phase_at=rest_next_at.isoformat(),
                        clear_job_id=True,
                    )
                    _logger.warning(
                        "[SCHEDULER] Mid-upload overflow detected! "
                        "Enforcing full %ss rest from upload completion. "
                        "Next phase at %s",
                        remaining_rest,
                        rest_next_at.isoformat(),
                    )
                    print(
                        "[SCHEDULER] Mid-upload overflow detected! "
                        f"Enforcing full {remaining_rest}s rest from upload completion. "
                        f"Next phase at {rest_next_at.isoformat()}",
                        flush=True,
                    )
                else:
                    _file_schedule_svc.update_scheduler_status(
                        clear_job_id=True
                    )

        except Exception as e:
            _logger.exception(f"Error in global scheduler thread loop iteration: {e}")
            _sleep_with_cancel(cancel_ev, 5)
