Skip to content

config

zocrates.config

Central configuration for the Zocrates RAG engine. All settings are read from environment variables with safe defaults.

ALLOW_ORIGINS module-attribute 

ALLOW_ORIGINS = [o.strip() for o in os.getenv('ALLOW_ORIGINS', '').split(',') if o.strip()] if os.getenv('ALLOW_ORIGINS') else list(DEFAULT_ALLOW_ORIGINS)

BACKEND_HOST module-attribute 

BACKEND_HOST = os.getenv('BACKEND_HOST', '127.0.0.1')

BACKEND_PORT module-attribute 

BACKEND_PORT = _resolve_backend_port()

BASE_BACKEND_PORT module-attribute 

BASE_BACKEND_PORT = 8765

CHUNK_MAX_CHARS module-attribute 

CHUNK_MAX_CHARS = int(os.getenv('CHUNK_MAX_CHARS', '4000'))

CHUNK_OVERLAP_CHARS module-attribute 

CHUNK_OVERLAP_CHARS = int(os.getenv('CHUNK_OVERLAP_CHARS', '200'))

CONVERSION_TIMEOUT_SECONDS module-attribute 

CONVERSION_TIMEOUT_SECONDS = int(os.getenv('CONVERSION_TIMEOUT_SECONDS', str(DEFAULT_CONVERSION_TIMEOUT_SECONDS)))

DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL module-attribute 

DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL = 'inception/mercury-decide'

DEFAULT_ADAPTIVE_COMPACTION_MIN_TURNS module-attribute 

DEFAULT_ADAPTIVE_COMPACTION_MIN_TURNS = 4

DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS module-attribute 

DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS = 2

DEFAULT_ADAPTIVE_COMPACTION_THRESHOLD module-attribute 

DEFAULT_ADAPTIVE_COMPACTION_THRESHOLD = 0.75

DEFAULT_ADAPTIVE_COMPACTION_TOKEN_CEILING module-attribute 

DEFAULT_ADAPTIVE_COMPACTION_TOKEN_CEILING = 100000

DEFAULT_ADAPTIVE_COMPACTION_TOKEN_FLOOR module-attribute 

DEFAULT_ADAPTIVE_COMPACTION_TOKEN_FLOOR = 30000

DEFAULT_ADAPTIVE_DECISION_MODEL module-attribute 

DEFAULT_ADAPTIVE_DECISION_MODEL = DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL

DEFAULT_ALLOW_ORIGINS module-attribute 

DEFAULT_ALLOW_ORIGINS = ['chrome://zocrates', 'moz-extension://*']

DEFAULT_AUTO_RENDER_MERMAID_DIAGRAMS module-attribute 

DEFAULT_AUTO_RENDER_MERMAID_DIAGRAMS = True

DEFAULT_AUTO_SPAWN_BACKEND module-attribute 

DEFAULT_AUTO_SPAWN_BACKEND = True

DEFAULT_BACKEND_PORT module-attribute 

DEFAULT_BACKEND_PORT = None

DEFAULT_CITATIONS_ENABLED module-attribute 

DEFAULT_CITATIONS_ENABLED = True

DEFAULT_CITATION_VERIFICATION_AUTO_START_LOCAL module-attribute 

DEFAULT_CITATION_VERIFICATION_AUTO_START_LOCAL = True

DEFAULT_CITATION_VERIFICATION_ENABLED module-attribute 

DEFAULT_CITATION_VERIFICATION_ENABLED = False

DEFAULT_CITATION_VERIFICATION_LOCAL_PORT module-attribute 

DEFAULT_CITATION_VERIFICATION_LOCAL_PORT = 8005

DEFAULT_CITATION_VERIFICATION_LOCAL_URL module-attribute 

DEFAULT_CITATION_VERIFICATION_LOCAL_URL = 'http://127.0.0.1:8005/v1/systemone'

DEFAULT_CITATION_VERIFICATION_MODEL module-attribute 

DEFAULT_CITATION_VERIFICATION_MODEL = '~typesafe/jev-latest'

DEFAULT_CITATION_VERIFICATION_PROVIDER module-attribute 

DEFAULT_CITATION_VERIFICATION_PROVIDER = 'openrouter'

DEFAULT_CITATION_VERIFICATION_THRESHOLD module-attribute 

DEFAULT_CITATION_VERIFICATION_THRESHOLD = 0.6

DEFAULT_CITATION_VERIFICATION_TIMEOUT_SEC module-attribute 

DEFAULT_CITATION_VERIFICATION_TIMEOUT_SEC = 10.0

DEFAULT_CITATION_VERIFICATION_TURN_BUDGET_SEC module-attribute 

DEFAULT_CITATION_VERIFICATION_TURN_BUDGET_SEC = 20.0

DEFAULT_COMPACTION_SUMMARIZER_MODEL module-attribute 

DEFAULT_COMPACTION_SUMMARIZER_MODEL = 'openrouter/~deepseek/deepseek-flash-latest'

DEFAULT_CONFIG module-attribute 

DEFAULT_CONFIG = {'llm_provider': 'openrouter_llm', 'llm_model': DEFAULT_LLM_MODEL, 'llm_base_url': DEFAULT_LLM_BASE_URL, 'llm_temperature': DEFAULT_LLM_TEMPERATURE, 'llm_max_tokens': DEFAULT_LLM_MAX_TOKENS, 'llm_num_retries': DEFAULT_LLM_NUM_RETRIES, 'ollama_num_ctx': DEFAULT_OLLAMA_NUM_CTX, 'override_agent_models': copy.deepcopy(DEFAULT_OVERRIDE_AGENT_MODELS), 'embedding_provider': 'local', 'embedding_model': 'jinaai/jina-embeddings-v5-text-nano', 'embedding_dimension': 768, 'document_converter_mode': 'local', 'document_converter_id': 'markitdown', 'docling_mode': 'auto', 'datalab_non_pdf_fallback': 'markitdown', 'datalab_merge_cross_page': DEFAULT_DATALAB_MERGE_CROSS_PAGE, 'conversion_timeout_seconds': DEFAULT_CONVERSION_TIMEOUT_SECONDS, 'enable_vlm_captioning': DEFAULT_ENABLE_VLM_CAPTIONING, 'vlm_captioning_provider': DEFAULT_VLM_CAPTIONING_PROVIDER, 'vlm_captioning_model': DEFAULT_VLM_CAPTIONING_MODEL, 'vlm_captioning_backup_model': DEFAULT_VLM_CAPTIONING_BACKUP_MODEL, 'vlm_captioning_local_model': DEFAULT_VLM_CAPTIONING_LOCAL_MODEL, 'vlm_max_images_per_doc': DEFAULT_VLM_MAX_IMAGES_PER_DOC, 'vlm_timeout_seconds': DEFAULT_VLM_TIMEOUT_SECONDS, 'semantic_similarity_threshold': DEFAULT_SEMANTIC_SIMILARITY_THRESHOLD, 'reranker_model': DEFAULT_RERANKER_MODEL, 'search_harness': copy.deepcopy(DEFAULT_SEARCH_HARNESS), 'graph_relations': copy.deepcopy(DEFAULT_GRAPH_RELATIONS), 'citations_enabled': DEFAULT_CITATIONS_ENABLED, 'references_section_enabled': DEFAULT_REFERENCES_SECTION_ENABLED, 'considered_sources_disclosure': DEFAULT_CONSIDERED_SOURCES_DISCLOSURE, 'auto_render_mermaid_diagrams': DEFAULT_AUTO_RENDER_MERMAID_DIAGRAMS, 'enable_mcp_server': DEFAULT_ENABLE_MCP_SERVER, 'auto_spawn_backend': DEFAULT_AUTO_SPAWN_BACKEND, 'enable_predictive_ghost_input': DEFAULT_ENABLE_PREDICTIVE_GHOST_INPUT, 'predictive_prompt_model': DEFAULT_PREDICTIVE_PROMPT_MODEL, 'sentence_attribution_model': DEFAULT_SENTENCE_ATTRIBUTION_MODEL, 'scholar_polite_email': DEFAULT_SCHOLAR_POLITE_EMAIL, 'user_research_profile': DEFAULT_USER_RESEARCH_PROFILE, 'max_research_profile_chars': DEFAULT_MAX_RESEARCH_PROFILE_CHARS, 'citation_verification_enabled': False, 'citation_verification_provider': 'openrouter', 'citation_verification_model': DEFAULT_CITATION_VERIFICATION_MODEL, 'citation_verification_threshold': 0.6, 'citation_verification_local_url': 'http://127.0.0.1:8005/v1/systemone', 'citation_verification_local_port': 8005, 'citation_verification_auto_start_local': True, 'citation_verification_timeout_sec': 10.0, 'citation_verification_turn_budget_sec': 20.0, 'enable_adaptive_compaction': DEFAULT_ENABLE_ADAPTIVE_COMPACTION, 'adaptive_compaction_retention_turns': DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS, 'adaptive_compaction_min_turns': DEFAULT_ADAPTIVE_COMPACTION_MIN_TURNS, 'adaptive_compaction_token_floor': DEFAULT_ADAPTIVE_COMPACTION_TOKEN_FLOOR, 'adaptive_compaction_token_ceiling': DEFAULT_ADAPTIVE_COMPACTION_TOKEN_CEILING, 'adaptive_compaction_threshold': DEFAULT_ADAPTIVE_COMPACTION_THRESHOLD, 'adaptive_compaction_decision_model': DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL, 'compaction_summarizer_model': DEFAULT_COMPACTION_SUMMARIZER_MODEL, 'max_saved_sessions': DEFAULT_MAX_SAVED_SESSIONS, 'session_retention_days': DEFAULT_SESSION_RETENTION_DAYS, 'session_db_auto_vacuum_on_startup': DEFAULT_SESSION_DB_AUTO_VACUUM_ON_STARTUP, 'prune_empty_sessions': DEFAULT_PRUNE_EMPTY_SESSIONS, 'backend_port': DEFAULT_BACKEND_PORT, 'bootstrap_handshake_timeout_ms': 300000, 'sync_auto_abort_failures': DEFAULT_SYNC_AUTO_ABORT_FAILURES}

DEFAULT_CONSIDERED_SOURCES_DISCLOSURE module-attribute 

DEFAULT_CONSIDERED_SOURCES_DISCLOSURE = True

DEFAULT_CONVERSION_TIMEOUT_SECONDS module-attribute 

DEFAULT_CONVERSION_TIMEOUT_SECONDS = 360

DEFAULT_DATALAB_MERGE_CROSS_PAGE module-attribute 

DEFAULT_DATALAB_MERGE_CROSS_PAGE = False

DEFAULT_ENABLE_ADAPTIVE_COMPACTION module-attribute 

DEFAULT_ENABLE_ADAPTIVE_COMPACTION = True

DEFAULT_ENABLE_MCP_SERVER module-attribute 

DEFAULT_ENABLE_MCP_SERVER = True

DEFAULT_ENABLE_PREDICTIVE_GHOST_INPUT module-attribute 

DEFAULT_ENABLE_PREDICTIVE_GHOST_INPUT = True

DEFAULT_ENABLE_VLM_CAPTIONING module-attribute 

DEFAULT_ENABLE_VLM_CAPTIONING = True

DEFAULT_GRAPH_RELATIONS module-attribute 

DEFAULT_GRAPH_RELATIONS = {'enabled': True, 'strategy': 'hybrid_corroborated', 'candidate_pool_size': 50, 'min_hybrid_score': 0.015, 'score_tau': 0.1, 'shared_tags_score': 0.7, 'include_human_curated': True, 'include_shared_tags': True, 'min_shared_tags': 2, 'sync_bundle_files': True}

DEFAULT_LLM_BASE_URL module-attribute 

DEFAULT_LLM_BASE_URL = ''

DEFAULT_LLM_MAX_TOKENS module-attribute 

DEFAULT_LLM_MAX_TOKENS = 16384

DEFAULT_LLM_MODEL module-attribute 

DEFAULT_LLM_MODEL = 'openrouter/~z-ai/glm-flash-latest'

DEFAULT_LLM_NUM_RETRIES module-attribute 

DEFAULT_LLM_NUM_RETRIES = 4

DEFAULT_LLM_TEMPERATURE module-attribute 

DEFAULT_LLM_TEMPERATURE = None

DEFAULT_MAX_RESEARCH_PROFILE_CHARS module-attribute 

DEFAULT_MAX_RESEARCH_PROFILE_CHARS = 1000

DEFAULT_MAX_SAVED_SESSIONS module-attribute 

DEFAULT_MAX_SAVED_SESSIONS = 100

DEFAULT_OLLAMA_NUM_CTX module-attribute 

DEFAULT_OLLAMA_NUM_CTX = 16384

DEFAULT_OVERRIDE_AGENT_MODELS module-attribute 

DEFAULT_OVERRIDE_AGENT_MODELS = {'coordinator': {'model': None, 'thinking_level': 'medium'}, 'library_search': {'model': None, 'thinking_level': 'low'}, 'library_synthesizer': {'model': None, 'thinking_level': 'high'}, 'online_search': {'model': None, 'thinking_level': 'low'}, 'online_synthesizer': {'model': None, 'thinking_level': 'high'}}

DEFAULT_PREDICTIVE_PROMPT_MODEL module-attribute 

