service_manager

zocrates.rag.service_manager

Service manager: holds the global/singleton LibrarySyncer instance.

_init_lock module-attribute

_init_lock = threading.Lock()

_syncer module-attribute

_syncer = None

logger module-attribute

logger = logging.getLogger(__name__)

LibrarySyncer

LibrarySyncer(zotero_data_dir, sqlite_path)

Coordinates RAG indexing pipeline.

Initialize the LibrarySyncer with all RAG pipeline components.

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path	Root Zotero data directory containing zotero.sqlite and the storage/ subdirectory.	required
sqlite_path	Path	Absolute path to zotero.sqlite.	required

Source code in zocrates/rag/library_syncer.py

def __init__(self, zotero_data_dir: Path, sqlite_path: Path) -> None:
    """Initialize the LibrarySyncer with all RAG pipeline components.

    Args:
        zotero_data_dir: Root Zotero data directory containing
            ``zotero.sqlite`` and the ``storage/`` subdirectory.
        sqlite_path: Absolute path to ``zotero.sqlite``.
    """
    self._zotero_data_dir = zotero_data_dir
    storage_dir = zotero_data_dir / "storage"

    self.db = ZoteroDB(sqlite_path, storage_dir)

    md_cache = get_corpus_dir(zotero_data_dir)
    active_converter = get_document_converter_id(zotero_data_dir)
    datalab_fallback = get_datalab_non_pdf_fallback(zotero_data_dir)
    self.processor = DocumentProcessor(
        md_cache,
        document_converter_id=active_converter,
        datalab_non_pdf_fallback=datalab_fallback,
    )

    embedder = get_embedder()
    self.store = VectorStore(
        chroma_dir=get_chroma_dir(zotero_data_dir),
        fts_db_path=get_fts_db_path(zotero_data_dir),
        embedder=embedder,
    )

    self._running = False
    self.active_context_keys = None

close

close()

Tear down resources held by this syncer instance.

Source code in zocrates/rag/library_syncer.py

def close(self) -> None:
    """Tear down resources held by this syncer instance."""
    logger.info("Closing LibrarySyncer and releasing resources.")
    try:
        if hasattr(self, 'store') and hasattr(self.store, 'close'):
            self.store.close()
        elif hasattr(self, 'store') and hasattr(self.store, '_client'):
            # PersistentClient doesn't require explicit close, but we clear references
            del self.store._client
    except Exception as exc:
        logger.warning("Error during LibrarySyncer teardown: %s", exc)

full_sync async

full_sync(progress_callback=None, cancel_check=None)

Scan every item, note, and standalone attachment in the library and ingest those not yet indexed.

Additive Phase 7 onboarding parameters (backward compatible):

Parameters:

Name	Type	Description	Default
progress_callback	Callable[[dict], None] | None	Optional synchronous callback invoked at each item boundary with a progress dict (stage, processed, skipped, errors, total, current_title, elapsed_ms). Callback failures are logged and never abort the sync (fail-open).	None
cancel_check	Callable[[], bool] | None	Optional cooperative-cancel predicate polled at the existing per-item yield points. When it returns True the sync finishes the current item and stops, skipping the tail graph-relations/orphan-purge phases. Resume semantics are untouched (store.is_ingested keying + SHA-256 conversion cache).	None

Source code in zocrates/rag/library_syncer.py

