summary_generator

zocrates.rag.summary_generator

Document summary generator utility with LiteLLM OpenRouter integration and deterministic extractive lead text heuristics.

DEFAULT_SUMMARY_MAX_INPUT_CHARS module-attribute

DEFAULT_SUMMARY_MAX_INPUT_CHARS = 3500

DEFAULT_SUMMARY_MAX_TOKENS module-attribute

DEFAULT_SUMMARY_MAX_TOKENS = 500

DEFAULT_SUMMARY_MODEL module-attribute

DEFAULT_SUMMARY_MODEL = 'openrouter/nvidia/nemotron-3.5-lightning:free'

DEFAULT_SUMMARY_TIMEOUT_SECONDS module-attribute

DEFAULT_SUMMARY_TIMEOUT_SECONDS = 10.0

SUMMARY_SYSTEM_PROMPT module-attribute

SUMMARY_SYSTEM_PROMPT = 'You are an expert academic research assistant. Provide a concise, dense, factual summary (2-3 sentences, 100-150 words) of the following document. Focus on the primary research problem, core methodology/system architecture, and key findings or results. Do not include conversational filler, meta-commentary, or any thinking process/scratchpad. Output ONLY the final summary text directly.'

_BOILERPLATE_LINE_RE module-attribute

_BOILERPLATE_LINE_RE = re.compile('(@[a-zA-Z0-9._-]+\\.[a-zA-Z]+|\\barxiv:\\s*[\\d.]+|\\bdoi:\\s*10\\.\\d+|\\bdepartment of\\b|\\buniversity\\b|\\bfaculty of\\b|\\bschool of\\b|\\binstitute of\\b|\\bcopyright\\s*(?:©|\\(c\\)|20\\d\\d|19\\d\\d)|\\ball rights reserved\\b|\\bcreative commons\\b|\\bcc-by\\b)', re.IGNORECASE)

_CODE_FENCE_RE module-attribute

_CODE_FENCE_RE = re.compile('(```[\\s\\S]*?```|~~~[\\s\\S]*?~~~)', re.MULTILINE)

_FIGURE_BLOCK_RE module-attribute

_FIGURE_BLOCK_RE = re.compile('<figure\\b[^>]*>[\\s\\S]*?</figure>', re.IGNORECASE | re.MULTILINE)

_HTML_COMMENT_RE module-attribute

_HTML_COMMENT_RE = re.compile('<!--[\\s\\S]*?-->', re.MULTILINE)

_IN_BODY_ABSTRACT_RE module-attribute

_IN_BODY_ABSTRACT_RE = re.compile('(?:^|\\n)(?:#{1,6}\\s+|(?:\\*\\*|__)\\s*)Abstract(?:\\s*(?:\\*\\*|__)|:|\\s*\\n)([\\s\\S]*?)(?=(?:\\n#{1,6}\\s+|\\n<figure|\\Z))', re.IGNORECASE)

_MD_FORMAT_RE module-attribute

_MD_FORMAT_RE = re.compile('(\\*\\*|__|\\*|_|`|~~)')

_MD_LINK_RE module-attribute

_MD_LINK_RE = re.compile('\\[([^\\]]+)\\]\\([^)]+\\)')

_SENTENCE_END_RE module-attribute

_SENTENCE_END_RE = re.compile('([.!?])(?:\\s+|\\Z)')

logger module-attribute

logger = logging.getLogger(__name__)

_clean_markdown_prose

_clean_markdown_prose(text)

Strip markdown links, formatting, and excess whitespace from a text snippet.

Source code in zocrates/rag/summary_generator.py

def _clean_markdown_prose(text: str) -> str:
    """Strip markdown links, formatting, and excess whitespace from a text snippet."""
    if not text:
        return ""
    # Convert [text](url) -> text
    cleaned = _MD_LINK_RE.sub(r"\1", text)
    # Strip formatting markers (*, **, _, __, `, ~~)
    cleaned = _MD_FORMAT_RE.sub("", cleaned)
    # Decode HTML entities
    cleaned = html.unescape(cleaned)
    # Collapse multiple whitespaces/newlines to single space
    cleaned = re.sub(r"\s+", " ", cleaned).strip()
    return cleaned

_truncate_to_sentence_boundary

_truncate_to_sentence_boundary(text, max_chars=900, max_words=160)

Truncate text at the last complete sentence boundary within char and word budgets.

Source code in zocrates/rag/summary_generator.py

def _truncate_to_sentence_boundary(text: str, max_chars: int = 900, max_words: int = 160) -> str:
    """Truncate text at the last complete sentence boundary within char and word budgets."""
    cleaned = _clean_markdown_prose(text)
    if not cleaned:
        return ""

    words = cleaned.split()
    if len(words) > max_words:
        cleaned = " ".join(words[:max_words])

    if len(cleaned) > max_chars:
        cleaned = cleaned[:max_chars]

    # Find the last sentence end marker
    matches = list(_SENTENCE_END_RE.finditer(cleaned))
    if matches:
        last_match = matches[-1]
        truncated = cleaned[: last_match.end()].strip()
        if len(truncated.split()) >= 15:
            return truncated

    # If no clean sentence boundary found or too short, return cleaned text with trailing period if needed
    if cleaned and not cleaned.endswith((".", "!", "?")):
        return cleaned + "."
    return cleaned