DEFAULT_PREDICTIVE_PROMPT_MODEL = 'openrouter/~deepseek/deepseek-flash-latest'

DEFAULT_PRUNE_EMPTY_SESSIONS module-attribute 

DEFAULT_PRUNE_EMPTY_SESSIONS = True

DEFAULT_REFERENCES_SECTION_ENABLED module-attribute 

DEFAULT_REFERENCES_SECTION_ENABLED = True

DEFAULT_RERANKER_MODEL module-attribute 

DEFAULT_RERANKER_MODEL = 'qwen/qwen3-reranker-8b'

DEFAULT_SCHOLAR_POLITE_EMAIL module-attribute 

DEFAULT_SCHOLAR_POLITE_EMAIL = ''

DEFAULT_SEARCH_HARNESS module-attribute 

DEFAULT_SEARCH_HARNESS = {'max_retrieval_tools': 10, 'budget_extension_increment': 5, 'max_absolute_retrieval_limit': 20, 'max_extension_count': 2, 'hard_word_ceiling': 7500, 'word_ceiling_extension_increment': 3750, 'max_absolute_word_ceiling': 15000, 'word_ceiling_hitl_trigger_percent': 80, 'max_rerank_chunks': 15, 'rerank_chunks_extension_increment': 5, 'max_absolute_rerank_chunks': 25, 'hitl_enabled': True, 'max_injected_toc_items': 3, 'max_unique_cached_tocs': 10, 'online_initial_retrieval_tools': 8, 'max_online_extensions': 2, 'online_budget_increment': 5, 'max_absolute_online_tools': 18}

DEFAULT_SEMANTIC_SIMILARITY_THRESHOLD module-attribute 

DEFAULT_SEMANTIC_SIMILARITY_THRESHOLD = 0.5

DEFAULT_SENTENCE_ATTRIBUTION_MODEL module-attribute 

DEFAULT_SENTENCE_ATTRIBUTION_MODEL = ''

DEFAULT_SESSION_DB_AUTO_VACUUM_ON_STARTUP module-attribute 

DEFAULT_SESSION_DB_AUTO_VACUUM_ON_STARTUP = True

DEFAULT_SESSION_RETENTION_DAYS module-attribute 

DEFAULT_SESSION_RETENTION_DAYS = 180

DEFAULT_SYNC_AUTO_ABORT_FAILURES module-attribute 

DEFAULT_SYNC_AUTO_ABORT_FAILURES = 10

DEFAULT_USER_RESEARCH_PROFILE module-attribute 

DEFAULT_USER_RESEARCH_PROFILE = ''

DEFAULT_VLM_CAPTIONING_BACKUP_MODEL module-attribute 

DEFAULT_VLM_CAPTIONING_BACKUP_MODEL = 'openrouter/nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free'

DEFAULT_VLM_CAPTIONING_LOCAL_MODEL module-attribute 

DEFAULT_VLM_CAPTIONING_LOCAL_MODEL = ''

DEFAULT_VLM_CAPTIONING_MODEL module-attribute 

DEFAULT_VLM_CAPTIONING_MODEL = 'openrouter/deepseek/deepseek-v4.1-flash'

DEFAULT_VLM_CAPTIONING_PROVIDER module-attribute 

DEFAULT_VLM_CAPTIONING_PROVIDER = 'openrouter'

DEFAULT_VLM_MAX_IMAGES_PER_DOC module-attribute 

DEFAULT_VLM_MAX_IMAGES_PER_DOC = 100

DEFAULT_VLM_MIN_FILE_SIZE_BYTES module-attribute 

DEFAULT_VLM_MIN_FILE_SIZE_BYTES = 4096

DEFAULT_VLM_MIN_IMAGE_DIMENSION module-attribute 

DEFAULT_VLM_MIN_IMAGE_DIMENSION = 120

DEFAULT_VLM_TIMEOUT_SECONDS module-attribute 

DEFAULT_VLM_TIMEOUT_SECONDS = 30

DETERMINISTIC_PORT_SPAN module-attribute 

DETERMINISTIC_PORT_SPAN = 100

EMBEDDING_DIMENSION module-attribute 

EMBEDDING_DIMENSION = int(os.getenv('EMBEDDING_DIMENSION', '768'))

EMBEDDING_MODE module-attribute 

EMBEDDING_MODE = os.getenv('EMBEDDING_MODE', 'local')

GOOGLE_CLOUD_LOCATION module-attribute 

GOOGLE_CLOUD_LOCATION = os.getenv('GOOGLE_CLOUD_LOCATION', 'global')

LOCAL_EMBEDDING_MODEL module-attribute 

LOCAL_EMBEDDING_MODEL = os.getenv('LOCAL_EMBEDDING_MODEL', 'jinaai/jina-embeddings-v5-text-nano')

PROVIDER_DEFAULT_MODELS module-attribute 

PROVIDER_DEFAULT_MODELS = {'openrouter_llm': 'openrouter/~z-ai/glm-flash-latest', 'openrouter': 'openrouter/~z-ai/glm-flash-latest', 'ollama': 'llama3.1:8b', 'vllm': 'default', 'openai_compatible': 'default'}

RETRIEVAL_FINAL_K module-attribute 

RETRIEVAL_FINAL_K = int(os.getenv('RETRIEVAL_FINAL_K', '8'))

RETRIEVAL_TOP_K module-attribute 

RETRIEVAL_TOP_K = int(os.getenv('RETRIEVAL_TOP_K', '20'))

ZOCRATES_DATA_SUBDIR module-attribute 

ZOCRATES_DATA_SUBDIR = 'zocrates-data'

ZOTERO_DATA_DIR module-attribute 

ZOTERO_DATA_DIR = Path(v) if (v := os.getenv('ZOTERO_DATA_DIR')) else None

_CACHED_AUTH_TOKEN module-attribute 

_CACHED_AUTH_TOKEN = None

_CONFIG_CACHE module-attribute 

_CONFIG_CACHE = {}

_CONFIG_CACHE_FINGERPRINT module-attribute 

_CONFIG_CACHE_FINGERPRINT = {}

_CONFIG_CACHE_LOCK module-attribute 

_CONFIG_CACHE_LOCK = threading.Lock()

_CONFIG_CACHE_MTIME_NS module-attribute 

_CONFIG_CACHE_MTIME_NS = _CONFIG_CACHE_FINGERPRINT

_last_config_mtime module-attribute 

_last_config_mtime = 0.0

get_config module-attribute 

get_config = load_zocrates_config

get_markdown_cache_dir module-attribute 

get_markdown_cache_dir = get_corpus_dir

load_or_create_config module-attribute 

load_or_create_config = load_zocrates_config

logger module-attribute 

logger = logging.getLogger(__name__)

update_config module-attribute 

update_config = save_zocrates_config

_parse_config_backend_port 

_parse_config_backend_port(zotero_data_dir=None)

Read-only peek at backend_port in zocrates_config.json.

Unlike load_zocrates_config this performs no file-lock side effects, no default backfilling, and no disk writes — it is safe to call during module import. Returns the port only when it parses to an int > 0.

Source code in zocrates/config.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def _parse_config_backend_port(zotero_data_dir: Path | None = None) -> int | None:
    """Read-only peek at ``backend_port`` in ``zocrates_config.json``.

    Unlike ``load_zocrates_config`` this performs no file-lock side effects,
    no default backfilling, and no disk writes — it is safe to call during
    module import. Returns the port only when it parses to an int > 0.
    """
    if zotero_data_dir is not None:
        resolved = Path(zotero_data_dir)
    else:
        resolved = _resolve_import_time_zotero_data_dir()
    if resolved is None:
        return None
    config_path = resolved / ZOCRATES_DATA_SUBDIR / "zocrates_config.json"
    if not config_path.exists():
        return None
    try:
        with open(config_path) as f:
            raw = json.load(f)
    except (OSError, json.JSONDecodeError):
        return None
    if not isinstance(raw, dict):
        return None
    port = raw.get("backend_port", DEFAULT_BACKEND_PORT)
    if port is None:
        return None
    try:
        port_int = int(port)
    except (ValueError, TypeError):
        return None
    return port_int if port_int > 0 else None

_parse_env_backend_port 

_parse_env_backend_port()

Parse the BACKEND_PORT environment override, or return None.

Source code in zocrates/config.py 
83
84
85
86
87
88
89
90
91
def _parse_env_backend_port() -> int | None:
    """Parse the ``BACKEND_PORT`` environment override, or return ``None``."""
    val = os.getenv("BACKEND_PORT")
    if val:
        try:
            return int(val)
        except (ValueError, TypeError):
            pass
    return None

_resolve_backend_port 

_resolve_backend_port()

Resolve the backend port.

Precedence
  1. BACKEND_PORT environment variable (explicit override).
  2. backend_port (int > 0) in zocrates_config.json.
  3. Deterministic offset: compute_deterministic_port of the resolved zocrates-data directory, stepped forward via find_free_port to the first free loopback port.

When no Zotero data directory can be resolved, the plain 8765 base port is returned and NO port probing is performed, so module imports stay deterministic in test and scripting environments.

Source code in zocrates/config.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def _resolve_backend_port() -> int:
    """Resolve the backend port.

    Precedence:
      1. ``BACKEND_PORT`` environment variable (explicit override).
      2. ``backend_port`` (int > 0) in ``zocrates_config.json``.
      3. Deterministic offset: ``compute_deterministic_port`` of the resolved
         zocrates-data directory, stepped forward via ``find_free_port`` to
         the first free loopback port.

    When no Zotero data directory can be resolved, the plain 8765 base port is
    returned and NO port probing is performed, so module imports stay
    deterministic in test and scripting environments.
    """
    env_port = _parse_env_backend_port()
    if env_port is not None:
        return env_port
    cfg_port = _parse_config_backend_port()
    if cfg_port is not None:
        return cfg_port
    resolved = _resolve_import_time_zotero_data_dir()
    if resolved is None:
        return BASE_BACKEND_PORT
    return find_free_port(compute_deterministic_port(resolved / ZOCRATES_DATA_SUBDIR))

_resolve_explicit_zotero_data_dir 

_resolve_explicit_zotero_data_dir(zotero_data_dir=None)

Resolve Zotero data directory from explicit arg, active syncer, or ZOTERO_DATA_DIR env.

Source code in zocrates/config.py 
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
def _resolve_explicit_zotero_data_dir(zotero_data_dir: Path | str | None = None) -> Path | None:
    """Resolve Zotero data directory from explicit arg, active syncer, or ZOTERO_DATA_DIR env."""
    if zotero_data_dir is not None:
        return Path(zotero_data_dir)
    try:
        from zocrates.rag.service_manager import get_syncer

        syncer = get_syncer()
        syncer_dir = getattr(syncer, "_zotero_data_dir", None)
        if isinstance(syncer_dir, (str, Path)):
            return Path(syncer_dir)
    except Exception:
        pass
    if ZOTERO_DATA_DIR:
        return Path(ZOTERO_DATA_DIR)
    if env_dir := os.getenv("ZOTERO_DATA_DIR"):
        return Path(env_dir)
    return None

_resolve_import_time_zotero_data_dir 

_resolve_import_time_zotero_data_dir()

Best-effort Zotero data-dir resolution that is safe at module import time.

Deliberately avoids resolve_zotero_data_dir's LibrarySyncer branch: importing app.rag.service_manager while app.config itself is still initializing would be a circular import, and no syncer exists at import time anyway. Mirrors the ZOTERO_DATA_DIR env + home-profile discovery glue used by resolve_zotero_data_dir without any side effects.

Source code in zocrates/config.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def _resolve_import_time_zotero_data_dir() -> Path | None:
    """Best-effort Zotero data-dir resolution that is safe at module import time.

    Deliberately avoids ``resolve_zotero_data_dir``'s ``LibrarySyncer`` branch:
    importing ``app.rag.service_manager`` while ``app.config`` itself is still
    initializing would be a circular import, and no syncer exists at import
    time anyway. Mirrors the ``ZOTERO_DATA_DIR`` env + home-profile discovery
    glue used by ``resolve_zotero_data_dir`` without any side effects.
    """
    if env_dir := os.getenv("ZOTERO_DATA_DIR"):
        return Path(env_dir)
    candidates = (
        list(Path.home().glob("Zotero*/zocrates-data"))
        + list(Path.home().glob("Zotero*/*/zocrates-data"))
        + list(Path.home().glob(".var/app/org.zotero.Zotero/data/zotero/zocrates-data"))
        + list(Path.home().glob(".zotero/zotero/*/zocrates-data"))
    )
    if candidates:
        return candidates[0].parent
    return None

_write_auth_sidecar 

_write_auth_sidecar(auth_file, token)

Write auth token to sidecar file with 0o600 permissions (best-effort).