async def full_sync(
    self,
    progress_callback: Callable[[dict], None] | None = None,
    cancel_check: Callable[[], bool] | None = None,
) -> dict:
    """Scan every item, note, and standalone attachment in the library and ingest those not yet indexed.

    Additive Phase 7 onboarding parameters (backward compatible):

    Args:
        progress_callback: Optional synchronous callback invoked at each
            item boundary with a progress dict (``stage``, ``processed``,
            ``skipped``, ``errors``, ``total``, ``current_title``,
            ``elapsed_ms``). Callback failures are logged and never abort
            the sync (fail-open).
        cancel_check: Optional cooperative-cancel predicate polled at the
            existing per-item yield points. When it returns True the sync
            finishes the current item and stops, skipping the tail
            graph-relations/orphan-purge phases. Resume semantics are
            untouched (``store.is_ingested`` keying + SHA-256 conversion
            cache).
    """
    sync_started = time.perf_counter()

    def _report_progress(stage: str, current_title: str = "") -> None:
        """Invoke the progress callback with current sync progress metrics.

        Args:
            stage: Current stage of synchronization (e.g. 'syncing', 'indexing').
            current_title: Title of the document currently being processed.
        """
        if progress_callback is None:
            return
        try:
            progress_callback({
                "stage": stage,
                "processed": processed,
                "skipped": skipped,
                "errors": errors,
                "total": total,
                "current_title": current_title,
                "elapsed_ms": int((time.perf_counter() - sync_started) * 1000),
            })
        except Exception as exc:
            logger.warning("Sync progress callback failed: %s", exc)

    def _cancel_requested() -> bool:
        """Check if cooperative cancellation was requested for the sync job.

        Returns:
            True if cancellation was requested, False otherwise.
        """
        if cancel_check is None:
            return False
        try:
            return bool(cancel_check())
        except Exception as exc:
            logger.warning("Sync cancel_check failed: %s", exc)
            return False

    try:
        auto_abort_threshold = int(get_sync_auto_abort_failures(self._zotero_data_dir))
    except (ValueError, TypeError):
        auto_abort_threshold = DEFAULT_SYNC_AUTO_ABORT_FAILURES
    consecutive_failures = 0
    cancelled = False
    auto_aborted = False
    # Synchronize metadata for existing bundles modified in Zotero
    try:
        staleness_res = await asyncio.to_thread(self.sync_metadata_staleness)
        logger.info("Metadata staleness sync in full_sync complete: %s", staleness_res)
    except Exception as exc:
        logger.warning("Error during metadata staleness sync in full_sync: %s", exc)
        staleness_res = {"updated": 0}

    items = self.db.get_all_items()
    standalone_notes = self.db.get_standalone_notes()
    standalone_attachments = self.db.get_standalone_attachments()

    total = len(items) + len(standalone_notes) + len(standalone_attachments)
    processed = 0
    skipped = 0
    errors = 0
    # Emit an initial snapshot immediately so SSE subscribers get the
    # determinate total before the first (potentially multi-minute)
    # conversion finishes.
    _report_progress("starting")

    logger.info(
        "Starting full library sync: %d regular items, %d standalone notes, %d standalone attachments found.",
        len(items),
        len(standalone_notes),
        len(standalone_attachments),
    )

    # Auto-backfill missing parent_item_key values in conversion_meta.db
    all_atts = self.db.get_all_attachments()
    for att in all_atts:
        if att.parent_item_key:
            self.processor.ensure_parent_item_key(att.item_key, att.parent_item_key)

    for item in items:
        try:
            result = await self._ingest_item(
                item.item_id, item.item_key, item.title
            )
            if result == "processed":
                processed += 1
            else:
                skipped += 1
            consecutive_failures = 0
        except Exception as exc:
            logger.error(
                "Error ingesting item %s: %s", item.item_key, exc, exc_info=True
            )
            errors += 1
            consecutive_failures += 1
        await asyncio.sleep(0)
        _report_progress("items", item.title)
        if _cancel_requested():
            cancelled = True
            break
        if auto_abort_threshold > 0 and consecutive_failures >= auto_abort_threshold:
            logger.warning(
                "Auto-aborting full sync after %d consecutive failures (threshold: %d).",
                consecutive_failures,
                auto_abort_threshold,
            )
            auto_aborted = True
            break

    for note in standalone_notes:
        try:
            result = await self._ingest_standalone_note(note)
            if result == "processed":
                processed += 1
            else:
                skipped += 1
            consecutive_failures = 0
        except Exception as exc:
            logger.error(
                "Error ingesting standalone note %s: %s", note.note_key, exc, exc_info=True
            )
            errors += 1
            consecutive_failures += 1
        await asyncio.sleep(0)
        _report_progress("notes", getattr(note, "title", "") or "")
        if _cancel_requested():
            cancelled = True
            break
        if auto_abort_threshold > 0 and consecutive_failures >= auto_abort_threshold:
            logger.warning(
                "Auto-aborting full sync after %d consecutive failures (threshold: %d).",
                consecutive_failures,
                auto_abort_threshold,
            )
            auto_aborted = True
            break

    for att in standalone_attachments:
        try:
            result = await self._ingest_standalone_attachment(att)
            if result == "processed":
                processed += 1
            else:
                skipped += 1
            consecutive_failures = 0
        except Exception as exc:
            logger.error(
                "Error ingesting standalone attachment %s: %s", att.item_key, exc, exc_info=True
            )
            errors += 1
            consecutive_failures += 1
        await asyncio.sleep(0)
        _report_progress("attachments", getattr(att, "title", "") or "")
        if _cancel_requested():
            cancelled = True
            break
        if auto_abort_threshold > 0 and consecutive_failures >= auto_abort_threshold:
            logger.warning(
                "Auto-aborting full sync after %d consecutive failures (threshold: %d).",
                consecutive_failures,
                auto_abort_threshold,
            )
            auto_aborted = True
            break

    # Pass of manual relations and shared tags synchronization
    # (skipped on cooperative cancel / auto-abort — the next successful
    # /sync executes it; plan §6.3 tail-phase semantics)
    meta_db_path = self.processor._cache_dir / "conversion_meta.db"
    if meta_db_path.exists() and not cancelled and not auto_aborted:
        try:
            def _sync_graph_relations_worker() -> None:
                """Sync human-curated relations, shared tags, and regenerate bundle markdown."""
                from zocrates.rag.bundle_catalog import regenerate_item_relations_md
                from zocrates.config import get_graph_relations_config
                from zocrates.rag.document_processor import _open_meta_db
                from zocrates.rag.graph_engine import sync_manual_relations, sync_shared_tags

                cfg = get_graph_relations_config(self._zotero_data_dir)
                if not cfg.get("enabled", True):
                    return

                with _open_meta_db(self.processor._cache_dir) as meta_conn:
                    n_manual = 0
                    n_tags = 0
                    if cfg.get("include_human_curated", True):
                        n_manual = sync_manual_relations(meta_conn, self._zotero_data_dir)
                    if cfg.get("include_shared_tags", True):
                        n_tags = sync_shared_tags(meta_conn, self._zotero_data_dir)

                    if (n_manual > 0 or n_tags > 0) and cfg.get("sync_bundle_files", True):
                        rows = meta_conn.execute("SELECT item_key FROM conversion_meta").fetchall()
                        for r in rows:
                            try:
                                regenerate_item_relations_md(r["item_key"], self.processor._cache_dir)
                            except Exception as exc_inner:
                                logger.warning(
                                    "Failed to regenerate item_relations.md for %s during full_sync: %s",
                                    r["item_key"],
                                    exc_inner,
                                )

            await asyncio.to_thread(_sync_graph_relations_worker)
        except Exception as exc:
            logger.warning(
                "Error syncing manual relations or shared tags in full_sync: %s",
                exc,
            )

    # Orphan detection and cleanup (skipped on cooperative cancel /
    # auto-abort — same tail-phase semantics as above)
    orphaned_keys: set[str] = set()
    if not cancelled and not auto_aborted:
        active_keys: set[str] = set()
        for item in items:
            active_keys.add(item.item_key)
            for att in self.db.get_attachments(item.item_id):
                active_keys.add(att.item_key)
            for note in self.db.get_notes_for_item(item.item_id):
                active_keys.add(note.note_key)
        for note in standalone_notes:
            active_keys.add(note.note_key)
        for att in standalone_attachments:
            active_keys.add(att.item_key)

        all_known_keys = (
            set(self.processor.list_cached_keys())
            | self.processor.get_all_meta_keys()
            | self.store.get_all_item_keys()
        )
        if not active_keys:
            logger.warning("Skipping full sync orphan purge: no active items found in database")
        else:
            orphaned_keys = all_known_keys - active_keys

        if orphaned_keys:
            logger.info("Cleaning up %d orphaned cached items...", len(orphaned_keys))
            for o_key in orphaned_keys:
                self.purge_item(o_key, regenerate_index=False)

    summary = {
        "total": total,
        "processed": processed,
        "skipped": skipped,
        "errors": errors,
        "orphans_cleaned": len(orphaned_keys),
        "metadata_refreshed": staleness_res.get("updated", 0),
        "cancelled": cancelled,
        "auto_aborted": auto_aborted,
    }
    self.processor.regenerate_root_index()
    logger.info("Full sync complete: %s", summary)
    return summary