extract_lead_text

extract_lead_text(text)

Extract substantive lead text or in-body abstract deterministically without an LLM.

Implements the 3-step heuristic

1. In-body 'Abstract' section detection (fast path).
2. Prose paragraph accumulation & boilerplate filtering (general path).
3. Safety fallback.

Parameters:

Name	Type	Description	Default
text	str	Converted Markdown document text (with or without frontmatter).	required

Returns:

Type	Description
str	Clean, informative extractive lead text string.

Source code in zocrates/rag/summary_generator.py

def extract_lead_text(text: str) -> str:
    """Extract substantive lead text or in-body abstract deterministically without an LLM.

    Implements the 3-step heuristic:
      1. In-body 'Abstract' section detection (fast path).
      2. Prose paragraph accumulation & boilerplate filtering (general path).
      3. Safety fallback.

    Args:
        text: Converted Markdown document text (with or without frontmatter).

    Returns:
        Clean, informative extractive lead text string.
    """
    body = strip_existing_frontmatter(text).strip()
    if not body:
        return "No abstract or summary available."

    # -----------------------------------------------------------------------
    # Step 1: In-body "Abstract" section detection (Fast Path)
    # -----------------------------------------------------------------------
    abs_match = _IN_BODY_ABSTRACT_RE.search(body)
    if abs_match:
        abstract_candidate = abs_match.group(1).strip()
        cleaned_abs = _clean_markdown_prose(abstract_candidate)
        word_count = len(cleaned_abs.split())
        if 20 <= word_count <= 400:
            return _truncate_to_sentence_boundary(cleaned_abs, max_chars=1200, max_words=200)

    # -----------------------------------------------------------------------
    # Step 2: Substantive lead prose paragraph accumulation (General Path)
    # -----------------------------------------------------------------------
    # Mask non-prose blocks
    masked = _CODE_FENCE_RE.sub("\n\n", body)
    masked = _HTML_COMMENT_RE.sub("\n\n", masked)
    masked = _FIGURE_BLOCK_RE.sub("\n\n", masked)

    paragraphs = re.split(r"\n\s*\n+", masked)
    collected_paras: list[str] = []
    total_words = 0

    for para in paragraphs:
        p = para.strip()
        if not p:
            continue

        # Skip ATX headings
        if p.startswith("#"):
            continue

        # Skip short navigation lines, link lists, or breadcrumbs
        if len(p) < 40 and ("[" in p or ">" in p or "|" in p):
            continue

        lines = [line.strip() for line in p.splitlines() if line.strip()]
        # Check if entire paragraph is boilerplate / metadata
        is_boilerplate = False
        for line in lines:
            if _BOILERPLATE_LINE_RE.search(line):
                is_boilerplate = True
                break
        if is_boilerplate:
            continue

        clean_p = _clean_markdown_prose(p)
        words = clean_p.split()
        # Only take substantive prose paragraphs (>= 15 words)
        if len(words) >= 15:
            collected_paras.append(clean_p)
            total_words += len(words)
            if total_words >= 90:
                break

    if collected_paras:
        combined = " ".join(collected_paras)
        result = _truncate_to_sentence_boundary(combined, max_chars=900, max_words=150)
        if result and len(result.split()) >= 15:
            return result

    # -----------------------------------------------------------------------
    # Step 3: Safety Fallback
    # -----------------------------------------------------------------------
    return "No abstract or summary available."

generate_summary

generate_summary(text, title=None, zotero_data_dir=None, model=None, timeout=DEFAULT_SUMMARY_TIMEOUT_SECONDS)

Generate a concise document summary using LiteLLM with extractive fallback.

Hierarchy

1. Model: Explicit override, configured summary_model, llm_model, or openai/gpt-oss-20b via OpenRouter.
2. Offline / Error Fallback: Deterministic extractive lead text.

Parameters:

Name	Type	Description	Default
text	str	Converted Markdown text.	required
title	str | None	Optional document title.	None
zotero_data_dir	Path | str | None	Optional Zotero data directory for loading config/keys.	None
model	str | None	Optional explicit model override.	None
timeout	float	Request timeout in seconds (default: 10s).	DEFAULT_SUMMARY_TIMEOUT_SECONDS

Returns:

Type	Description
tuple[str, str]	A 2-tuple of (summary_text, source_tag) where source_tag is 'generated' or 'extractive'.

Source code in zocrates/rag/summary_generator.py