Source code in zocrates/config.py 
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
def _write_auth_sidecar(auth_file: Path, token: str) -> None:
    """Write auth token to sidecar file with 0o600 permissions (best-effort)."""
    try:
        auth_file.parent.mkdir(parents=True, exist_ok=True)
        fd = os.open(str(auth_file), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                f.write(token + "\n")
        except Exception:
            with contextlib.suppress(OSError):
                os.close(fd)
            raise
        with contextlib.suppress(OSError):
            os.chmod(auth_file, 0o600)
    except OSError as exc:
        logger.debug("Could not write backend.auth sidecar at %s: %s", auth_file, exc)

check_config_freshness 

check_config_freshness(zotero_data_dir=None)

Check if zocrates_config.json on disk has been modified since last check.

If _last_config_mtime is uninitialized (0.0), initializes it and returns False. If current mtime > _last_config_mtime, updates _last_config_mtime and returns True. Otherwise returns False. Handles OSError safely.

Source code in zocrates/config.py 
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
def check_config_freshness(zotero_data_dir: Path | str | None = None) -> bool:
    """Check if zocrates_config.json on disk has been modified since last check.

    If _last_config_mtime is uninitialized (0.0), initializes it and returns False.
    If current mtime > _last_config_mtime, updates _last_config_mtime and returns True.
    Otherwise returns False. Handles OSError safely.
    """
    global _last_config_mtime
    resolved_dir = resolve_zotero_data_dir(zotero_data_dir)
    if resolved_dir is None:
        return False
    config_path = get_config_path(resolved_dir)
    try:
        if not config_path.exists():
            return False
        current_mtime = config_path.stat().st_mtime
        if _last_config_mtime == 0.0:
            _last_config_mtime = current_mtime
            return False
        if current_mtime > _last_config_mtime:
            _last_config_mtime = current_mtime
            return True
        return False
    except OSError:
        return False

clear_config_cache 

clear_config_cache(zotero_data_dir=None)

Clear in-memory config cache for a specific data dir or globally.

Source code in zocrates/config.py 
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
def clear_config_cache(zotero_data_dir: Path | str | None = None) -> None:
    """Clear in-memory config cache for a specific data dir or globally."""
    resolved_path = None
    if zotero_data_dir is not None:
        resolved = resolve_zotero_data_dir(zotero_data_dir)
        if resolved is not None:
            resolved_path = get_config_path(resolved).resolve()
    with _CONFIG_CACHE_LOCK:
        if resolved_path is not None:
            _CONFIG_CACHE.pop(resolved_path, None)
            _CONFIG_CACHE_FINGERPRINT.pop(resolved_path, None)
        else:
            _CONFIG_CACHE.clear()
            _CONFIG_CACHE_FINGERPRINT.clear()

compute_deterministic_port 

compute_deterministic_port(data_dir)

Return the deterministic backend port for the given data directory.

Pinned formula — MUST stay byte-for-byte identical to the JavaScript implementation in zotero-plugin/addon/lib/portOffset.js so the Zotero frontend and the Python backend derive the same port independently::

8765 + (int.from_bytes(hashlib.sha256(str(data_dir).encode("utf-8")).digest()[:8], "big") % 100)

Yields a stable port in the 8765..8864 range per library path, preventing EADDRINUSE collisions across concurrent Zotero profiles.

Source code in zocrates/config.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def compute_deterministic_port(data_dir: Path) -> int:
    """Return the deterministic backend port for the given data directory.

    Pinned formula — MUST stay byte-for-byte identical to the JavaScript
    implementation in ``zotero-plugin/addon/lib/portOffset.js`` so the Zotero
    frontend and the Python backend derive the same port independently::

        8765 + (int.from_bytes(hashlib.sha256(str(data_dir).encode("utf-8")).digest()[:8], "big") % 100)

    Yields a stable port in the 8765..8864 range per library path, preventing
    ``EADDRINUSE`` collisions across concurrent Zotero profiles.
    """
    digest = hashlib.sha256(str(data_dir).encode("utf-8")).digest()
    return BASE_BACKEND_PORT + (int.from_bytes(digest[:8], "big") % DETERMINISTIC_PORT_SPAN)

delete_secure_api_key 

delete_secure_api_key(key_name='openrouter_api_key', env_var='OPENROUTER_API_KEY')

Delete a named API key from the session environment (os.environ).

Zotero Preferences / Login Manager is the secure persistent vault. Backend keys are in-memory session variables.

Source code in zocrates/config.py 
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
def delete_secure_api_key(
    key_name: str = "openrouter_api_key",
    env_var: str | None = "OPENROUTER_API_KEY",
) -> None:
    """Delete a named API key from the session environment (os.environ).

    Zotero Preferences / Login Manager is the secure persistent vault.
    Backend keys are in-memory session variables.
    """
    save_secure_api_key(None, key_name=key_name, env_var=env_var)

find_free_port 

find_free_port(start, span=100)

Return the first free loopback TCP port from start (inclusive).

Probes 127.0.0.1 with an outbound TCP connect (250ms timeout). If the candidate is occupied, steps forward up to span candidates (last probed candidate is start + span - 1). If every candidate is busy, returns start itself — the backend will surface the bind error.

Source code in zocrates/config.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def find_free_port(start: int, span: int = 100) -> int:
    """Return the first free loopback TCP port from ``start`` (inclusive).

    Probes ``127.0.0.1`` with an outbound TCP connect (250ms timeout). If the
    candidate is occupied, steps forward up to ``span`` candidates (last
    probed candidate is ``start + span - 1``). If every candidate is busy,
    returns ``start`` itself — the backend will surface the bind error.
    """
    for candidate in range(start, start + span):
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
                probe.settimeout(0.25)
                if probe.connect_ex(("127.0.0.1", candidate)) != 0:
                    return candidate
        except OSError:
            # A failed probe (e.g. timeout) cannot confirm availability;
            # treat the candidate as occupied and keep stepping.
            continue
    return start

get_auth_token 

get_auth_token(zotero_data_dir=None)

Resolve or generate the local backend authentication token.

Resolution order
  1. ZOCRATES_AUTH_TOKEN environment variable.
  2. Sidecar file <zocrates-data>/backend.auth.
  3. Cached in-memory token, or newly generated 256-bit hex token (secrets.token_hex(32)) persisted to backend.auth with mode 0o600.
Source code in zocrates/config.py 
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
def get_auth_token(zotero_data_dir: Path | str | None = None) -> str:
    """Resolve or generate the local backend authentication token.

    Resolution order:
      1. ``ZOCRATES_AUTH_TOKEN`` environment variable.
      2. Sidecar file ``<zocrates-data>/backend.auth``.
      3. Cached in-memory token, or newly generated 256-bit hex token
         (``secrets.token_hex(32)``) persisted to ``backend.auth`` with mode ``0o600``.
    """
    global _CACHED_AUTH_TOKEN

    env_token = os.getenv("ZOCRATES_AUTH_TOKEN", "").strip()
    if env_token:
        if zotero_data_dir is not None:
            auth_file = Path(zotero_data_dir) / ZOCRATES_DATA_SUBDIR / "backend.auth"
            if not auth_file.exists():
                _write_auth_sidecar(auth_file, env_token)
        return env_token

    if zotero_data_dir is not None:
        resolved_dir = Path(zotero_data_dir)
        auth_file = resolved_dir / ZOCRATES_DATA_SUBDIR / "backend.auth"
        try:
            if auth_file.is_file():
                file_token = auth_file.read_text(encoding="utf-8").strip()
                if file_token:
                    _CACHED_AUTH_TOKEN = file_token
                    return file_token
        except OSError:
            pass
        token = _CACHED_AUTH_TOKEN or secrets.token_hex(32)
        _write_auth_sidecar(auth_file, token)
        _CACHED_AUTH_TOKEN = token
        return token

    if _CACHED_AUTH_TOKEN:
        return _CACHED_AUTH_TOKEN

    resolved_dir = _resolve_explicit_zotero_data_dir(None)
    if resolved_dir is not None:
        auth_file = resolved_dir / ZOCRATES_DATA_SUBDIR / "backend.auth"
        try:
            if auth_file.is_file():
                file_token = auth_file.read_text(encoding="utf-8").strip()
                if file_token:
                    _CACHED_AUTH_TOKEN = file_token
                    return file_token
        except OSError:
            pass
        token = secrets.token_hex(32)
        _write_auth_sidecar(auth_file, token)
        _CACHED_AUTH_TOKEN = token
        return token

    _CACHED_AUTH_TOKEN = secrets.token_hex(32)
    return _CACHED_AUTH_TOKEN

get_auto_render_mermaid_diagrams 

get_auto_render_mermaid_diagrams(zotero_data_dir=None, cfg=None)

Return whether auto-rendering Mermaid diagrams is enabled from zocrates_config.json.

Source code in zocrates/config.py 
1756
1757
1758
1759
1760
1761
1762
1763
def get_auto_render_mermaid_diagrams(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether auto-rendering Mermaid diagrams is enabled from zocrates_config.json."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("auto_render_mermaid_diagrams", DEFAULT_AUTO_RENDER_MERMAID_DIAGRAMS))

get_auto_spawn_backend 

get_auto_spawn_backend(zotero_data_dir=None, cfg=None)

Return whether the Zotero bootstrap may auto-provision/spawn a managed backend.

Source code in zocrates/config.py 
1766
1767
1768
1769
1770
1771
1772
1773
1774
def get_auto_spawn_backend(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether the Zotero bootstrap may auto-provision/spawn a managed backend."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    val = cfg.get("auto_spawn_backend")
    return bool(val if val is not None else DEFAULT_AUTO_SPAWN_BACKEND)

get_backend_port 

get_backend_port(zotero_data_dir=None, cfg=None)

Return configured backend port from environment or zocrates_config.json, or None if dynamic.

Delegates to the shared env/config parse helpers used by _resolve_backend_port (env BACKEND_PORT > config "backend_port") instead of duplicating the parsing logic. None means the dynamic deterministic offset tier applies (see compute_deterministic_port).

Source code in zocrates/config.py 
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
def get_backend_port(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> int | None:
    """Return configured backend port from environment or zocrates_config.json, or None if dynamic.

    Delegates to the shared env/config parse helpers used by
    ``_resolve_backend_port`` (env BACKEND_PORT > config "backend_port")
    instead of duplicating the parsing logic. ``None`` means the dynamic
    deterministic offset tier applies (see ``compute_deterministic_port``).
    """
    env_port = _parse_env_backend_port()
    if env_port is not None:
        return env_port
    if cfg is not None:
        port = cfg.get("backend_port")
        if port is None:
            return None
        try:
            port_int = int(port)
            return port_int if port_int > 0 else None
        except (ValueError, TypeError):
            return None
    return _parse_config_backend_port(zotero_data_dir)

get_chroma_dir 

get_chroma_dir(zotero_data_dir)

Return (and create if needed) the ChromaDB storage directory.

Parameters:

Name Type Description Default
zotero_data_dir Path

Root Zotero data directory.

 required

Returns:

Type Description
Path

Path to <zocrates_data>/chroma/, created if absent.
Source code in zocrates/config.py 
607
608
609
610
611
612
613
614
615
616
617
618
619
def get_chroma_dir(zotero_data_dir: Path) -> Path:
    """Return (and create if needed) the ChromaDB storage directory.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Path to ``<zocrates_data>/chroma/``, created if absent.
    """
    zocrates_dir = get_zocrates_data_dir(zotero_data_dir)
    chroma_dir = zocrates_dir / "chroma"
    chroma_dir.mkdir(parents=True, exist_ok=True)
    return chroma_dir

get_citation_verification_enabled 

get_citation_verification_enabled(zotero_data_dir=None, cfg=None)

Return whether citation verification via decision models is enabled from zocrates_config.json.

Source code in zocrates/config.py 
1787
1788
1789
1790
1791
1792
1793
1794
def get_citation_verification_enabled(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether citation verification via decision models is enabled from zocrates_config.json."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("citation_verification_enabled", DEFAULT_CITATION_VERIFICATION_ENABLED))

get_citations_enabled 

get_citations_enabled(zotero_data_dir=None, cfg=None)

Return whether in-text citations are enabled from zocrates_config.json.

Source code in zocrates/config.py 
1726
1727
1728
1729
1730
1731
1732
1733
def get_citations_enabled(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether in-text citations are enabled from zocrates_config.json."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("citations_enabled", DEFAULT_CITATIONS_ENABLED))

get_collection_name 

get_collection_name(model_name, dimension)

Generate a sanitized, versioned ChromaDB collection name from model and dimension.

Replaces any non-alphanumeric characters with underscores, collapses consecutive underscores, and appends the dimension suffix.

Example::

get_collection_name("jinaai/jina-embeddings-v5-text-nano", 768)
# -> 'zotero_library_jinaai_jina_embeddings_v5_text_nano_768d'

Parameters:

Name Type Description Default
model_name str

Hugging Face or Gemini model identifier.

 required
dimension int

Embedding vector dimension.

 required

Returns:

Type Description
str

A lowercase, underscore-delimited collection name string.
Source code in zocrates/config.py 
2123
2124
2125
2126
2127
2128
2129
2130
2131
2132
2133
2134
2135
2136
2137
2138
2139
2140
2141
2142
2143
2144
def get_collection_name(model_name: str, dimension: int) -> str:
    """
    Generate a sanitized, versioned ChromaDB collection name from model and dimension.

    Replaces any non-alphanumeric characters with underscores, collapses
    consecutive underscores, and appends the dimension suffix.

    Example::

        get_collection_name("jinaai/jina-embeddings-v5-text-nano", 768)
        # -> 'zotero_library_jinaai_jina_embeddings_v5_text_nano_768d'

    Args:
        model_name: Hugging Face or Gemini model identifier.
        dimension: Embedding vector dimension.

    Returns:
        A lowercase, underscore-delimited collection name string.
    """
    sanitized_model = re.sub(r"[^a-zA-Z0-9_]", "_", model_name.strip())
    sanitized_model = re.sub(r"_+", "_", sanitized_model).strip("_")
    return f"zotero_library_{sanitized_model}_{dimension}d"

get_compaction_config 

get_compaction_config(zotero_data_dir=None, cfg=None)

Return adaptive context compaction settings from zocrates_config.json.

Source code in zocrates/config.py 
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
def get_compaction_config(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Return adaptive context compaction settings from zocrates_config.json."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return {
        "enable_adaptive_compaction": bool(
            cfg.get("enable_adaptive_compaction")
            if cfg.get("enable_adaptive_compaction") is not None
            else DEFAULT_ENABLE_ADAPTIVE_COMPACTION
        ),
        "adaptive_compaction_retention_turns": int(
            cfg.get("adaptive_compaction_retention_turns")
            if cfg.get("adaptive_compaction_retention_turns") is not None
            else DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS
        ),
        "adaptive_compaction_min_turns": int(
            cfg.get("adaptive_compaction_min_turns")
            if cfg.get("adaptive_compaction_min_turns") is not None
            else DEFAULT_ADAPTIVE_COMPACTION_MIN_TURNS
        ),
        "adaptive_compaction_token_floor": int(
            cfg.get("adaptive_compaction_token_floor")
            if cfg.get("adaptive_compaction_token_floor") is not None
            else DEFAULT_ADAPTIVE_COMPACTION_TOKEN_FLOOR
        ),
        "adaptive_compaction_token_ceiling": int(
            cfg.get("adaptive_compaction_token_ceiling")
            if cfg.get("adaptive_compaction_token_ceiling") is not None
            else DEFAULT_ADAPTIVE_COMPACTION_TOKEN_CEILING
        ),
        "adaptive_compaction_threshold": float(
            cfg.get("adaptive_compaction_threshold")
            if cfg.get("adaptive_compaction_threshold") is not None
            else DEFAULT_ADAPTIVE_COMPACTION_THRESHOLD
        ),
        "adaptive_compaction_decision_model": str(
            cfg.get("adaptive_compaction_decision_model")
            if cfg.get("adaptive_compaction_decision_model") is not None
            else (cfg.get("adaptive_decision_model") or DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL)
        ),
        "compaction_summarizer_model": str(
            cfg.get("compaction_summarizer_model")
            if cfg.get("compaction_summarizer_model") is not None
            else DEFAULT_COMPACTION_SUMMARIZER_MODEL
        ).strip(),
    }

get_config_path 

get_config_path(zotero_data_dir)

Return the path to the Zocrates config JSON file.

Parameters:

Name Type Description Default
zotero_data_dir Path

Root Zotero data directory.

 required

Returns:

Type Description
Path

Path to <zocrates_data>/zocrates_config.json.
Source code in zocrates/config.py 
649
650
651
652
653
654
655
656
657
658
def get_config_path(zotero_data_dir: Path) -> Path:
    """Return the path to the Zocrates config JSON file.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Path to ``<zocrates_data>/zocrates_config.json``.
    """
    return get_zocrates_data_dir(zotero_data_dir) / "zocrates_config.json"

get_considered_sources_disclosure 

get_considered_sources_disclosure(zotero_data_dir=None, cfg=None)

Return whether considered sources disclosure is enabled from zocrates_config.json.

Source code in zocrates/config.py 
1746
1747
1748
1749
1750
1751
1752
1753
def get_considered_sources_disclosure(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether considered sources disclosure is enabled from zocrates_config.json."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("considered_sources_disclosure", DEFAULT_CONSIDERED_SOURCES_DISCLOSURE))

get_conversion_timeout_seconds 

get_conversion_timeout_seconds(zotero_data_dir=None, cfg=None)

Return the timeout in seconds for document conversion operations.

Checks CONVERSION_TIMEOUT_SECONDS environment variable first; if unset, falls back to conversion_timeout_seconds in zocrates_config.json (defaulting to 360s to accommodate CPU-based layout inference in Docling).

Source code in zocrates/config.py 
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
def get_conversion_timeout_seconds(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> int:
    """Return the timeout in seconds for document conversion operations.

    Checks ``CONVERSION_TIMEOUT_SECONDS`` environment variable first;
    if unset, falls back to ``conversion_timeout_seconds`` in ``zocrates_config.json``
    (defaulting to 360s to accommodate CPU-based layout inference in Docling).
    """
    env_val = os.getenv("CONVERSION_TIMEOUT_SECONDS")
    if env_val is not None:
        try:
            return max(30, int(env_val))
        except ValueError:
            pass
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    try:
        return max(
            30,
            int(
                cfg.get(
                    "conversion_timeout_seconds",
                    DEFAULT_CONVERSION_TIMEOUT_SECONDS,
                )
            ),
        )
    except (ValueError, TypeError):
        return DEFAULT_CONVERSION_TIMEOUT_SECONDS

get_coordinator_model 

get_coordinator_model(cfg=None, zotero_data_dir=None)

Return the resolved coordinator model identifier.

Checks coordinator_agent.model at runtime if available; otherwise falls back to (cfg.get("override_agent_models") or {}).get("coordinator", {}).get("model"), then cfg.get("llm_model"), then DEFAULT_LLM_MODEL. Resolves with _resolve_model_name(coord_model, provider).

Source code in zocrates/config.py 
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
2093
2094
2095
2096
2097
def get_coordinator_model(
    cfg: dict[str, Any] | None = None,
    zotero_data_dir: Path | str | None = None,
) -> str:
    """Return the resolved coordinator model identifier.

    Checks coordinator_agent.model at runtime if available; otherwise falls back to
    (cfg.get("override_agent_models") or {}).get("coordinator", {}).get("model"),
    then cfg.get("llm_model"), then DEFAULT_LLM_MODEL. Resolves with
    _resolve_model_name(coord_model, provider).
    """
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)

    coord_model = None
    try:
        from zocrates.agents.app import coordinator_agent

        if coordinator_agent is not None and getattr(coordinator_agent, "model", None) is not None:
            c_model = coordinator_agent.model
            if isinstance(c_model, str) and c_model.strip():
                coord_model = c_model.strip()
            elif hasattr(c_model, "model") and isinstance(c_model.model, str) and c_model.model.strip():
                coord_model = c_model.model.strip()
            elif hasattr(c_model, "model") and c_model.model:
                coord_model = str(c_model.model).strip()
    except Exception:
        pass

    if not coord_model:
        override_entry = (cfg.get("override_agent_models") or {}).get("coordinator")
        coord_override = None
        if isinstance(override_entry, dict):
            coord_override = override_entry.get("model")
        elif isinstance(override_entry, str):
            coord_override = override_entry

        if coord_override:
            coord_model = str(coord_override).strip()
        elif cfg.get("llm_model"):
            coord_model = str(cfg.get("llm_model")).strip()
        else:
            coord_model = DEFAULT_LLM_MODEL

    provider = cfg.get("llm_provider") or "openrouter_llm"
    from zocrates.llm import _resolve_model_name

    return _resolve_model_name(coord_model, provider)

get_corpus_dir 

get_corpus_dir(zotero_data_dir)

Return (and create if needed) the document corpus directory.

Parameters:

Name Type Description Default
zotero_data_dir Path

Root Zotero data directory.

 required

Returns:

Type Description
Path

Path to <zocrates_data>/corpus/, created if absent.
Source code in zocrates/config.py 
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
def get_corpus_dir(zotero_data_dir: Path) -> Path:
    """Return (and create if needed) the document corpus directory.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Path to ``<zocrates_data>/corpus/``, created if absent.
    """
    zocrates_dir = get_zocrates_data_dir(zotero_data_dir)
    legacy_dir = zocrates_dir / "markdown_cache"
    corpus_dir = zocrates_dir / "corpus"
    if legacy_dir.is_dir() and not corpus_dir.exists():
        try:
            legacy_dir.rename(corpus_dir)
            logger.info("Migrated legacy markdown_cache to %s", corpus_dir)
        except OSError as exc:
            logger.warning(
                "Failed to automatically migrate legacy markdown_cache to corpus: %s. "
                "Please rename '%s' to '%s' manually.",
                exc,
                legacy_dir,
                corpus_dir,
            )
    elif legacy_dir.is_dir() and corpus_dir.is_dir():
        logger.debug("Both corpus/ and legacy markdown_cache/ exist. Using %s.", corpus_dir)
    corpus_dir.mkdir(parents=True, exist_ok=True)
    return corpus_dir

get_datalab_api_key 

get_datalab_api_key(zotero_data_dir=None)

Return the Datalab API key from environment variables.

Zotero Preferences / Login Manager is the authoritative persistent vault for the Datalab API key. The backend stores the key in-memory (synced from Zotero via environment variable DATALAB_API_KEY on startup or POST /config at runtime).

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Optional path to the Zotero data directory.

 None

Returns:

Type Description
str | None

The API key string or None if not configured.
Source code in zocrates/config.py 
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
def get_datalab_api_key(zotero_data_dir: Path | None = None) -> str | None:
    """Return the Datalab API key from environment variables.

    Zotero Preferences / Login Manager is the authoritative persistent vault for
    the Datalab API key. The backend stores the key in-memory (synced from Zotero
    via environment variable DATALAB_API_KEY on startup or POST /config at runtime).

    Args:
        zotero_data_dir: Optional path to the Zotero data directory.

    Returns:
        The API key string or None if not configured.
    """
    return get_secure_api_key(
        "datalab_api_key",
        env_vars=["DATALAB_API_KEY", "CHANDRA_API_KEY"],
    )

get_datalab_merge_cross_page 

get_datalab_merge_cross_page(zotero_data_dir=None, cfg=None)

Return whether Datalab cross-page table and paragraph merging is enabled.

Checks DATALAB_MERGE_CROSS_PAGE environment variable first (evaluating truthiness: "1", "true", "yes", "on"); if unset, falls back to in-memory cfg if provided, otherwise loads from zocrates_config.json (defaulting to False).

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
bool

True if enabled, False otherwise.
Source code in zocrates/config.py 
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
def get_datalab_merge_cross_page(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether Datalab cross-page table and paragraph merging is enabled.

    Checks ``DATALAB_MERGE_CROSS_PAGE`` environment variable first (evaluating
    truthiness: "1", "true", "yes", "on"); if unset, falls back to in-memory
    cfg if provided, otherwise loads from ``zocrates_config.json``
    (defaulting to ``False``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        ``True`` if enabled, ``False`` otherwise.
    """
    env_val = os.getenv("DATALAB_MERGE_CROSS_PAGE")
    if env_val is not None:
        return env_val.strip().lower() in ("1", "true", "yes", "on")
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("datalab_merge_cross_page", DEFAULT_DATALAB_MERGE_CROSS_PAGE))

get_datalab_non_pdf_fallback 

get_datalab_non_pdf_fallback(zotero_data_dir=None, cfg=None)

Return the fallback converter ID for non-PDF items when Datalab is active ('markitdown' or 'docling_local').

Checks DATALAB_NON_PDF_FALLBACK environment variable first; if unset, falls back to datalab_non_pdf_fallback in zocrates_config.json (defaulting to "markitdown").

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
str

Fallback converter ID string ("markitdown" or "docling_local").
Source code in zocrates/config.py 
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
def get_datalab_non_pdf_fallback(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the fallback converter ID for non-PDF items when Datalab is active ('markitdown' or 'docling_local').

    Checks ``DATALAB_NON_PDF_FALLBACK`` environment variable first;
    if unset, falls back to ``datalab_non_pdf_fallback`` in ``zocrates_config.json``
    (defaulting to ``"markitdown"``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Fallback converter ID string (``"markitdown"`` or ``"docling_local"``).
    """
    env_val = os.getenv("DATALAB_NON_PDF_FALLBACK", "")
    if env_val:
        return env_val.lower()
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return cfg.get("datalab_non_pdf_fallback", "markitdown").lower()

get_docling_mode 

get_docling_mode(zotero_data_dir=None, cfg=None)

Return the Docling operating mode ('auto', 'fast', or 'full').

Checks DOCUMENT_CONVERTER_DOCLING_MODE or DOCLING_MODE environment variable first; if unset, falls back to docling_mode in zocrates_config.json (defaulting to "auto").

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
str

Docling mode string ("auto", "fast", or "full").
Source code in zocrates/config.py 
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
def get_docling_mode(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the Docling operating mode ('auto', 'fast', or 'full').

    Checks ``DOCUMENT_CONVERTER_DOCLING_MODE`` or ``DOCLING_MODE`` environment
    variable first; if unset, falls back to ``docling_mode`` in ``zocrates_config.json``
    (defaulting to ``"auto"``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Docling mode string (``"auto"``, ``"fast"``, or ``"full"``).
    """
    env_val = os.getenv("DOCUMENT_CONVERTER_DOCLING_MODE") or os.getenv("DOCLING_MODE")
    if env_val:
        return env_val.lower()
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return cfg.get("docling_mode", "auto").lower()

get_document_converter_id 

get_document_converter_id(zotero_data_dir=None, cfg=None)

Return the document converter ID from the Zocrates config.

Checks DOCUMENT_CONVERTER_ID environment variable first; if unset, falls back to document_converter_id in zocrates_config.json (defaulting to "markitdown").

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
str

Converter ID string (e.g., "markitdown", "docling_local").
Source code in zocrates/config.py 
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
def get_document_converter_id(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the document converter ID from the Zocrates config.

    Checks ``DOCUMENT_CONVERTER_ID`` environment variable first;
    if unset, falls back to ``document_converter_id`` in ``zocrates_config.json``
    (defaulting to ``"markitdown"``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Converter ID string (e.g., ``"markitdown"``, ``"docling_local"``).
    """
    env_val = os.getenv("DOCUMENT_CONVERTER_ID", "")
    if env_val:
        return env_val
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return cfg.get("document_converter_id", "markitdown")

get_document_converter_mode 

get_document_converter_mode(zotero_data_dir=None, cfg=None)

Return the document converter mode ('local' or 'cloud').

Checks the DOCUMENT_CONVERTER_MODE environment variable first; if unset, falls back to the document_converter_mode in zocrates_config.json (defaulting to "local").

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
str

Mode string ("local" or "cloud").
Source code in zocrates/config.py 
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
def get_document_converter_mode(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the document converter mode ('local' or 'cloud').

    Checks the ``DOCUMENT_CONVERTER_MODE`` environment variable first;
    if unset, falls back to the ``document_converter_mode`` in ``zocrates_config.json``
    (defaulting to ``"local"``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Mode string (``"local"`` or ``"cloud"``).
    """
    env_val = os.getenv("DOCUMENT_CONVERTER_MODE", "")
    if env_val:
        return env_val.lower()
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return cfg.get("document_converter_mode", "local")

get_enable_vlm_captioning 

get_enable_vlm_captioning(zotero_data_dir=None, cfg=None)

Return whether VLM image captioning and figure enrichment is enabled.

Checks ENABLE_VLM_CAPTIONING environment variable first; if unset, falls back to enable_vlm_captioning in zocrates_config.json (defaulting to True).

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
bool

True if enabled, False otherwise.
Source code in zocrates/config.py 
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
def get_enable_vlm_captioning(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether VLM image captioning and figure enrichment is enabled.

    Checks ``ENABLE_VLM_CAPTIONING`` environment variable first;
    if unset, falls back to ``enable_vlm_captioning`` in ``zocrates_config.json``
    (defaulting to ``True``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        ``True`` if enabled, ``False`` otherwise.
    """
    env_val = os.getenv("ENABLE_VLM_CAPTIONING")
    if env_val is not None:
        return env_val.lower() in ("true", "1", "yes")
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("enable_vlm_captioning", DEFAULT_ENABLE_VLM_CAPTIONING))

get_force_cpu_pref 

get_force_cpu_pref(zotero_data_dir=None, cfg=None)

Return whether CPU-only execution mode is forced for local embeddings.

Checks the FORCE_CPU environment variable first; if unset or falsy, falls back to the force_cpu value in the persisted Zocrates config file.

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory. If None, only the environment variable is consulted and the config file is not read.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
bool

True if CPU-only mode should be forced, False otherwise.
Source code in zocrates/config.py 
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
def get_force_cpu_pref(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether CPU-only execution mode is forced for local embeddings.

    Checks the ``FORCE_CPU`` environment variable first; if unset or falsy,
    falls back to the ``force_cpu`` value in the persisted Zocrates config file.

    Args:
        zotero_data_dir: Path to the Zotero data directory. If ``None``,
            only the environment variable is consulted and the config file
            is not read.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        ``True`` if CPU-only mode should be forced, ``False`` otherwise.
    """
    val = os.getenv("FORCE_CPU", "")
    if val.lower() in ("true", "1", "yes"):
        return True
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("force_cpu", False))

get_fts_db_path 

get_fts_db_path(zotero_data_dir)

Return the path to the FTS5 SQLite database.

Parameters:

Name Type Description Default
zotero_data_dir Path

Root Zotero data directory.

 required

Returns:

Type Description
Path

Path to <zocrates_data>/fts_index.db.
Source code in zocrates/config.py 
622
623
624
625
626
627
628
629
630
631
632
def get_fts_db_path(zotero_data_dir: Path) -> Path:
    """Return the path to the FTS5 SQLite database.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Path to ``<zocrates_data>/fts_index.db``.
    """
    zocrates_dir = get_zocrates_data_dir(zotero_data_dir)
    return zocrates_dir / "fts_index.db"

get_graph_relations_config 

get_graph_relations_config(zotero_data_dir=None, cfg=None)

Return the graph relations configuration dict from zocrates_config.json merged with defaults.

Parameters:

Name Type Description Default
zotero_data_dir Path | str | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
dict[str, Any]

Dict containing graph relations parameters.
Source code in zocrates/config.py 
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
def get_graph_relations_config(
    zotero_data_dir: Path | str | None = None,
    cfg: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Return the graph relations configuration dict from zocrates_config.json merged with defaults.

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Dict containing graph relations parameters.
    """
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    res = copy.deepcopy(DEFAULT_GRAPH_RELATIONS)
    graph_cfg = cfg.get("graph_relations")
    if isinstance(graph_cfg, dict):
        res.update(graph_cfg)
    return res

get_log_dir 

get_log_dir(zotero_data_dir)

Return (and create if needed) the Zocrates log directory.

Parameters:

Name Type Description Default
zotero_data_dir Path

Root Zotero data directory.

 required

Returns:

Type Description
Path

Path to <zocrates_data>/logs/, created if absent.
Source code in zocrates/config.py 
635
636
637
638
639
640
641
642
643
644
645
646
def get_log_dir(zotero_data_dir: Path) -> Path:
    """Return (and create if needed) the Zocrates log directory.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Path to ``<zocrates_data>/logs/``, created if absent.
    """
    log_dir = get_zocrates_data_dir(zotero_data_dir) / "logs"
    log_dir.mkdir(parents=True, exist_ok=True)
    return log_dir

get_max_saved_sessions 

get_max_saved_sessions(zotero_data_dir=None, cfg=None)

Return maximum number of saved sessions to retain, or None if disabled/unlimited.

Source code in zocrates/config.py 
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
def get_max_saved_sessions(
    zotero_data_dir: Path | str | None = None,
    cfg: dict[str, Any] | None = None,
) -> int | None:
    """Return maximum number of saved sessions to retain, or None if disabled/unlimited."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    val = cfg.get("max_saved_sessions", DEFAULT_MAX_SAVED_SESSIONS)
    if val is None:
        return None
    try:
        return int(val)
    except (ValueError, TypeError):
        return DEFAULT_MAX_SAVED_SESSIONS

get_ollama_num_ctx 

get_ollama_num_ctx(zotero_data_dir=None, cfg=None)

Return configured Ollama context window size (num_ctx), defaulting to 16384.

Source code in zocrates/config.py 
2034
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045
2046
2047
def get_ollama_num_ctx(
    zotero_data_dir: Path | str | None = None,
    cfg: dict[str, Any] | None = None,
) -> int:
    """Return configured Ollama context window size (num_ctx), defaulting to 16384."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    val = cfg.get("ollama_num_ctx")
    if val is not None:
        try:
            return int(val)
        except (ValueError, TypeError):
            pass
    return DEFAULT_OLLAMA_NUM_CTX

get_openalex_api_key 

get_openalex_api_key(zotero_data_dir=None)

Retrieve the configured OpenAlex API key from the environment.

Source code in zocrates/config.py 
1308
1309
1310
1311
1312
1313
def get_openalex_api_key(zotero_data_dir: Path | None = None) -> str | None:
    """Retrieve the configured OpenAlex API key from the environment."""
    return get_secure_api_key(
        "openalex_api_key",
        env_vars=["OPENALEX_API_KEY"],
    )

get_prune_empty_sessions 

get_prune_empty_sessions(zotero_data_dir=None, cfg=None)

Return whether automatic pruning of abandoned empty sessions is enabled.

Source code in zocrates/config.py 
1941
1942
1943
1944
1945
1946
1947
1948
def get_prune_empty_sessions(
    zotero_data_dir: Path | str | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether automatic pruning of abandoned empty sessions is enabled."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("prune_empty_sessions", DEFAULT_PRUNE_EMPTY_SESSIONS))

get_references_section_enabled 

get_references_section_enabled(zotero_data_dir=None, cfg=None)

Return whether references section is enabled from zocrates_config.json.

Source code in zocrates/config.py 
1736
1737
1738
1739
1740
1741
1742
1743
def get_references_section_enabled(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether references section is enabled from zocrates_config.json."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("references_section_enabled", DEFAULT_REFERENCES_SECTION_ENABLED))

get_scholar_polite_email 

get_scholar_polite_email(zotero_data_dir=None, cfg=None)

Return configured email for scholarly discovery APIs polite pool.

Source code in zocrates/config.py 
1797
1798
1799
1800
1801
1802
1803
1804
1805
def get_scholar_polite_email(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return configured email for scholarly discovery APIs polite pool."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    val = cfg.get("scholar_polite_email", DEFAULT_SCHOLAR_POLITE_EMAIL)
    return str(val).strip() if val else ""

get_search_harness_config 

get_search_harness_config(zotero_data_dir=None, cfg=None)

Return the search harness configuration dict from zocrates_config.json merged with defaults.

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
dict[str, Any]

Dict containing search harness parameters (budget, limits, ceilings, HITL flags).
Source code in zocrates/config.py 
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
def get_search_harness_config(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Return the search harness configuration dict from zocrates_config.json merged with defaults.

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Dict containing search harness parameters (budget, limits, ceilings, HITL flags).
    """
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    res = copy.deepcopy(DEFAULT_SEARCH_HARNESS)
    harness_cfg = cfg.get("search_harness")
    if isinstance(harness_cfg, dict):
        res.update(harness_cfg)
    return res

get_secure_api_key 

get_secure_api_key(key_name='openrouter_api_key', env_vars=('OPENROUTER_API_KEY', 'OPEN_ROUTER_API_KEY'))

Load a named API key from in-memory session environment variables.

Zotero Preferences / Login Manager serves as the secure persistent vault. Backend keys are in-memory session variables (os.environ) synced from Zotero.

Parameters:

Name Type Description Default
key_name str

Entry identifier (e.g., 'openrouter_api_key', 'openai_api_key').

 'openrouter_api_key'
env_vars list[str] | tuple[str, ...] | None

Environment variable names to check.

 ('OPENROUTER_API_KEY', 'OPEN_ROUTER_API_KEY')

Returns:

Type Description
str | None

The API key string or None if not found.
Source code in zocrates/config.py 
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
def get_secure_api_key(
    key_name: str = "openrouter_api_key",
    env_vars: list[str] | tuple[str, ...] | None = ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY"),
) -> str | None:
    """Load a named API key from in-memory session environment variables.

    Zotero Preferences / Login Manager serves as the secure persistent vault.
    Backend keys are in-memory session variables (os.environ) synced from Zotero.

    Args:
        key_name: Entry identifier (e.g., 'openrouter_api_key', 'openai_api_key').
        env_vars: Environment variable names to check.

    Returns:
        The API key string or None if not found.
    """
    if env_vars:
        for var in env_vars:
            val = os.getenv(var)
            if val:
                return val
    return None

get_semantic_similarity_threshold 

get_semantic_similarity_threshold(zotero_data_dir=None, cfg=None)

Return the minimum cosine similarity threshold for semantic vector retrieval.

Reads strictly from semantic_similarity_threshold in zocrates_config.json (clamped between 0.0 and 1.0, defaulting to 0.50).

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
float

Float similarity threshold between 0.0 and 1.0.
Source code in zocrates/config.py 
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
def get_semantic_similarity_threshold(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> float:
    """Return the minimum cosine similarity threshold for semantic vector retrieval.

    Reads strictly from ``semantic_similarity_threshold`` in ``zocrates_config.json``
    (clamped between 0.0 and 1.0, defaulting to ``0.50``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Float similarity threshold between 0.0 and 1.0.
    """
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    try:
        val = float(cfg.get("semantic_similarity_threshold", DEFAULT_SEMANTIC_SIMILARITY_THRESHOLD))
        return max(0.0, min(1.0, val))
    except (ValueError, TypeError):
        return DEFAULT_SEMANTIC_SIMILARITY_THRESHOLD

get_sentence_attribution_model 

get_sentence_attribution_model(cfg=None, zotero_data_dir=None)

Return the resolved model identifier for sentence attribution.

If cfg.get("sentence_attribution_model") is set and non-empty, resolves it with _resolve_model_name(custom_model, provider). If empty, null, or unset, delegates to get_coordinator_model(cfg, zotero_data_dir).

Source code in zocrates/config.py 
2100
2101
2102
2103
2104
2105
2106
2107
2108
2109
2110
2111
2112
2113
2114
2115
2116
2117
2118
2119
2120
def get_sentence_attribution_model(
    cfg: dict[str, Any] | None = None,
    zotero_data_dir: Path | str | None = None,
) -> str:
    """Return the resolved model identifier for sentence attribution.

    If cfg.get("sentence_attribution_model") is set and non-empty, resolves it
    with _resolve_model_name(custom_model, provider). If empty, null, or unset,
    delegates to get_coordinator_model(cfg, zotero_data_dir).
    """
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)

    custom_model = cfg.get("sentence_attribution_model")
    if custom_model is not None and isinstance(custom_model, str) and custom_model.strip():
        provider = cfg.get("llm_provider") or "openrouter_llm"
        from zocrates.llm import _resolve_model_name

        return _resolve_model_name(custom_model.strip(), provider)

    return get_coordinator_model(cfg, zotero_data_dir)

get_session_auto_vacuum_on_startup 

get_session_auto_vacuum_on_startup(zotero_data_dir=None, cfg=None)

Return whether session database auto-vacuum on startup is enabled.

Source code in zocrates/config.py 
1931
1932
1933
1934
1935
1936
1937
1938
def get_session_auto_vacuum_on_startup(
    zotero_data_dir: Path | str | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether session database auto-vacuum on startup is enabled."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("session_db_auto_vacuum_on_startup", DEFAULT_SESSION_DB_AUTO_VACUUM_ON_STARTUP))

get_session_maintenance_config 

get_session_maintenance_config(zotero_data_dir=None, cfg=None)

Return session database retention and maintenance settings dictionary from zocrates_config.json.

Source code in zocrates/config.py 
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
def get_session_maintenance_config(
    zotero_data_dir: Path | str | None = None,
    cfg: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Return session database retention and maintenance settings dictionary from zocrates_config.json."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return {
        "max_saved_sessions": cfg.get("max_saved_sessions", DEFAULT_MAX_SAVED_SESSIONS),
        "session_retention_days": cfg.get("session_retention_days", DEFAULT_SESSION_RETENTION_DAYS),
        "session_db_auto_vacuum_on_startup": bool(
            cfg.get("session_db_auto_vacuum_on_startup", DEFAULT_SESSION_DB_AUTO_VACUUM_ON_STARTUP)
        ),
        "prune_empty_sessions": bool(
            cfg.get("prune_empty_sessions", DEFAULT_PRUNE_EMPTY_SESSIONS)
        ),
    }

get_session_retention_days 

get_session_retention_days(zotero_data_dir=None, cfg=None)

Return maximum age in days for session retention, or None if disabled/unlimited.

Source code in zocrates/config.py 
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
def get_session_retention_days(
    zotero_data_dir: Path | str | None = None,
    cfg: dict[str, Any] | None = None,
) -> int | None:
    """Return maximum age in days for session retention, or None if disabled/unlimited."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    val = cfg.get("session_retention_days", DEFAULT_SESSION_RETENTION_DAYS)
    if val is None:
        return None
    try:
        return int(val)
    except (ValueError, TypeError):
        return DEFAULT_SESSION_RETENTION_DAYS

get_sync_auto_abort_failures 

get_sync_auto_abort_failures(zotero_data_dir=None, cfg=None)

Return the consecutive-failure auto-abort threshold for full-library sync jobs.

Reads sync_auto_abort_failures from zocrates_config.json (defaulting to DEFAULT_SYNC_AUTO_ABORT_FAILURES). Values <= 0 disable auto-abort.

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None
Source code in zocrates/config.py 
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
def get_sync_auto_abort_failures(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> int:
    """Return the consecutive-failure auto-abort threshold for full-library sync jobs.

    Reads ``sync_auto_abort_failures`` from ``zocrates_config.json`` (defaulting
    to ``DEFAULT_SYNC_AUTO_ABORT_FAILURES``). Values <= 0 disable auto-abort.

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.
    """
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    try:
        return int(cfg.get("sync_auto_abort_failures", DEFAULT_SYNC_AUTO_ABORT_FAILURES))
    except (ValueError, TypeError):
        return DEFAULT_SYNC_AUTO_ABORT_FAILURES

get_user_research_profile 

get_user_research_profile(zotero_data_dir=None, cfg=None)

Return the configured user research profile (stripped), or "" if unset.

Source code in zocrates/config.py 
1808
1809
1810
1811
1812
1813
1814
1815
1816
def get_user_research_profile(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the configured user research profile (stripped), or ``""`` if unset."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    val = cfg.get("user_research_profile", DEFAULT_USER_RESEARCH_PROFILE)
    return str(val).strip() if val else ""

get_vlm_captioning_backup_model 

get_vlm_captioning_backup_model(zotero_data_dir=None, cfg=None)

Return the backup VLM model identifier for cloud captioning fallback.

Checks VLM_CAPTIONING_BACKUP_MODEL environment variable first; if unset, falls back to vlm_captioning_backup_model in zocrates_config.json (defaulting to "openrouter/nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free").

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
str

Backup model identifier string.
Source code in zocrates/config.py 
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
def get_vlm_captioning_backup_model(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the backup VLM model identifier for cloud captioning fallback.

    Checks ``VLM_CAPTIONING_BACKUP_MODEL`` environment variable first;
    if unset, falls back to ``vlm_captioning_backup_model`` in ``zocrates_config.json``
    (defaulting to ``"openrouter/nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free"``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Backup model identifier string.
    """
    env_val = os.getenv("VLM_CAPTIONING_BACKUP_MODEL", "")
    if env_val:
        return env_val
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return cfg.get(
        "vlm_captioning_backup_model", DEFAULT_VLM_CAPTIONING_BACKUP_MODEL
    )

get_vlm_captioning_local_model 

get_vlm_captioning_local_model(zotero_data_dir=None, cfg=None)

Return the local Ollama model name for figure captioning.

Checks VLM_CAPTIONING_LOCAL_MODEL environment variable first; if unset, falls back to vlm_captioning_local_model in zocrates_config.json (defaulting to "").

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
str

Local model name string.
Source code in zocrates/config.py 
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
def get_vlm_captioning_local_model(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the local Ollama model name for figure captioning.

    Checks ``VLM_CAPTIONING_LOCAL_MODEL`` environment variable first;
    if unset, falls back to ``vlm_captioning_local_model`` in ``zocrates_config.json``
    (defaulting to ``""``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Local model name string.
    """
    env_val = os.getenv("VLM_CAPTIONING_LOCAL_MODEL", "")
    if env_val:
        return env_val
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return cfg.get("vlm_captioning_local_model", DEFAULT_VLM_CAPTIONING_LOCAL_MODEL)

get_vlm_captioning_model 

get_vlm_captioning_model(zotero_data_dir=None, cfg=None)

Return the primary VLM model identifier for cloud captioning.

Checks VLM_CAPTIONING_MODEL environment variable first; if unset, falls back to vlm_captioning_model in zocrates_config.json (defaulting to "openrouter/nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free").

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
str

Model identifier string.
Source code in zocrates/config.py 
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
def get_vlm_captioning_model(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the primary VLM model identifier for cloud captioning.

    Checks ``VLM_CAPTIONING_MODEL`` environment variable first;
    if unset, falls back to ``vlm_captioning_model`` in ``zocrates_config.json``
    (defaulting to ``"openrouter/nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free"``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Model identifier string.
    """
    env_val = os.getenv("VLM_CAPTIONING_MODEL", "")
    if env_val:
        return env_val
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return cfg.get("vlm_captioning_model", DEFAULT_VLM_CAPTIONING_MODEL)

get_vlm_captioning_provider 

get_vlm_captioning_provider(zotero_data_dir=None, cfg=None)

Return the configured VLM provider ('openrouter' or 'ollama').

Checks VLM_CAPTIONING_PROVIDER environment variable first; if unset, falls back to vlm_captioning_provider in zocrates_config.json (defaulting to "openrouter").

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
str

Provider string ('openrouter' or 'ollama').
Source code in zocrates/config.py 
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
def get_vlm_captioning_provider(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the configured VLM provider ('openrouter' or 'ollama').

    Checks ``VLM_CAPTIONING_PROVIDER`` environment variable first;
    if unset, falls back to ``vlm_captioning_provider`` in ``zocrates_config.json``
    (defaulting to ``"openrouter"``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Provider string ('openrouter' or 'ollama').
    """
    env_val = os.getenv("VLM_CAPTIONING_PROVIDER", "")
    if env_val:
        return env_val.lower()
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return cfg.get("vlm_captioning_provider", DEFAULT_VLM_CAPTIONING_PROVIDER).lower()

get_vlm_max_images_per_doc 

get_vlm_max_images_per_doc(zotero_data_dir=None, cfg=None)

Return the maximum number of images to caption per document.

Checks VLM_MAX_IMAGES_PER_DOC environment variable first; if unset, falls back to vlm_max_images_per_doc in zocrates_config.json (defaulting to 20).

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
int

Max images integer ceiling.
Source code in zocrates/config.py 
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
def get_vlm_max_images_per_doc(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> int:
    """Return the maximum number of images to caption per document.

    Checks ``VLM_MAX_IMAGES_PER_DOC`` environment variable first;
    if unset, falls back to ``vlm_max_images_per_doc`` in ``zocrates_config.json``
    (defaulting to ``20``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Max images integer ceiling.
    """
    env_val = os.getenv("VLM_MAX_IMAGES_PER_DOC")
    if env_val is not None:
        try:
            return int(env_val)
        except ValueError:
            pass
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    try:
        return int(cfg.get("vlm_max_images_per_doc", DEFAULT_VLM_MAX_IMAGES_PER_DOC))
    except (ValueError, TypeError):
        return DEFAULT_VLM_MAX_IMAGES_PER_DOC

get_vlm_timeout_seconds 

get_vlm_timeout_seconds(zotero_data_dir=None, cfg=None)

Return the timeout in seconds for individual VLM captioning requests.

Checks VLM_TIMEOUT_SECONDS environment variable first; if unset, falls back to vlm_timeout_seconds in zocrates_config.json (defaulting to 30).

Parameters:

Name Type Description Default
zotero_data_dir Path | None

Path to the Zotero data directory.

 None
cfg dict[str, Any] | None

Optional in-memory config dict. If provided, bypasses disk read.

 None

Returns:

Type Description
int

Timeout integer in seconds.
Source code in zocrates/config.py 
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
def get_vlm_timeout_seconds(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> int:
    """Return the timeout in seconds for individual VLM captioning requests.

    Checks ``VLM_TIMEOUT_SECONDS`` environment variable first;
    if unset, falls back to ``vlm_timeout_seconds`` in ``zocrates_config.json``
    (defaulting to ``30``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Timeout integer in seconds.
    """
    env_val = os.getenv("VLM_TIMEOUT_SECONDS")
    if env_val is not None:
        try:
            return int(env_val)
        except ValueError:
            pass
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    try:
        return int(cfg.get("vlm_timeout_seconds", DEFAULT_VLM_TIMEOUT_SECONDS))
    except (ValueError, TypeError):
        return DEFAULT_VLM_TIMEOUT_SECONDS

get_zocrates_data_dir 

get_zocrates_data_dir(zotero_data_dir=None)

Return (and create if needed) the Zocrates data directory.

Parameters:

Name Type Description Default
zotero_data_dir Path | str | None

Root Zotero data directory (Path, string, or None for auto-discovery).

 None

Returns:

Type Description
Path

Path to <zotero_data_dir>/zocrates-data/, created if absent.
Source code in zocrates/config.py 
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
def get_zocrates_data_dir(zotero_data_dir: Path | str | None = None) -> Path:
    """Return (and create if needed) the Zocrates data directory.

    Args:
        zotero_data_dir: Root Zotero data directory (Path, string, or None for auto-discovery).

    Returns:
        Path to ``<zotero_data_dir>/zocrates-data/``, created if absent.
    """
    resolved = resolve_zotero_data_dir(zotero_data_dir)
    if resolved is None:
        raise RuntimeError(
            "Cannot determine Zotero data directory. "
            "Please ensure Zotero is running and registered, or set ZOTERO_DATA_DIR."
        )
    zocrates_dir = resolved / ZOCRATES_DATA_SUBDIR
    zocrates_dir.mkdir(parents=True, exist_ok=True)
    return zocrates_dir

is_adaptive_compaction_enabled 

is_adaptive_compaction_enabled(zotero_data_dir=None, cfg=None)

Return whether adaptive context compaction is enabled from zocrates_config.json.

Source code in zocrates/config.py 
1819
1820
1821
1822
1823
1824
1825
1826
1827
def is_adaptive_compaction_enabled(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether adaptive context compaction is enabled from zocrates_config.json."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    val = cfg.get("enable_adaptive_compaction")
    return bool(val if val is not None else DEFAULT_ENABLE_ADAPTIVE_COMPACTION)

is_context_compaction_enabled 

is_context_compaction_enabled(zotero_data_dir=None, cfg=None)

Deprecated alias for is_adaptive_compaction_enabled.

Source code in zocrates/config.py 
1830
1831
1832
1833
1834
1835
def is_context_compaction_enabled(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Deprecated alias for is_adaptive_compaction_enabled."""
    return is_adaptive_compaction_enabled(zotero_data_dir, cfg=cfg)

is_local_decision_model 

is_local_decision_model(model_name)

Check if model name indicates local Laya decision model.

Source code in zocrates/config.py 
1838
1839
1840
1841
1842
1843
1844
1845
def is_local_decision_model(model_name: str | None) -> bool:
    """Check if model name indicates local Laya decision model."""
    cleaned = (model_name or "").strip().lower()
    return (
        cleaned in ("laya", "local", "local_laya", "local/laya")
        or cleaned.startswith("local:")
        or cleaned.startswith("local/")
    )

is_mcp_server_enabled 

is_mcp_server_enabled(zotero_data_dir=None, cfg=None)

Return whether local MCP server access is enabled from zocrates_config.json.

Source code in zocrates/config.py 
1777
1778
1779
1780
1781
1782
1783
1784
def is_mcp_server_enabled(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Return whether local MCP server access is enabled from zocrates_config.json."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return bool(cfg.get("enable_mcp_server", DEFAULT_ENABLE_MCP_SERVER))

load_zocrates_config 

load_zocrates_config(zotero_data_dir=None)

Load the Zocrates config from the data directory.

If zotero_data_dir is None, attempts to obtain it from the active LibrarySyncer service or the ZOTERO_DATA_DIR environment variable.

Returns config merged with standard defaults.

Parameters:

Name Type Description Default
zotero_data_dir Path | str | None

Root Zotero data directory.

 None

Returns:

Type Description
dict

Parsed JSON config dict merged with defaults.
Source code in zocrates/config.py 
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
def load_zocrates_config(zotero_data_dir: Path | str | None = None) -> dict:
    """Load the Zocrates config from the data directory.

    If ``zotero_data_dir`` is ``None``, attempts to obtain it from the active
    ``LibrarySyncer`` service or the ``ZOTERO_DATA_DIR`` environment variable.

    Returns config merged with standard defaults.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Parsed JSON config dict merged with defaults.
    """
    defaults = copy.deepcopy(DEFAULT_CONFIG)

    resolved_dir = resolve_zotero_data_dir(zotero_data_dir)
    if resolved_dir is None:
        return defaults

    config_path = get_config_path(resolved_dir)
    resolved_path = config_path.resolve()
    if not config_path.exists():
        with _CONFIG_CACHE_LOCK:
            _CONFIG_CACHE.pop(resolved_path, None)
            _CONFIG_CACHE_FINGERPRINT.pop(resolved_path, None)
        return defaults

    try:
        stat_res = config_path.stat()
        current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
    except OSError:
        return defaults

    with _CONFIG_CACHE_LOCK:
        if resolved_path in _CONFIG_CACHE and _CONFIG_CACHE_FINGERPRINT.get(resolved_path) == current_fingerprint:
            return copy.deepcopy(_CONFIG_CACHE[resolved_path])

    if config_path.exists():
        try:
            lock_path = str(config_path) + ".lock"
            with FileLock(lock_path):
                try:
                    stat_res = config_path.stat()
                    current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
                except OSError:
                    return defaults

                with _CONFIG_CACHE_LOCK:
                    if resolved_path in _CONFIG_CACHE and _CONFIG_CACHE_FINGERPRINT.get(resolved_path) == current_fingerprint:
                        return copy.deepcopy(_CONFIG_CACHE[resolved_path])

                with open(config_path) as f:
                    loaded = json.load(f)

                # Backfill override_agent_models if missing or incomplete
                needs_save = False

                if "override_agent_models" not in loaded or not isinstance(loaded["override_agent_models"], dict):
                    loaded["override_agent_models"] = copy.deepcopy(DEFAULT_OVERRIDE_AGENT_MODELS)
                    needs_save = True
                else:
                    for role, default_role_cfg in DEFAULT_OVERRIDE_AGENT_MODELS.items():
                        if role not in loaded["override_agent_models"]:
                            loaded["override_agent_models"][role] = copy.deepcopy(default_role_cfg)
                            needs_save = True

                # Backfill search_harness if missing or incomplete in existing config file on disk
                if "search_harness" not in loaded or not isinstance(loaded["search_harness"], dict):
                    loaded["search_harness"] = copy.deepcopy(DEFAULT_SEARCH_HARNESS)
                    needs_save = True
                else:
                    for k, default_val in DEFAULT_SEARCH_HARNESS.items():
                        if k not in loaded["search_harness"]:
                            loaded["search_harness"][k] = copy.deepcopy(default_val)
                            needs_save = True

                # Backfill graph_relations if missing or incomplete in existing config file on disk
                if "graph_relations" not in loaded or not isinstance(loaded["graph_relations"], dict):
                    loaded["graph_relations"] = copy.deepcopy(DEFAULT_GRAPH_RELATIONS)
                    needs_save = True
                else:
                    for k, default_val in DEFAULT_GRAPH_RELATIONS.items():
                        if k not in loaded["graph_relations"]:
                            loaded["graph_relations"][k] = copy.deepcopy(default_val)
                            needs_save = True

                # Backfill llm_num_retries if missing from existing config file on disk
                if "llm_num_retries" not in loaded:
                    loaded["llm_num_retries"] = DEFAULT_LLM_NUM_RETRIES
                    needs_save = True

                # Backfill llm_base_url if missing from existing config file on disk
                if "llm_base_url" not in loaded:
                    loaded["llm_base_url"] = DEFAULT_LLM_BASE_URL
                    needs_save = True

                # Backfill citations options if missing from existing config file on disk
                if "citations_enabled" not in loaded:
                    loaded["citations_enabled"] = DEFAULT_CITATIONS_ENABLED
                    needs_save = True
                if "references_section_enabled" not in loaded:
                    loaded["references_section_enabled"] = DEFAULT_REFERENCES_SECTION_ENABLED
                    needs_save = True
                if "considered_sources_disclosure" not in loaded:
                    loaded["considered_sources_disclosure"] = DEFAULT_CONSIDERED_SOURCES_DISCLOSURE
                    needs_save = True
                if "auto_render_mermaid_diagrams" not in loaded:
                    loaded["auto_render_mermaid_diagrams"] = DEFAULT_AUTO_RENDER_MERMAID_DIAGRAMS
                    needs_save = True
                if "enable_mcp_server" not in loaded:
                    loaded["enable_mcp_server"] = DEFAULT_ENABLE_MCP_SERVER
                    needs_save = True
                if "auto_spawn_backend" not in loaded:
                    loaded["auto_spawn_backend"] = DEFAULT_AUTO_SPAWN_BACKEND
                    needs_save = True
                if "enable_predictive_ghost_input" not in loaded:
                    loaded["enable_predictive_ghost_input"] = DEFAULT_ENABLE_PREDICTIVE_GHOST_INPUT
                    needs_save = True
                if "predictive_prompt_model" not in loaded:
                    loaded["predictive_prompt_model"] = DEFAULT_PREDICTIVE_PROMPT_MODEL
                    needs_save = True
                if "sentence_attribution_model" not in loaded:
                    loaded["sentence_attribution_model"] = DEFAULT_SENTENCE_ATTRIBUTION_MODEL
                    needs_save = True
                if "scholar_polite_email" not in loaded:
                    loaded["scholar_polite_email"] = DEFAULT_SCHOLAR_POLITE_EMAIL
                    needs_save = True
                if "conversion_timeout_seconds" not in loaded:
                    loaded["conversion_timeout_seconds"] = DEFAULT_CONVERSION_TIMEOUT_SECONDS
                    needs_save = True
                if "datalab_merge_cross_page" not in loaded:
                    loaded["datalab_merge_cross_page"] = DEFAULT_DATALAB_MERGE_CROSS_PAGE
                    needs_save = True
                if "user_research_profile" not in loaded:
                    loaded["user_research_profile"] = DEFAULT_USER_RESEARCH_PROFILE
                    needs_save = True
                if "max_research_profile_chars" not in loaded:
                    loaded["max_research_profile_chars"] = DEFAULT_MAX_RESEARCH_PROFILE_CHARS
                    needs_save = True
                if "max_saved_sessions" not in loaded:
                    loaded["max_saved_sessions"] = DEFAULT_MAX_SAVED_SESSIONS
                    needs_save = True
                if "session_retention_days" not in loaded:
                    loaded["session_retention_days"] = DEFAULT_SESSION_RETENTION_DAYS
                    needs_save = True
                if "session_db_auto_vacuum_on_startup" not in loaded:
                    loaded["session_db_auto_vacuum_on_startup"] = DEFAULT_SESSION_DB_AUTO_VACUUM_ON_STARTUP
                    needs_save = True
                if "prune_empty_sessions" not in loaded:
                    loaded["prune_empty_sessions"] = DEFAULT_PRUNE_EMPTY_SESSIONS
                    needs_save = True
                if "backend_port" not in loaded:
                    loaded["backend_port"] = DEFAULT_BACKEND_PORT
                    needs_save = True

                # Backfill Phase 7 onboarding SSOT keys if missing from existing
                # config files on disk (embedding_dimension / sync_auto_abort_failures)
                if "embedding_dimension" not in loaded:
                    loaded["embedding_dimension"] = DEFAULT_CONFIG["embedding_dimension"]
                    needs_save = True
                if "sync_auto_abort_failures" not in loaded:
                    loaded["sync_auto_abort_failures"] = DEFAULT_SYNC_AUTO_ABORT_FAILURES
                    needs_save = True

                if "vlm_captioning_backup_model" not in loaded:
                    loaded["vlm_captioning_backup_model"] = DEFAULT_CONFIG["vlm_captioning_backup_model"]
                    needs_save = True
                if "vlm_captioning_model" not in loaded:
                    loaded["vlm_captioning_model"] = DEFAULT_CONFIG["vlm_captioning_model"]
                    needs_save = True
                if "vlm_captioning_provider" not in loaded:
                    loaded["vlm_captioning_provider"] = DEFAULT_CONFIG["vlm_captioning_provider"]
                    needs_save = True
                if "enable_vlm_captioning" not in loaded:
                    loaded["enable_vlm_captioning"] = DEFAULT_CONFIG["enable_vlm_captioning"]
                    needs_save = True
                if "vlm_max_images_per_doc" not in loaded:
                    loaded["vlm_max_images_per_doc"] = DEFAULT_CONFIG["vlm_max_images_per_doc"]
                    needs_save = True

                # Detect and migrate deprecated compaction keys from disk
                if "compaction_token_threshold" in loaded and "adaptive_compaction_token_ceiling" not in loaded:
                    loaded["adaptive_compaction_token_ceiling"] = loaded["compaction_token_threshold"]
                    needs_save = True
                if "enable_context_compaction" in loaded and "enable_adaptive_compaction" not in loaded:
                    loaded["enable_adaptive_compaction"] = loaded["enable_context_compaction"]
                    needs_save = True
                if "adaptive_decision_model" in loaded and "adaptive_compaction_decision_model" not in loaded:
                    loaded["adaptive_compaction_decision_model"] = loaded.pop("adaptive_decision_model")
                    needs_save = True
                elif "adaptive_decision_model" in loaded:
                    loaded.pop("adaptive_decision_model", None)
                    needs_save = True

                # Purge all 5 deprecated compaction keys
                deprecated_compaction_keys = (
                    "compaction_interval",
                    "compaction_overlap_size",
                    "compaction_token_threshold",
                    "compaction_event_retention_size",
                    "enable_context_compaction",
                )
                for dep_key in deprecated_compaction_keys:
                    if dep_key in loaded:
                        loaded.pop(dep_key, None)
                        needs_save = True

                # Backfill missing adaptive compaction options
                adaptive_compaction_defaults = {
                    "enable_adaptive_compaction": DEFAULT_ENABLE_ADAPTIVE_COMPACTION,
                    "adaptive_compaction_retention_turns": DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS,
                    "adaptive_compaction_min_turns": DEFAULT_ADAPTIVE_COMPACTION_MIN_TURNS,
                    "adaptive_compaction_token_floor": DEFAULT_ADAPTIVE_COMPACTION_TOKEN_FLOOR,
                    "adaptive_compaction_token_ceiling": DEFAULT_ADAPTIVE_COMPACTION_TOKEN_CEILING,
                    "adaptive_compaction_threshold": DEFAULT_ADAPTIVE_COMPACTION_THRESHOLD,
                    "adaptive_compaction_decision_model": DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL,
                    "compaction_summarizer_model": DEFAULT_COMPACTION_SUMMARIZER_MODEL,
                }
                for k, default_val in adaptive_compaction_defaults.items():
                    if k not in loaded:
                        loaded[k] = default_val
                        needs_save = True

                # Disallow local Laya for compaction decisions on disk; reset to default remote model
                if is_local_decision_model(loaded.get("adaptive_compaction_decision_model")):
                    loaded["adaptive_compaction_decision_model"] = DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL
                    needs_save = True

                # Backfill citation verification options if missing from existing config file on disk
                verification_defaults = {
                    "citation_verification_enabled": DEFAULT_CITATION_VERIFICATION_ENABLED,
                    "citation_verification_provider": DEFAULT_CITATION_VERIFICATION_PROVIDER,
                    "citation_verification_model": DEFAULT_CITATION_VERIFICATION_MODEL,
                    "citation_verification_threshold": DEFAULT_CITATION_VERIFICATION_THRESHOLD,
                    "citation_verification_local_url": DEFAULT_CITATION_VERIFICATION_LOCAL_URL,
                    "citation_verification_local_port": DEFAULT_CITATION_VERIFICATION_LOCAL_PORT,
                    "citation_verification_auto_start_local": DEFAULT_CITATION_VERIFICATION_AUTO_START_LOCAL,
                    "citation_verification_timeout_sec": DEFAULT_CITATION_VERIFICATION_TIMEOUT_SEC,
                    "citation_verification_turn_budget_sec": DEFAULT_CITATION_VERIFICATION_TURN_BUDGET_SEC,
                }
                for k, default_val in verification_defaults.items():
                    if k not in loaded:
                        loaded[k] = default_val
                        needs_save = True
                # Migrate legacy default 3.0s/10.0s timeout and budget to 10.0s/20.0s for CPU inference headroom
                try:
                    if float(loaded.get("citation_verification_timeout_sec", 0)) == 3.0:
                        loaded["citation_verification_timeout_sec"] = DEFAULT_CITATION_VERIFICATION_TIMEOUT_SEC
                        needs_save = True
                except (ValueError, TypeError):
                    pass
                try:
                    if float(loaded.get("citation_verification_turn_budget_sec", 0)) == 10.0:
                        loaded["citation_verification_turn_budget_sec"] = DEFAULT_CITATION_VERIFICATION_TURN_BUDGET_SEC
                        needs_save = True
                except (ValueError, TypeError):
                    pass

                if needs_save:
                    try:
                        with open(config_path, "w") as f_out:
                            json.dump(loaded, f_out, indent=2)
                        stamp_config_mtime(resolved_dir)
                    except Exception as save_err:
                        logger.warning(
                            "Failed to backfill default dialogue agent models to %s: %s",
                            config_path,
                            save_err,
                        )
                    try:
                        stat_res = config_path.stat()
                        current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
                    except OSError:
                        pass

                res = dict(defaults)
                res.update(loaded)
                with _CONFIG_CACHE_LOCK:
                    _CONFIG_CACHE[resolved_path] = copy.deepcopy(res)
                    _CONFIG_CACHE_FINGERPRINT[resolved_path] = current_fingerprint
                return copy.deepcopy(res)
        except Exception as e:
            logger.warning("Failed to load config from %s: %s", config_path, e)
    return defaults

rearm_config_freshness 

rearm_config_freshness()

Re-arm the config freshness sentinel so a failed hot-swap can be retried immediately.

Sets _last_config_mtime to 1.0 (epoch + 1s). Since any actual file mtime is greater than 1.0, the next check_config_freshness call will re-trigger True.

Source code in zocrates/config.py 
499
500
501
502
503
504
505
506
def rearm_config_freshness() -> None:
    """Re-arm the config freshness sentinel so a failed hot-swap can be retried immediately.

    Sets _last_config_mtime to 1.0 (epoch + 1s). Since any actual file mtime is greater
    than 1.0, the next check_config_freshness call will re-trigger True.
    """
    global _last_config_mtime
    _last_config_mtime = 1.0

resolve_zotero_data_dir 

resolve_zotero_data_dir(zotero_data_dir=None)

Resolve the active Zotero data directory with auto-discovery fallback.

Parameters:

Name Type Description Default
zotero_data_dir Path | str | None

Explicit data directory path or None.

 None

Returns:

Type Description
Path | None

Resolved Path to the Zotero data directory, or None if not located.
Source code in zocrates/config.py 
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
def resolve_zotero_data_dir(zotero_data_dir: Path | str | None = None) -> Path | None:
    """Resolve the active Zotero data directory with auto-discovery fallback.

    Args:
        zotero_data_dir: Explicit data directory path or None.

    Returns:
        Resolved Path to the Zotero data directory, or None if not located.
    """
    if zotero_data_dir is not None:
        return Path(zotero_data_dir)
    try:
        from zocrates.rag.service_manager import get_syncer
        syncer = get_syncer()
        if syncer and getattr(syncer, "_zotero_data_dir", None):
            return Path(syncer._zotero_data_dir)
    except Exception:
        pass
    if ZOTERO_DATA_DIR:
        return Path(ZOTERO_DATA_DIR)
    if env_dir := os.getenv("ZOTERO_DATA_DIR"):
        return Path(env_dir)
    # Auto-discover active Zotero profile in user home directory (fallback for offline scripts)
    candidates = (
        list(Path.home().glob("Zotero*/zocrates-data"))
        + list(Path.home().glob("Zotero*/*/zocrates-data"))
        + list(Path.home().glob(".var/app/org.zotero.Zotero/data/zotero/zocrates-data"))
        + list(Path.home().glob(".zotero/zotero/*/zocrates-data"))
    )
    if candidates:
        selected = candidates[0].parent
        if len(candidates) > 1:
            logger.warning(
                "Multiple Zotero profiles found with zocrates-data (%s). "
                "Defaulting to %s. Set ZOTERO_DATA_DIR to select a specific profile.",
                [str(c.parent) for c in candidates],
                selected,
            )
        else:
            logger.debug("Auto-discovered Zotero data directory: %s", selected)
        return selected
    return None

save_datalab_api_key 

save_datalab_api_key(api_key)

Save the Datalab API key in the session environment.

Parameters:

Name Type Description Default
api_key str | SecretStr | None

SecretStr, raw string key, or None/empty to delete.

 required
Source code in zocrates/config.py 
1299
1300
1301
1302
1303
1304
1305
def save_datalab_api_key(api_key: str | SecretStr | None) -> None:
    """Save the Datalab API key in the session environment.

    Args:
        api_key: SecretStr, raw string key, or None/empty to delete.
    """
    save_secure_api_key(api_key, key_name="datalab_api_key", env_var="DATALAB_API_KEY")

save_openalex_api_key 

save_openalex_api_key(api_key)

Save or clear the OpenAlex API key in the session environment.

Source code in zocrates/config.py 
1316
1317
1318
def save_openalex_api_key(api_key: str | SecretStr | None) -> None:
    """Save or clear the OpenAlex API key in the session environment."""
    save_secure_api_key(api_key, key_name="openalex_api_key", env_var="OPENALEX_API_KEY")

save_secure_api_key 

save_secure_api_key(api_key, key_name='openrouter_api_key', env_var='OPENROUTER_API_KEY')

Store a named API key in the session environment.

Zotero Preferences / Login Manager (via the front-end extension) serves as the authoritative secure persistent vault for user credentials across restarts. The backend server stores keys in-memory as session variables (os.environ) synced from Zotero (via environment variables on startup and POST /config during runtime).

Does NOT persist plain-text keys to disk configuration files.

Parameters:

Name Type Description Default
api_key str | SecretStr | None

SecretStr, raw string key, or None/empty to delete.

 required
key_name str

Entry identifier (e.g., 'openrouter_api_key', 'datalab_api_key').

 'openrouter_api_key'
env_var str | None

Associated environment variable name to sync in memory.

 'OPENROUTER_API_KEY'
Source code in zocrates/config.py 
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
def save_secure_api_key(
    api_key: str | SecretStr | None,
    key_name: str = "openrouter_api_key",
    env_var: str | None = "OPENROUTER_API_KEY",
) -> None:
    """Store a named API key in the session environment.

    Zotero Preferences / Login Manager (via the front-end extension) serves as
    the authoritative secure persistent vault for user credentials across restarts.
    The backend server stores keys in-memory as session variables (os.environ) synced
    from Zotero (via environment variables on startup and POST /config during runtime).

    Does NOT persist plain-text keys to disk configuration files.

    Args:
        api_key: SecretStr, raw string key, or None/empty to delete.
        key_name: Entry identifier (e.g., 'openrouter_api_key', 'datalab_api_key').
        env_var: Associated environment variable name to sync in memory.
    """
    if isinstance(api_key, SecretStr):
        secret_val = api_key.get_secret_value()
    elif api_key is not None:
        secret_val = str(api_key)
    else:
        secret_val = ""

    if secret_val and secret_val.strip():
        val = secret_val.strip()
        if env_var:
            os.environ[env_var] = val
    else:
        if env_var:
            os.environ.pop(env_var, None)

save_zocrates_config 

save_zocrates_config(config, zotero_data_dir=None)

Persist config (merged into any existing values) to the data directory.

If zotero_data_dir is None, attempts to obtain it from the active LibrarySyncer service or the ZOTERO_DATA_DIR environment variable. Uses a FileLock to prevent concurrent write corruption.

API keys are automatically intercepted and routed to the in-memory environment via save_secure_api_key, ensuring they are never stored in plain text on disk.

Parameters:

Name Type Description Default
config dict

Key/value pairs to merge into the existing config file.

 required
zotero_data_dir Path | None

Root Zotero data directory.

 None
Source code in zocrates/config.py 
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
def save_zocrates_config(config: dict, zotero_data_dir: Path | None = None) -> None:
    """Persist ``config`` (merged into any existing values) to the data directory.

    If ``zotero_data_dir`` is ``None``, attempts to obtain it from the active
    ``LibrarySyncer`` service or the ``ZOTERO_DATA_DIR`` environment variable.
    Uses a ``FileLock`` to prevent concurrent write corruption.

    API keys are automatically intercepted and routed to the in-memory environment via
    ``save_secure_api_key``, ensuring they are never stored in plain text on disk.

    Args:
        config: Key/value pairs to merge into the existing config file.
        zotero_data_dir: Root Zotero data directory.
    """
    config_copy = dict(config)

    profile_val = config_copy.get("user_research_profile")
    if profile_val is not None:
        if not isinstance(profile_val, str):
            raise ValueError("user_research_profile must be a string.")
        max_chars = int(config_copy.get("max_research_profile_chars") or DEFAULT_MAX_RESEARCH_PROFILE_CHARS)
        if len(profile_val) > max_chars:
            raise ValueError(
                f"user_research_profile exceeds the maximum of "
                f"{max_chars} characters ({len(profile_val)} given)."
            )

    port_val = config_copy.get("backend_port")
    if port_val is not None:
        try:
            port_int = int(port_val)
            if not (1 <= port_int <= 65535):
                raise ValueError(
                    f"backend_port must be between 1 and 65535, got {port_int}."
                )
            config_copy["backend_port"] = port_int
        except (ValueError, TypeError) as port_err:
            raise ValueError(f"Invalid backend_port: {port_err}") from port_err

    for k in list(config_copy.keys()):
        if k.endswith("_api_key") or k == "cloud_api_key":
            val = config_copy.pop(k)
            if "openrouter" in k:
                save_secure_api_key(val, key_name="openrouter_api_key", env_var="OPENROUTER_API_KEY")
            else:
                env_var = k.upper()
                save_secure_api_key(val, key_name=k, env_var=env_var)

    zotero_data_dir = resolve_zotero_data_dir(zotero_data_dir)
    if zotero_data_dir is None:
        logger.warning("Cannot save config: Zotero data directory is unresolved.")
        return

    config_path = get_config_path(zotero_data_dir)
    lock_path = str(config_path) + ".lock"
    try:
        with FileLock(lock_path):
            existing: dict = {}
            if config_path.exists():
                try:
                    with open(config_path) as f:
                        existing = json.load(f)
                except Exception as e:
                    logger.warning(
                        "Failed to read existing config at %s: %s", config_path, e
                    )
            # Ensure plain text API keys are migrated to environment and purged from JSON file
            for k in list(existing.keys()):
                if k.endswith("_api_key") or k == "cloud_api_key":
                    legacy_val = existing.pop(k, None)
                    if legacy_val:
                        try:
                            if "openrouter" in k:
                                save_secure_api_key(legacy_val, key_name="openrouter_api_key", env_var="OPENROUTER_API_KEY")
                            else:
                                env_var = k.upper()
                                save_secure_api_key(legacy_val, key_name=k, env_var=env_var)
                        except Exception:
                            pass
            if "override_agent_models" not in existing and "override_agent_models" not in config_copy:
                existing["override_agent_models"] = copy.deepcopy(DEFAULT_OVERRIDE_AGENT_MODELS)
            elif "override_agent_models" in config_copy and isinstance(config_copy["override_agent_models"], dict):
                base_overrides = copy.deepcopy(existing.get("override_agent_models") or DEFAULT_OVERRIDE_AGENT_MODELS)
                for role, role_cfg in config_copy["override_agent_models"].items():
                    if role in base_overrides and isinstance(role_cfg, dict) and isinstance(base_overrides[role], dict):
                        base_overrides[role].update(role_cfg)
                    else:
                        base_overrides[role] = role_cfg
                config_copy["override_agent_models"] = base_overrides
            if "sentence_attribution_model" in config_copy:
                existing["sentence_attribution_model"] = config_copy["sentence_attribution_model"]
            elif "sentence_attribution_model" not in existing:
                existing["sentence_attribution_model"] = DEFAULT_SENTENCE_ATTRIBUTION_MODEL
            existing.update(config_copy)
            with open(config_path, "w") as f:
                json.dump(existing, f, indent=2)
            stamp_config_mtime(zotero_data_dir)
            logger.info("Saved Zocrates config to %s", config_path)
            resolved_path = config_path.resolve()
            with _CONFIG_CACHE_LOCK:
                _CONFIG_CACHE.pop(resolved_path, None)
                _CONFIG_CACHE_FINGERPRINT.pop(resolved_path, None)
    except Exception as e:
        logger.warning("Failed to save config to %s: %s", config_path, e)

stamp_config_mtime 

stamp_config_mtime(zotero_data_dir=None)

Stamp the in-memory config mtime sentinel with the current mtime of zocrates_config.json.

Source code in zocrates/config.py 
458
459
460
461
462
463
464
465
466
467
468
469
def stamp_config_mtime(zotero_data_dir: Path | str | None = None) -> None:
    """Stamp the in-memory config mtime sentinel with the current mtime of zocrates_config.json."""
    global _last_config_mtime
    resolved_dir = resolve_zotero_data_dir(zotero_data_dir)
    if resolved_dir is None:
        return
    config_path = get_config_path(resolved_dir)
    try:
        if config_path.exists():
            _last_config_mtime = config_path.stat().st_mtime
    except OSError:
        pass