ingest_items async

ingest_items(item_ids)

Force-ingest a list of item, note, or attachment IDs (e.g. newly added by Zotero).

Source code in zocrates/rag/library_syncer.py

async def ingest_items(self, item_ids: list[int]) -> dict:
    """Force-ingest a list of item, note, or attachment IDs (e.g. newly added by Zotero)."""
    items = self.db.get_items_by_ids(item_ids)
    processed = 0
    errors = 0

    for item in items:
        try:
            if item.item_type == "note":
                note = self.db.get_note_by_id(item.item_id)
                if note:
                    await self._ingest_standalone_note(note, force=True)
                    processed += 1
            elif item.item_type == "attachment":
                att = self.db.get_attachment_by_id(item.item_id)
                if att:
                    await self._ingest_standalone_attachment(att, force=True)
                    processed += 1
            else:
                await self._ingest_item(
                    item.item_id, item.item_key, item.title, force=True
                )
                processed += 1
        except Exception as exc:
            logger.error(
                "Error ingesting item %s: %s", item.item_key, exc, exc_info=True
            )
            errors += 1
        await asyncio.sleep(0)

    summary = {"requested": len(item_ids), "processed": processed, "errors": errors}
    self.processor.regenerate_root_index()
    return summary

purge_item

purge_item(item_key, regenerate_index=True)