def generate_summary(
    text: str,
    title: str | None = None,
    zotero_data_dir: Path | str | None = None,
    model: str | None = None,
    timeout: float = DEFAULT_SUMMARY_TIMEOUT_SECONDS,
) -> tuple[str, str]:
    """Generate a concise document summary using LiteLLM with extractive fallback.

    Hierarchy:
      1. Model: Explicit override, configured `summary_model`, `llm_model`, or `openai/gpt-oss-20b` via OpenRouter.
      2. Offline / Error Fallback: Deterministic extractive lead text.

    Args:
        text: Converted Markdown text.
        title: Optional document title.
        zotero_data_dir: Optional Zotero data directory for loading config/keys.
        model: Optional explicit model override.
        timeout: Request timeout in seconds (default: 10s).

    Returns:
        A 2-tuple of `(summary_text, source_tag)` where source_tag is 'generated' or 'extractive'.
    """
    body = strip_existing_frontmatter(text).strip()
    if not body:
        return ("No abstract or summary available.", "extractive")

    # Guardrail: Truncate input to lead text budget (~3500 chars)
    lead_input = body[:DEFAULT_SUMMARY_MAX_INPUT_CHARS]

    # Resolve OpenRouter API key and model preferences
    cfg = load_zocrates_config(zotero_data_dir)
    api_key = get_secure_api_key("openrouter_api_key", ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY"))

    resolved_model = (
        model
        or cfg.get("summary_model")
        or DEFAULT_SUMMARY_MODEL
    )
    if not resolved_model.startswith("openrouter/") and not resolved_model.startswith(("ollama/", "ollama_chat/")):
        resolved_model = f"openrouter/{resolved_model}"

    is_local = resolved_model.startswith(("ollama/", "ollama_chat/"))

    if not api_key and not is_local:
        logger.debug("No OpenRouter API key configured; falling back to deterministic extractive lead text.")
        extractive = extract_lead_text(body)
        return extractive, "extractive"

    if api_key or is_local:
        doc_context = f"Title: {title}\n\n" if title else ""
        user_prompt = f"{doc_context}Document Text (first section):\n{lead_input}\n\nSummary:"

        messages = [
            {"role": "system", "content": SUMMARY_SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt},
        ]

        try:
            logger.info("Attempting summary generation with %s ...", resolved_model)
            extra_body = {"reasoning": {"effort": "none"}} if resolved_model.startswith("openrouter/") else {}
            resp = litellm.completion(
                model=resolved_model,
                messages=messages,
                api_key=api_key,
                max_tokens=DEFAULT_SUMMARY_MAX_TOKENS,
                timeout=timeout,
                temperature=0.2,
                extra_body=extra_body,
            )
            raw_summary = (
                resp.choices[0].message.content
                if resp.choices and resp.choices[0].message
                else ""
            )
            cleaned = (raw_summary or "").strip()
            # Strip <think>...</think> tags (handling unclosed and pre-injected tags)
            cleaned = re.sub(r"<think>[\s\S]*?(?:</think>|$)", "", cleaned, flags=re.IGNORECASE).strip()
            cleaned = re.sub(r"^[\s\S]*?</think>", "", cleaned, flags=re.IGNORECASE).strip()
            # Strip "Here's a thinking process: ..." blocks
            cleaned = re.sub(
                r"(?i)(?:here'?s a thinking process|thinking process|thought process):[\s\S]*?(?=(?:\n\n[A-Z]|\n\n\*\*Summary|\n\nSummary|\Z))",
                "",
                cleaned,
            ).strip()
            # Strip leading "Summary:" label
            cleaned = re.sub(r"(?i)^\*?\*?(?:summary|final summary)\*?\*?:\s*", "", cleaned).strip()

            if cleaned and len(cleaned.split()) >= 8:
                # Ensure clean sentence ending
                if not cleaned.endswith((".", "!", "?")):
                    cleaned += "."
                return cleaned, "generated"
        except Exception as exc:
            logger.warning(
                "Summary generation failed with model %s: %s; falling back to extractive",
                resolved_model,
                exc,
            )

    # LLM attempt failed or no API key available: use deterministic extractive lead text
    logger.info("Using deterministic extractive lead text fallback for summary.")
    extractive = extract_lead_text(body)
    return extractive, "extractive"

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

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

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

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

strip_existing_frontmatter

strip_existing_frontmatter(markdown)

Strip leading YAML frontmatter block if present.

Parameters:

Name	Type	Description	Default
markdown	str	Input Markdown string.	required

Returns:

Type	Description
str	Markdown content with leading YAML frontmatter removed.

Source code in zocrates/rag/bundle_catalog.py

def strip_existing_frontmatter(markdown: str) -> str:
    """Strip leading YAML frontmatter block if present.

    Args:
        markdown: Input Markdown string.

    Returns:
        Markdown content with leading YAML frontmatter removed.
    """
    m = _FRONTMATTER_RE.match(markdown)
    if m:
        return markdown[m.end() :].lstrip("\r\n")
    return markdown