Atomically purge an item across all four datastores: disk bundle, metadata DB, vector/FTS store, and root index.

Source code in zocrates/rag/library_syncer.py

def purge_item(self, item_key: str, regenerate_index: bool = True) -> None:
    """Atomically purge an item across all four datastores: disk bundle, metadata DB, vector/FTS store, and root index."""
    if not item_key:
        return
    self.processor.delete_cached(item_key, update_root_index=False)
    self.store.delete_document(item_key)
    if regenerate_index:
        self.processor.regenerate_root_index()

sync_item async

sync_item(item_id_or_key=None, item_key=None, document_title='', force=False)

Sync a single item, note, or attachment by ID or key.

Source code in zocrates/rag/library_syncer.py

async def sync_item(
    self,
    item_id_or_key: int | str | None = None,
    item_key: str | None = None,
    document_title: str = "",
    force: bool = False,
) -> str:
    """Sync a single item, note, or attachment by ID or key."""
    target_key: str | None = item_key
    target_id: int | None = None

    if isinstance(item_id_or_key, str):
        target_key = item_id_or_key
    elif isinstance(item_id_or_key, int):
        target_id = item_id_or_key

    if target_id is None and target_key is not None:
        item = self.db.get_item_by_key(target_key)
        if item:
            return await self._ingest_item(item.item_id, item.item_key, item.title, force=force)
        att = self.db.get_attachment_by_key(target_key)
        if att:
            return await self._ingest_standalone_attachment(att, force=force)
        note = self.db.get_note_by_key(target_key)
        if note:
            return await self._ingest_standalone_note(note, force=force)
        return "skipped"

    if target_id is not None:
        item = self.db.get_item(target_id)
        if item:
            return await self._ingest_item(item.item_id, item.item_key, item.title, force=force)
        att = self.db.get_attachment_by_id(target_id)
        if att:
            return await self._ingest_standalone_attachment(att, force=force)
        note = self.db.get_note_by_id(target_id)
        if note:
            return await self._ingest_standalone_note(note, force=force)

    return "skipped"

sync_metadata_staleness

sync_metadata_staleness()

Scan cached bundles and synchronize frontmatter and FTS5 for any modified in Zotero.

Executes a <5ms batched staleness check comparing conversion_meta.db against zotero.sqlite. If items are stale, updates their YAML frontmatter, toc.json, chunks_fts Chunk 0, and Chroma metadata without re-running document converters or OCR engines.

Returns:

Type	Description
dict[str, int]	Summary dict containing counts of total, updated, skipped, and errored items.

Source code in zocrates/rag/library_syncer.py

def sync_metadata_staleness(self) -> dict[str, int]:
    """Scan cached bundles and synchronize frontmatter and FTS5 for any modified in Zotero.

    Executes a <5ms batched staleness check comparing conversion_meta.db against
    zotero.sqlite. If items are stale, updates their YAML frontmatter, toc.json,
    chunks_fts Chunk 0, and Chroma metadata without re-running document converters
    or OCR engines.

    Returns:
        Summary dict containing counts of total, updated, skipped, and errored items.
    """
    from zocrates.rag.frontmatter_refresher import sync_stale_metadata
    md_cache = get_corpus_dir(self._zotero_data_dir)
    res = sync_stale_metadata(
        zotero_dir=self._zotero_data_dir,
        store=self.store,
        cache_dir=md_cache,
    )

    # Sync-time janitor: purge headless or deleted items on backend startup
    try:
        all_known = (
            set(self.processor.list_cached_keys())
            | self.processor.get_all_meta_keys()
            | self.store.get_all_item_keys()
        )
        if all_known and self.db:
            active_keys = self.db.get_active_item_keys(list(all_known))
            if not active_keys:
                logger.warning("Skipping startup orphan purge: no active items found in database")
            else:
                orphans = all_known - active_keys
                if orphans:
                    logger.info("Cleaning up %d orphaned items during staleness sync...", len(orphans))
                    for o_key in orphans:
                        self.purge_item(o_key, regenerate_index=False)
                    self.processor.regenerate_root_index()
                    res["orphans_cleaned"] = len(orphans)
    except Exception as exc:
        logger.warning("Error checking orphans during staleness sync: %s", exc)

    return res

get_syncer

get_syncer()

Get the active LibrarySyncer instance. Raises RuntimeError if not registered.

Source code in zocrates/rag/service_manager.py

def get_syncer() -> "LibrarySyncer":
    """Get the active LibrarySyncer instance. Raises RuntimeError if not registered."""
    if _syncer is None:
        raise RuntimeError(
            "Zocrates backend services have not been registered/initialized yet. "
            "Please register your Zotero data directory first."
        )
    return _syncer

initialize_services

initialize_services(zotero_data_dir)

Initialize the LibrarySyncer and database connections.

Thread-safe: uses a lock to prevent concurrent initialization. Tears down the previous instance before creating a new one.

Source code in zocrates/rag/service_manager.py

def initialize_services(zotero_data_dir: Path) -> "LibrarySyncer":
    """Initialize the LibrarySyncer and database connections.

    Thread-safe: uses a lock to prevent concurrent initialization.
    Tears down the previous instance before creating a new one.
    """
    global _syncer
    with _init_lock:
        if _syncer is not None:
            logger.info("Tearing down previous LibrarySyncer instance.")
            try:
                _syncer.close()
            except Exception as exc:
                logger.warning("Error closing previous syncer: %s", exc)

        sqlite_path = zotero_data_dir / "zotero.sqlite"
        if not sqlite_path.exists():
            raise FileNotFoundError(f"zotero.sqlite not found at {sqlite_path}")

        logger.info(
            "Initializing Zocrates services with data directory: %s",
            zotero_data_dir,
        )
        from zocrates.utils.services import get_session_service
        get_session_service(zotero_data_dir)

        from zocrates.rag.library_syncer import LibrarySyncer

        _syncer = LibrarySyncer(zotero_data_dir=zotero_data_dir, sqlite_path=sqlite_path)

        # Ensure default zocrates_config.json is persisted if not present
        import copy
        from zocrates.config import (
            DEFAULT_OVERRIDE_AGENT_MODELS,
            DEFAULT_SEARCH_HARNESS,
            get_config_path,
            save_zocrates_config,
        )
        config_path = get_config_path(zotero_data_dir)
        if not config_path.exists():
            save_zocrates_config(
                {
                    "document_converter_id": "markitdown",
                    "document_converter_mode": "local",
                    "embedding_provider": "local",
                    "embedding_model": "jinaai/jina-embeddings-v5-text-nano",
                    "force_cpu": False,
                    "override_agent_models": copy.deepcopy(
                        DEFAULT_OVERRIDE_AGENT_MODELS
                    ),
                    "search_harness": copy.deepcopy(DEFAULT_SEARCH_HARNESS),
                },
                zotero_data_dir,
            )

        return _syncer