search_pipeline

zocrates.agents.library_search.search_pipeline

Dynamic search pipeline: retrieval loop and synthesis workflow node for Dialogue mode.

ACTIVE_CONTEXT_ITEMS module-attribute

ACTIVE_CONTEXT_ITEMS = 'active_context_items'

ACTIVE_CONTEXT_ITEMS_STR module-attribute

ACTIVE_CONTEXT_ITEMS_STR = 'active_context_items_str'

CACHED_TOCS module-attribute

CACHED_TOCS = 'cached_tocs'

HITL_SNAPSHOT module-attribute

HITL_SNAPSHOT = 'hitl_snapshot'

LIBRARY_SEARCH_AGENT_INSTRUCTION module-attribute

LIBRARY_SEARCH_AGENT_INSTRUCTION = '# Role and Objectives\n\nYou are the Search Subagent for the Zocrates literature retrieval pipeline (a plugin of the Zotero application).\n\nYour objectives are:\n- Retrieve and accumulate high-quality evidence to satisfy the specified search goal thoroughly.\n- Avoid wasteful or redundant retrieval actions while doing so.\n\n# Available Tools\n\n1. `search_corpus(query, item_key="active", chunk_type=None)`:\n   - Conducts hybrid semantic and keyword search across indexed items/papers.\n   - Set `item_key` to scope search:\n       - "active" (default): searches only the currently selected Zotero items (or entire library if none selected).\n       - None, "all", or "*": searches the ENTIRE library (unfiltered). Use this to discover related papers, do a broad literature search, or explore beyond actively selected items.\n       - Specific key(s) (e.g., item_key="HCZLU22L" or item_key=["HCZLU22L", "M5Z7SFWM"]): filters strictly to those items.\n   - Chunk types: \n       - Omit or pass None to search all chunk types (default).\n       - "text": narrative prose and body paragraphs.\n       - "frontmatter": inspect bibliographic details (authors, year, publication, etc).\n       - "figure" or "table": target images, figures, or tabular data.\n       - "equation" or "code": chunks with LaTeX equations or code blocks.  \n   \n2. `get_document_toc(item_key)`:\n   - Fetches the Table of Contents, full Abstract, Figures list, and Tables list with asset IDs (s0, s1..., f1..., t1...).\n   - The abstract is included in full in this tool\'s output; you do not need to re-fetch it separately.\n   - When the active item has multiple attachments or notes, the context will list each with its own key using\n     `[Attachment Key: ...]` or `[Note Key: ...]`. Use those specific keys directly with this tool to inspect\n     a particular attachment or note.\n\n3. `read_document(item_key, asset_id=None, page=None)`:\n   - Reads the paper\'s abstract/summary (\'abstract\'), exact sections (e.g., \'s2\'), figures with visual captions (\'f1\'), or tables with CSV data (\'t1\').\n   - You can pass multiple asset IDs (e.g., asset_id=[\'s2\', \'s3\', \'f1\']).\n   - Only request asset_id=\'abstract\' if you skipped `get_document_toc`; the ToC output already contains the abstract.\n   - Omitting `asset_id` performs a paginated full-document read (use `page` to continue). Reserve this for short documents or notes whose ToC has no section structure.\n   - `[Attachment Key: ...]` and `[Note Key: ...]` values from the active context are valid `item_key` arguments\n     for this tool and for `get_document_toc` and `search_corpus`.\n\n4. `get_related_items(item_key)`:\n   - Retrieves verified semantically related and user-curated connected papers from the literature network graph.\n   - Returns target paper keys, titles, confidence scores (0.50 - 1.0), and relation types (\'human_curated_and_hybrid_search\', \'human_curated\', \'hybrid_search\', \'shared_tags\').\n   - Use this tool when exploring literature adjacent to a paper or when investigating specific gaps.\n\n## Termination Tools\n\n1. `conclude_search(summary, notes=None, gaps=None)`:\n   - MANDATORY TERMINATION TOOL: Concludes the retrieval phase. Your summary is surfaced to the user and preserved as context for future turns; the Literature Synthesizer composes the final answer directly from the accumulated evidence.\n   - Call this tool as soon as you have collected the required evidence, or immediately upon exhausting your retrieval tool calls.\n   - If there are no gaps, do not specify the `gaps` parameter.\n   - Calling this tool concludes your turn. Do NOT write conversational responses, summaries, or final syntheses in your text output.\n\n2. `request_search_extension(justification, covered_summary, missing_aspects)`:\n   - Proactively request permission from the user to extend the retrieval tool budget and evidence capacity.\n   - Use this tool when you realize that crucial papers, sections, or comparative aspects cannot be thoroughly covered under the current limit, or when evidence volume approaches the ceiling (~80%).\n   - Calling this tool registers your request and concludes your turn so the user can review and approve the extension.\n\n# Retrieval Strategy Guidelines\n\n## General Strategies\n  - **Broad to Narrow**:\n    1. If the search goal has a broad scope (general question), start by doing an exploratory search.\n    2. Then drill down into individual papers to get more information as needed (focused search).\n  - **Narrow (to Broad, if necessary)**:\n    1. If the search goal is targeted to the items(s) in context, do a focused search on the individual paper(s).\n    2. If the search goal suggests a scope expansion is needed, or if the context items have substantial evidence gaps, then broaden the search through a graph traversal or exploratory search.\n\n## Search Workflows\n- **Exploratory Search (Scope: Whole-Library / Multi-Paper)**:\n  Use `search_corpus` for:\n  - Multi-paper comparative queries or when multiple active items are selected (scope via `item_key="active"` or a list of keys).\n  - Searching for specific unknown phrases, terminology, or quantitative findings when the target section cannot be identified from the Table of Contents.\n  - Entire-library exploration: pass item_key="all" or None to search the whole library when no active items are selected, OR to expand your search beyond the active context when items are selected. \n  - Distinct search angles: Each search should test a fundamentally different aspect or keyword terminology. Avoid repeating searches with slight paraphrases or minor synonym swaps of prior queries.  \n  - Search bibliographic metadata (titles, authors, years, publication) by specifying `chunk_type="frontmatter"`.\n- **Focused Search (Scope: Single Item / Specific Paper)**:\n  When investigating a specific paper or when a single item is active in context:\n  - Call `get_document_toc(item_key)` first to inspect the abstract, document structure, section headings, and asset IDs (unless the document\'s outline is already provided in your prompt — then skip straight to `read_document`).\n    - Assets can be sections, figures, and tables. Each has a corresponding asset_id, such as `s0`, `s1`..., `f1`..., `t1`...\n  - Then call `read_document(item_key, asset_id=[\'sX\', \'fY\', ...])` to fetch the targeted sections, figures, or tables.\n    - Batch multiple assets: Always pass a list of asset IDs in a single call (e.g., asset_id=[\'s2\', \'s3\', \'f1\']) rather than issuing separate calls per section.                                                   \n    - Surgical section selection: Target only the section(s) directly relevant to the question (e.g., Methodology or Results) rather than reading the entire document. Exception: for short documents or notes without a section structure, a paginated full read (asset_id omitted) is appropriate.\n  - When multiple items are active, ensure every active item is covered before concluding.\n  This structured workflow retrieves clean, coherent sections and is more efficient and accurate than vector search.\n- **User-Selected Focal Snippets**:\n  When `# Context: User-Selected Focal Snippets from Attachment` is present, each `<text_snippet>` and `<image_snippet>` contains a `key="..."` attribute identifying the parent document or attachment item. You can use this key directly with `read_document(item_key=...)` or `get_document_toc(item_key=...)` to retrieve surrounding sections, context, or tables.\n- **Paper Relations & Literature Graph Traversal**:\n  Use `get_related_items(item_key)` for incremental hopping through an item\'s graph of semantic relations to other papers.\n  - An item\'s Table of Contents notes if related items exist (`**Related Items**: N`). Call this tool to get an item\'s immediate related neighbors. \n  - After discovering relevant connected paper keys from this tool, follow up by inspecting their outlines (`get_document_toc`) and/or reading their key sections (`read_document`).\n  - If you think it\'s worthwhile to pursue a path along the graph, you can call `get_related_items` again to explore additional relationships.\n  - Each hop into a neighboring paper costs several tool calls; prioritize the 1-2 most promising neighbors (by title and confidence score) before drilling down.\n  \n## Caution: Early Exit on Negative Evidence\nIf 2-3 genuinely distinct searches return zero matching results or clearly unrelated papers, conclude via `conclude_search` and explicitly note the absence of literature in your summary. Do not exit after a single null result if a meaningfully different search angle remains untried.\n\n# Bounded Retrieval Rules\n\n- Strict Retrieval Tool Budget: You operate under a finite budget of retrieval tool calls (search_corpus, get_document_toc, read_document, get_related_items). \n  - The system tracks your usage and alerts you in tool responses as you approach or reach the limit. \n  - Termination tools (`conclude_search`, `request_search_extension`) are free: they do not consume your retrieval budget.\n  - If essential facets remain unaddressed as your budget is depleted, you may request an extension (`request_search_extension`). \n- Budget is a ceiling, not a quota: Most targeted questions require only a handful of tool calls (e.g., 1 ToC lookup + 1 batched section read, or 1 search + 1 targeted read). Conclude once the search goal is thoroughly addressed.\n- Retrieved evidence is automatically accumulated in the background across all turns. Do not re-read assets or repeat searches. Redundant calls burn your tool budget without adding new evidence.\n\n# Termination Protocol - Mandatory Termination Contract\n\n- You MUST call `conclude_search` or `request_search_extension` to report what you found and what remains unaddressed or missing.\n- Your text output must NEVER contain a final answer, explanation, or synthesis of the research question. Writing answers is strictly prohibited and breaks pipeline role separation; the Literature Synthesizer handles the final response.\n- As soon as your planned retrieval actions are complete, call `conclude_search` or `request_search_extension` and stop.\n'

LIBRARY_SYNTHESIZER_AGENT_INSTRUCTION module-attribute

LIBRARY_SYNTHESIZER_AGENT_INSTRUCTION = synthesizer_instruction.format(attribution_via_citations=with_citations)

LIBRARY_SYNTHESIZER_AGENT_INSTRUCTION_STANDARD module-attribute

LIBRARY_SYNTHESIZER_AGENT_INSTRUCTION_STANDARD = synthesizer_instruction.format(attribution_via_citations=without_citations)

MAX_AUDIT_HISTORY_TURNS module-attribute

MAX_AUDIT_HISTORY_TURNS = 5

MAX_INJECTED_TOC_ITEMS module-attribute

MAX_INJECTED_TOC_ITEMS = 3

RESEARCH_TURN_HISTORY module-attribute

RESEARCH_TURN_HISTORY = 'research_turn_history'

RETRIEVED_EVIDENCE module-attribute

RETRIEVED_EVIDENCE = 'retrieved_evidence'

TEMP_EXTENSION_COUNT module-attribute

TEMP_EXTENSION_COUNT = 'temp:extension_count'

TEMP_EXTENSION_REQUESTED module-attribute

TEMP_EXTENSION_REQUESTED = 'temp:extension_requested'

TEMP_EXTENSION_REQUEST_PASS module-attribute

TEMP_EXTENSION_REQUEST_PASS = 'temp:extension_request_pass'

TEMP_MAX_RERANK_CHUNKS module-attribute

TEMP_MAX_RERANK_CHUNKS = 'temp:max_rerank_chunks'

TEMP_RETRIEVAL_TOOL_BUDGET module-attribute

TEMP_RETRIEVAL_TOOL_BUDGET = 'temp:retrieval_tool_budget'

TEMP_RETRIEVAL_TOOL_COUNT module-attribute

TEMP_RETRIEVAL_TOOL_COUNT = 'temp:retrieval_tool_count'

TEMP_RETRIEVED_EVIDENCE module-attribute

TEMP_RETRIEVED_EVIDENCE = 'temp:retrieved_evidence'

TEMP_SEARCH_QUERY module-attribute

TEMP_SEARCH_QUERY = 'temp:search_query'

TEMP_SEARCH_REPORT module-attribute

TEMP_SEARCH_REPORT = 'temp:search_report'

TEMP_SEEN_ASSETS module-attribute

TEMP_SEEN_ASSETS = 'temp:seen_assets'

TEMP_SEEN_CHUNK_IDS module-attribute

TEMP_SEEN_CHUNK_IDS = 'temp:seen_chunk_ids'

TEMP_WORD_CEILING module-attribute

TEMP_WORD_CEILING = 'temp:word_ceiling'

USER_PROMPT module-attribute

USER_PROMPT = 'user_prompt'

USER_RESEARCH_PROFILE_STR module-attribute

USER_RESEARCH_PROFILE_STR = 'user_research_profile_str'

library_search module-attribute

library_search = search_library

logger module-attribute

logger = logging.getLogger(__name__)

search_pipeline_node module-attribute

search_pipeline_node = search_library_pipeline_node

_auto_generate_completion_report

_auto_generate_completion_report(search_query, evidence)

Synthesize a completion report when conclude_search is not explicitly called.

Source code in zocrates/agents/library_search/search_pipeline.py

def _auto_generate_completion_report(
    search_query: str,
    evidence: list[dict[str, Any]],
) -> dict[str, Any]:
    """Synthesize a completion report when `conclude_search` is not explicitly called."""
    total_words = sum(
        c.get("word_count")
        if c.get("word_count") is not None
        else len(c.get("text", "").split())
        for c in evidence
    )
    if not evidence:
        return {
            "summary": "Retrieval completed with 0 evidence chunks retrieved.",
            "notes": "No matching chunks found in active scope.",
            "gaps": [search_query],
            "source": "auto_generated",
        }

    unique_headings: list[str] = []
    seen_headings = set()
    for c in evidence:
        h = c.get("heading")
        if h and h not in seen_headings:
            seen_headings.add(h)
            unique_headings.append(h)

    chunk_types = sorted(list({c.get("chunk_type", "text") for c in evidence}))
    top_headings = unique_headings[:5]

    return {
        "summary": (
            f"Retrieval completed with {len(evidence)} evidence chunks "
            f"({', '.join(chunk_types)}) covering: {', '.join(top_headings) if top_headings else 'general content'} (~{total_words} words)."
        ),
        "notes": "Search concluded via auto-completion safeguard.",
        "gaps": [],
        "source": "auto_generated",
    }

_clean_user_facing_synthesis

_clean_user_facing_synthesis(text)

Post-process synthesizer output to deterministically strip thinking tags, leaked internal keys, or chunk IDs.

Source code in zocrates/agents/library_search/search_pipeline.py

def _clean_user_facing_synthesis(text: str) -> str:
    """Post-process synthesizer output to deterministically strip thinking tags, leaked internal keys, or chunk IDs."""
    if not text:
        return text

    # Strip thinking traces safely without mangling code blocks or legitimate inline mentions
    text = _strip_thinking_tags(text)

    # Strip (ITEMKEY: "Title") -> ("Title") or (ITEMKEY: Title) -> (Title)
    cleaned = re.sub(r"\([A-Z0-9]{8}:\s*([\"']?[^)]+[\"']?)\)", r"(\1)", text)

    # Strip [Key: ITEMKEY], [Attachment Key: ITEMKEY], or [Note Key: ITEMKEY]
    cleaned = re.sub(r"[\[\(](?:Attachment\s+|Note\s+)?Key:\s*[A-Za-z0-9_-]+[\]\)]\s*", "", cleaned)

    # Strip [Chunk: ITEMKEY:s1] or Chunk ITEMKEY:s1 or (Chunk: ITEMKEY:s1)
    cleaned = re.sub(r"[\[\(]?[Cc]hunk:?\s*[A-Z0-9]{8}:[a-z0-9]+[\]\)]?", "", cleaned)

    # Strip [Chunk: ITEMKEY::1] or Chunk ITEMKEY::1
    cleaned = re.sub(r"[\[\(]?[Cc]hunk:?\s*[A-Z0-9]{8}::\d+[\]\)]?", "", cleaned)

    # Clean up any resulting empty parentheses or multiple spaces
    cleaned = re.sub(r"\(\s*\)", "", cleaned)
    cleaned = re.sub(r"[ \t]{2,}", " ", cleaned)
    return cleaned.strip()

_enqueue_isolation_scope_closing_event async

_enqueue_isolation_scope_closing_event(ctx, *, result)

Emit a terminal finish_task event that closes ctx's isolation scope.

ADK's Runner._find_active_task_scope treats any non-empty isolation_scope as an unclosed in-flight task agent awaiting user input unless an event carrying a finish_task FunctionResponse (result "Task completed." or "Task failed.") is recorded for that scope. library_search stamps the pipeline's events with search_pipeline:{fc_id}, so without explicit closure ADK tags subsequent user turns with the stale scope and the coordinator (isolation_scope=None) drops them, leaving the coordinator stuck on the first turn's prompt. Emitting the closing event on the invocation context lets Runner._find_active_task_scope recognize the scope as finished.

Parameters:

Name	Type	Description	Default
ctx	Context	The dynamically executed search pipeline context.	required
result	str	Terminal finish_task result ("Task completed." or "Task failed.").	required

Source code in zocrates/agents/library_search/search_pipeline.py

async def _enqueue_isolation_scope_closing_event(
    ctx: Context,
    *,
    result: str,
) -> None:
    """Emit a terminal ``finish_task`` event that closes ``ctx``'s isolation scope.

    ADK's ``Runner._find_active_task_scope`` treats any non-empty
    ``isolation_scope`` as an unclosed in-flight task agent awaiting user input
    unless an event carrying a ``finish_task`` FunctionResponse (result
    ``"Task completed."`` or ``"Task failed."``) is recorded for that scope.
    ``library_search`` stamps the pipeline's events with
    ``search_pipeline:{fc_id}``, so without explicit closure ADK tags
    subsequent user turns with the stale scope and the coordinator
    (``isolation_scope=None``) drops them, leaving the coordinator stuck on the
    first turn's prompt. Emitting the closing event on the invocation context
    lets ``Runner._find_active_task_scope`` recognize the scope as finished.

    Args:
        ctx: The dynamically executed search pipeline context.
        result: Terminal ``finish_task`` result ("Task completed." or "Task failed.").
    """
    await enqueue_isolation_scope_closing_event(
        ctx,
        result=result,
        default_author="search_library_pipeline_node",
    )

_ensure_isolation_scope

_ensure_isolation_scope(ctx)

Recover and assign isolation scope to ctx and its invocation context if unset.

Source code in zocrates/agents/library_search/search_pipeline.py

def _ensure_isolation_scope(ctx: Context) -> str | None:
    """Recover and assign isolation scope to ctx and its invocation context if unset."""
    return ensure_isolation_scope(
        ctx,
        prefixes=("search_library_pipeline:", "search_pipeline:"),
    )

_get_applicable_cached_tocs

_get_applicable_cached_tocs(active_items, cached_tocs, max_items=MAX_INJECTED_TOC_ITEMS)

Return formatted cached TOC outlines only if 100% of active items are cached.

If active items are empty (whole library search), or if ANY active item lacks a cached TOC, returns an empty list to prevent asymmetric cognitive bias.

Source code in zocrates/agents/library_search/search_pipeline.py

def _get_applicable_cached_tocs(
    active_items: list[dict[str, Any]] | None,
    cached_tocs: dict[str, Any] | None,
    max_items: int = MAX_INJECTED_TOC_ITEMS,
) -> list[str]:
    """Return formatted cached TOC outlines only if 100% of active items are cached.

    If active items are empty (whole library search), or if ANY active item
    lacks a cached TOC, returns an empty list to prevent asymmetric cognitive bias.
    """
    if not active_items or not isinstance(active_items, list) or not cached_tocs:
        return []

    if len(active_items) > max_items:
        return []

    # Complete Active Coverage Gate: Every active item must be in cached_tocs
    for item in active_items:
        if not _is_item_cached(item, cached_tocs):
            return []

    outlines: list[str] = []
    seen_texts: set[str] = set()
    for item in active_items:
        entry = _resolve_cached_entry(item, cached_tocs)
        if not entry or not entry.get("outline_text"):
            return []
        text = entry["outline_text"]
        if text not in seen_texts:
            seen_texts.add(text)
            outlines.append(text)

    return outlines

_get_pipeline_interrupt_ids

_get_pipeline_interrupt_ids(ctx)

Scan session events and ctx.resume_inputs to return set of all pipeline interrupt IDs.

Source code in zocrates/agents/library_search/search_pipeline.py

def _get_pipeline_interrupt_ids(ctx: Context) -> set[str]:
    """Scan session events and ctx.resume_inputs to return set of all pipeline interrupt IDs."""
    return get_pipeline_interrupt_ids(ctx)

_has_context_continuity

_has_context_continuity(prior_active_items, current_active_items)

Return True if prior turn and current turn share active context scope.

Source code in zocrates/agents/library_search/search_pipeline.py

def _has_context_continuity(prior_active_items: Any, current_active_items: Any) -> bool:
    """Return True if prior turn and current turn share active context scope."""
    p_keys = {
        item.get("item_key") or item.get("key")
        for item in (prior_active_items or [])
        if isinstance(item, dict) and (item.get("item_key") or item.get("key"))
    }
    c_keys = {
        item.get("item_key") or item.get("key")
        for item in (current_active_items or [])
        if isinstance(item, dict) and (item.get("item_key") or item.get("key"))
    }
    # Both active: must share at least one paper
    if p_keys and c_keys:
        return bool(p_keys & c_keys)
    # Both whole library (empty active selection): continuous search thread
    if not p_keys and not c_keys:
        return True
    # Scope changed between single-paper and whole library: no continuity
    return False

_is_item_cached

_is_item_cached(item, cached_tocs)

Check if item is cached. Primary: parent_key (input_key alias). Secondary: child attachments.

Source code in zocrates/agents/library_search/search_pipeline.py

def _is_item_cached(item: dict[str, Any], cached_tocs: dict[str, Any]) -> bool:
    """Check if item is cached. Primary: parent_key (input_key alias). Secondary: child attachments."""
    if not item or not isinstance(item, dict) or not cached_tocs:
        return False
    parent_key = item.get("item_key") or item.get("key")
    if parent_key and parent_key in cached_tocs:
        return True
    children = item.get("children") or []
    for child in children:
        if isinstance(child, dict):
            c_key = child.get("item_key") or child.get("key")
            if c_key and c_key in cached_tocs:
                return True
    return False

_recover_isolation_scope

_recover_isolation_scope(ctx)

Recover the search pipeline's isolation scope from snapshot or session events if unset.

Source code in zocrates/agents/library_search/search_pipeline.py

def _recover_isolation_scope(ctx: Context) -> str | None:
    """Recover the search pipeline's isolation scope from snapshot or session events if unset."""
    return recover_isolation_scope(
        ctx,
        prefixes=("search_library_pipeline:", "search_pipeline:"),
    )

_resolve_cached_entry

_resolve_cached_entry(item, cached_tocs)

Resolve the cached TOC entry for an item, checking parent key then children.

Source code in zocrates/agents/library_search/search_pipeline.py

def _resolve_cached_entry(item: dict[str, Any], cached_tocs: dict[str, Any]) -> dict[str, Any] | None:
    """Resolve the cached TOC entry for an item, checking parent key then children."""
    if not item or not isinstance(item, dict) or not cached_tocs:
        return None
    parent_key = item.get("item_key") or item.get("key")
    if parent_key and parent_key in cached_tocs:
        return cached_tocs[parent_key]
    children = item.get("children") or []
    for child in children:
        if isinstance(child, dict):
            c_key = child.get("item_key") or child.get("key")
            if c_key and c_key in cached_tocs:
                return cached_tocs[c_key]
    return None

_rewrite_relative_markdown_images

_rewrite_relative_markdown_images(text, chunks, active=None)

Rewrite any relative images/ Markdown image links to zocrates-img:// URIs.

Source code in zocrates/agents/library_search/search_pipeline.py

def _rewrite_relative_markdown_images(
    text: str, chunks: list[dict[str, Any]], active: Any = None
) -> str:
    """Rewrite any relative images/ Markdown image links to zocrates-img:// URIs."""
    if not text or "images/" not in text:
        return text

    file_to_key: dict[str, str] = {}
    fallback_key = None
    for c in chunks:
        k = c.get("item_key")
        if k:
            if not fallback_key:
                fallback_key = k
            img_p = c.get("image_path")
            if img_p:
                from pathlib import Path
                fname = Path(str(img_p)).name
                file_to_key[fname] = k
                file_to_key[str(img_p)] = k
    if not fallback_key and active and isinstance(active, list):
        for a in active:
            if isinstance(a, dict) and (a.get("key") or a.get("item_key")):
                fallback_key = a.get("key") or a.get("item_key")
                break

    def _repl(m: re.Match) -> str:
        """Rewrite relative Markdown image links to zocrates-img custom protocol URIs.

        Args:
            m: Regular expression match object containing image prefix and relative path.

        Returns:
            Rewritten Markdown image syntax with resolved zocrates-img URI.
        """
        prefix = m.group(1)
        rel_path = m.group(2)
        from pathlib import Path
        fname = Path(rel_path).name
        matched_key = file_to_key.get(fname) or file_to_key.get(rel_path) or fallback_key
        if matched_key:
            return f"{prefix}zocrates-img://{matched_key}/{rel_path})"
        return m.group(0)

    return re.sub(r"(!\[[^\]]*\]\()\s*(?:\./)?(images/[^)\s]+)\)", _repl, text)

_run_search_pipeline async

_run_search_pipeline(ctx, node_input)

Core search-pipeline body: subagent resolution, retrieval, rerank, and synthesis.

Source code in zocrates/agents/library_search/search_pipeline.py

async def _run_search_pipeline(ctx: Context, node_input: str) -> str:
    """Core search-pipeline body: subagent resolution, retrieval, rerank, and synthesis."""
    state = ctx.state
    snapshot = state.get(HITL_SNAPSHOT)
    if snapshot and snapshot.get("search_query"):
        search_query = snapshot["search_query"]
    else:
        raw_query = node_input or state.get(TEMP_SEARCH_QUERY, "")
        # Sanitize search query: strip any accidental numeric citation brackets ([1], [2], [1, 2], etc.)
        # that the coordinator may have copied from conversation history into its delegated query.
        search_query = re.sub(
            r"\[(?:E\s?|Chunk\s?)?(\d{1,2}(?:\s*[,;]\s*\d{1,2})*)\]|【(\d{1,2}(?:\s*[,;]\s*\d{1,2})*)】",
            "",
            raw_query,
        ).strip()
    state[TEMP_SEARCH_QUERY] = search_query

    cfg = load_zocrates_config()
    h_cfg = cfg.get("search_harness") or {}
    hitl_enabled = bool(h_cfg.get("hitl_enabled", True))
    max_extensions = int(h_cfg.get("max_extension_count", 2))
    budget_increment = int(h_cfg.get("budget_extension_increment", 5))
    word_increment = int(h_cfg.get("word_ceiling_extension_increment", 3750))
    chunk_increment = int(h_cfg.get("rerank_chunks_extension_increment", 5))

    completed_extensions: list[int] = (
        list(snapshot.get("completed_extensions", []))
        if (snapshot and isinstance(snapshot, dict))
        else []
    )
    existing_interrupt_ids = _get_pipeline_interrupt_ids(ctx)

    if snapshot:
        ceilings = snapshot.get("ceilings", {})
        if state.get(TEMP_RETRIEVAL_TOOL_BUDGET) is None:
            state[TEMP_RETRIEVAL_TOOL_BUDGET] = ceilings.get("budget", h_cfg.get("max_retrieval_tools", 10))
        if state.get(TEMP_WORD_CEILING) is None:
            state[TEMP_WORD_CEILING] = ceilings.get("word_ceiling", h_cfg.get("hard_word_ceiling", 7500))
        if state.get(TEMP_MAX_RERANK_CHUNKS) is None:
            state[TEMP_MAX_RERANK_CHUNKS] = ceilings.get("max_chunks", h_cfg.get("max_rerank_chunks", 15))

    if state.get(TEMP_RETRIEVAL_TOOL_BUDGET) is None:
        state[TEMP_RETRIEVAL_TOOL_BUDGET] = int(h_cfg.get("max_retrieval_tools", 10))
    if state.get(TEMP_WORD_CEILING) is None:
        state[TEMP_WORD_CEILING] = int(h_cfg.get("hard_word_ceiling", 7500))
    if state.get(TEMP_MAX_RERANK_CHUNKS) is None:
        state[TEMP_MAX_RERANK_CHUNKS] = int(h_cfg.get("max_rerank_chunks", 15))
    if state.get(TEMP_EXTENSION_COUNT) is None:
        state[TEMP_EXTENSION_COUNT] = 0

    # Clear invocation user_content so ADK single_turn mode does not prepend
    # the root conversational chat message or _SINGLE_TURN_NUDGE into subagent requests.
    # Subagents receive ONLY their dedicated node_input.
    if hasattr(ctx, "_invocation_context") and ctx._invocation_context is not None:
        ctx._invocation_context.user_content = None

    # Subagent resolution: resolve from parent coordinator's _subagents if available
    parent_agent = (
        getattr(ctx._invocation_context, "agent", None)
        if hasattr(ctx, "_invocation_context")
        else None
    )
    subagents = getattr(parent_agent, "_subagents", {}) if parent_agent else {}
    search_agent = subagents.get("library_search")
    synth_agent = subagents.get("library_synthesizer")

    # Fallback to coordinator agent subagents when resuming directly into search_library_pipeline_node
    if (search_agent is None or synth_agent is None) and getattr(parent_agent, "name", None) in (
        "search_library_pipeline_node",
    ):
        from zocrates.agents.app import coordinator_agent

        if coordinator_agent and hasattr(coordinator_agent, "_subagents"):
            coord_subagents = getattr(coordinator_agent, "_subagents", {})
            if search_agent is None:
                search_agent = coord_subagents.get("library_search")
            if synth_agent is None:
                synth_agent = coord_subagents.get("library_synthesizer")

    # Fallback for standalone node execution outside of coordinator
    if search_agent is None or synth_agent is None:
        logger.warning(
            "Subagents could not be resolved from coordinator invocation context. "
            "Falling back to default standalone subagent construction via get_llm_model()."
        )
        from zocrates.llm import get_llm_model

        default_model = get_llm_model()
        if search_agent is None:
            search_agent = create_library_search_agent(default_model)
        if synth_agent is None:
            synth_agent = create_library_synthesizer_agent(default_model)

    active_items_str = state.get(ACTIVE_CONTEXT_ITEMS_STR, "")
    active_items = state.get(ACTIVE_CONTEXT_ITEMS)
    if active_items and isinstance(active_items, list):
        search_active_context_str = format_active_items(active_items, include_keys=True)
    else:
        search_active_context_str = active_items_str

    cached_tocs = state.get(CACHED_TOCS) or {}
    max_injected_tocs = int(h_cfg.get("max_injected_toc_items", MAX_INJECTED_TOC_ITEMS))
    applicable_tocs = _get_applicable_cached_tocs(active_items, cached_tocs, max_items=max_injected_tocs)

    outline_block = ""
    if applicable_tocs:
        joined_outlines = "\n\n".join(applicable_tocs)
        outline_block = (
            f"## Pre-Retrieved Document Outline(s) for Active Items:\n"
            f"{joined_outlines}\n\n"
            f"*(Note: Table of Contents is already provided above for these active items. "
            f"Do NOT call `get_document_toc` for them. You can call `read_document` using your chosen asset IDs with either key.)*\n\n"
        )

    focal_snippets_str = state.get("focal_snippets_str") or ""
    snippet_block = ""
    if focal_snippets_str.strip():
        s = focal_snippets_str.strip()
        if "# Context: User-Selected Focal Snippets from Attachment" not in s:
            snippet_block = f"# Context: User-Selected Focal Snippets from Attachment\n{s}"
        else:
            snippet_block = s

    input_parts: list[str] = []
    if search_active_context_str and search_active_context_str.strip():
        input_parts.append(f"# Context: Active Items Selected in Zotero\n{search_active_context_str.strip()}")
    if outline_block.strip():
        input_parts.append(outline_block.strip())
    if snippet_block.strip():
        input_parts.append(snippet_block.strip())
    input_parts.append(f"# Target Search Goal\n{search_query}")
    search_node_input = "\n\n".join(input_parts)

    active_isolation_scope = _ensure_isolation_scope(ctx)

    # Skip search_initial on resume since it was executed in the initial invocation
    curr_inv_id = getattr(ctx, "invocation_id", None) or (
        getattr(ctx._invocation_context, "invocation_id", None)
        if hasattr(ctx, "_invocation_context") else None
    )
    session = getattr(ctx, "session", None)
    if not session and hasattr(ctx, "_invocation_context"):
        session = getattr(ctx._invocation_context, "session", None)
    if not curr_inv_id and session and getattr(session, "events", None):
        for ev in reversed(session.events):
            if getattr(ev, "invocation_id", None):
                curr_inv_id = ev.invocation_id
                break

    has_initial_run = bool(snapshot)
    if not has_initial_run and curr_inv_id:
        if session and getattr(session, "events", None):
            for ev in session.events:
                if getattr(ev, "invocation_id", None) == curr_inv_id:
                    node_path = getattr(getattr(ev, "node_info", None), "path", "") or ""
                    if "search_initial" in node_path:
                        has_initial_run = True
                        break
    if not has_initial_run:
        await ctx.run_node(
            search_agent,
            node_input=search_node_input,
            run_id="search_initial",
            override_isolation_scope=active_isolation_scope,
        )

    # Deep Research Extension Loop
    extension_count = 0
    while True:
        evidence = state.get(TEMP_RETRIEVED_EVIDENCE) or (snapshot.get("evidence") if snapshot else None) or []
        search_report = state.get(TEMP_SEARCH_REPORT) or (snapshot.get("search_report") if snapshot else None)
        if search_report is None:
            search_report = _auto_generate_completion_report(
                search_query=search_query,
                evidence=evidence,
            )

        total_words = sum(
            c.get("word_count") if c.get("word_count") is not None else len(c.get("text", "").split())
            for c in evidence
        )
        current_calls = int(state.get(TEMP_RETRIEVAL_TOOL_COUNT) or (snapshot.get("tool_count") if snapshot else 0) or 0)
        current_budget = int(state.get(TEMP_RETRIEVAL_TOOL_BUDGET, 10))
        current_word_ceiling = int(state.get(TEMP_WORD_CEILING, 7500))
        current_max_chunks = int(state.get(TEMP_MAX_RERANK_CHUNKS, 15))
        target_pass = extension_count + 1
        req_pass = state.get(TEMP_EXTENSION_REQUEST_PASS)
        if req_pass is not None:
            extension_requested = (req_pass == target_pass)
        else:
            extension_requested = (
                bool(state.get(TEMP_EXTENSION_REQUESTED, False))
                or bool(snapshot.get("extension_requested", False) if snapshot else False)
            ) if extension_count == 0 else False
        gaps = search_report.get("gaps", [])

        target_int_id = f"interrupt_ext_{extension_count + 1}"
        is_known_checkpoint = target_int_id in existing_interrupt_ids
        is_historical_pass = bool(
            snapshot and extension_count < int(snapshot.get("extension_count", 0))
        ) or is_known_checkpoint

        should_trigger = (
            hitl_enabled
            and (extension_count < max_extensions)
            and (
                is_historical_pass
                or extension_requested
                or (bool(gaps) and (current_calls >= current_budget or total_words >= int(0.8 * current_word_ceiling)))
            )
        )

        if not should_trigger:
            break

        if is_historical_pass and snapshot and snapshot.get("trigger_reason"):
            trigger_reason = snapshot.get("trigger_reason")
        elif extension_requested or (is_historical_pass and snapshot and snapshot.get("extension_requested")):
            trigger_reason = "agent_request"
        elif current_calls >= current_budget:
            trigger_reason = "tool_budget_exhausted"
        elif total_words >= int(0.8 * current_word_ceiling):
            trigger_reason = "word_ceiling_approached"
        else:
            trigger_reason = "unknown"

        agent_justification = (
            search_report.get("notes")
            if (trigger_reason == "agent_request" and search_report.get("source") == "subagent_request")
            else None
        )

        # Save snapshot before checkpoint
        state[HITL_SNAPSHOT] = {
            "search_query": search_query,
            "evidence": evidence,
            "search_report": search_report,
            "extension_count": extension_count,
            "tool_count": current_calls,
            "extension_requested": extension_requested,
            "isolation_scope": active_isolation_scope,
            "completed_extensions": list(completed_extensions),
            "ceilings": {
                "budget": current_budget,
                "word_ceiling": current_word_ceiling,
                "max_chunks": current_max_chunks,
            },
            "seen_chunk_ids": list(state.get(TEMP_SEEN_CHUNK_IDS, [])),
            "seen_assets": list(state.get(TEMP_SEEN_ASSETS, [])),
            "trigger_reason": trigger_reason,
            "trigger_details": {
                "tool_calls_used": current_calls,
                "tool_budget": current_budget,
                "words_used": total_words,
                "word_ceiling": current_word_ceiling,
            },
            "agent_justification": agent_justification,
        }

        checkpoint_payload = {
            "type": "deep_research_extension",
            "extension_number": extension_count + 1,
            "max_extensions": max_extensions,
            "current_calls": current_calls,
            "current_words": total_words,
            "covered_summary": search_report.get("summary", ""),
            "missing_aspects": gaps,
            "trigger_reason": trigger_reason,
            "trigger_details": {
                "tool_calls_used": current_calls,
                "tool_budget": current_budget,
                "words_used": total_words,
                "word_ceiling": current_word_ceiling,
            },
            "agent_justification": agent_justification,
        }

        decision = await ctx.run_node(
            search_budget_checkpoint,
            node_input=checkpoint_payload,
            run_id=f"checkpoint_{extension_count}",
            override_isolation_scope=active_isolation_scope,
        )
        action = decision.get("action") if isinstance(decision, dict) else str(decision or "")

        if action != "extend_budget":
            # User chose "synthesize" or dismissed -> proceed directly to synthesis
            break

        # User granted extension
        new_budget = current_budget + budget_increment
        new_word_ceiling = current_word_ceiling + word_increment
        new_max_chunks = current_max_chunks + chunk_increment
        new_extension_count = extension_count + 1

        state[TEMP_RETRIEVAL_TOOL_BUDGET] = new_budget
        state[TEMP_WORD_CEILING] = new_word_ceiling
        state[TEMP_MAX_RERANK_CHUNKS] = new_max_chunks
        state[TEMP_EXTENSION_COUNT] = new_extension_count

        guidance = (decision.get("guidance") or "").strip() if isinstance(decision, dict) else ""
        gaps_list = "\n".join(f"- {g}" for g in gaps) if gaps else "- Continue targeted literature retrieval"
        continuation_prompt = (
            f"# Deep Research Extension Granted (Extension {new_extension_count} of {max_extensions})\n"
            "The user granted additional retrieval tool calls and evidence capacity.\n\n"
            f"## Remaining Gaps to Retrieve:\n{gaps_list}\n"
        )
        if guidance:
            continuation_prompt += f"\n## User Steering Guidance:\n{guidance}\n"

        if new_extension_count not in completed_extensions:
            await ctx.run_node(
                search_agent,
                node_input=continuation_prompt,
                run_id=f"search_extension_{new_extension_count}",
                override_isolation_scope=active_isolation_scope,
            )
            completed_extensions.append(new_extension_count)

        # Full-dict snapshot reassignment to ensure ADK State commits changes
        state[HITL_SNAPSHOT] = {
            "search_query": search_query,
            "evidence": state.get(TEMP_RETRIEVED_EVIDENCE) or evidence,
            "search_report": state.get(TEMP_SEARCH_REPORT) or search_report,
            "extension_count": new_extension_count,
            "tool_count": int(state.get(TEMP_RETRIEVAL_TOOL_COUNT) or current_calls),
            "extension_requested": False,
            "isolation_scope": active_isolation_scope,
            "completed_extensions": list(completed_extensions),
            "ceilings": {
                "budget": new_budget,
                "word_ceiling": new_word_ceiling,
                "max_chunks": new_max_chunks,
            },
            "seen_chunk_ids": list(state.get(TEMP_SEEN_CHUNK_IDS, [])),
            "seen_assets": list(state.get(TEMP_SEEN_ASSETS, [])),
            "trigger_reason": trigger_reason,
            "trigger_details": {
                "tool_calls_used": current_calls,
                "tool_budget": current_budget,
                "words_used": total_words,
                "word_ceiling": current_word_ceiling,
            },
            "agent_justification": agent_justification,
        }
        extension_count = new_extension_count

    evidence = state.get(TEMP_RETRIEVED_EVIDENCE) or (snapshot.get("evidence") if snapshot else []) or []

    # Active items abstract fallback: if search concluded without accumulating chunks
    # but active items are selected, backfill their abstracts so the synthesizer is not starved of evidence.
    if not evidence and active_items and isinstance(active_items, list):
        import json
        from zocrates.agents.common.toolkit import _resolve_bundle_dir
        for act in active_items:
            if not isinstance(act, dict):
                continue
            act_key = act.get("key") or act.get("item_key")
            if not act_key:
                continue
            try:
                resolved = _resolve_bundle_dir(act_key)
                if resolved:
                    b_dir, r_key, _ = resolved
                    toc_json_path = b_dir / "toc.json"
                    if toc_json_path.exists():
                        with open(toc_json_path, encoding="utf-8") as f:
                            t_data = json.load(f)
                        raw_abs = (t_data.get("abstract") or "").strip()
                        if raw_abs and raw_abs != "No abstract or summary available." and t_data.get("abstract_source") != "none":
                            evidence.append({
                                "chunk_id": f"{r_key}:abstract",
                                "item_key": r_key,
                                "title": act.get("title") or t_data.get("title", ""),
                                "chunk_type": "text",
                                "heading": "Abstract",
                                "text": raw_abs,
                                "word_count": len(raw_abs.split()),
                                "retrieval_method": "direct_read",
                                "cosine_similarity": None,
                                "rrf_score": None,
                                "rerank_score": None,
                                "image_path": None,
                                "structured_content": None,
                                "authors": act.get("authors") or act.get("creators"),
                                "year": act.get("year"),
                            })
            except Exception as exc:
                logger.debug("Failed active item abstract fallback: %s", exc)
        if evidence:
            state[TEMP_RETRIEVED_EVIDENCE] = evidence

    search_report = state.get(TEMP_SEARCH_REPORT) or (snapshot.get("search_report") if snapshot else None)
    if search_report is None:
        search_report = _auto_generate_completion_report(
            search_query=search_query,
            evidence=evidence,
        )
        state[TEMP_SEARCH_REPORT] = search_report

    # Rerank evidence with active dynamic ceilings
    active_max_chunks = int(state.get(TEMP_MAX_RERANK_CHUNKS, 15))
    active_word_ceiling = int(state.get(TEMP_WORD_CEILING, 7500))
    evidence = filter_substantive_evidence(evidence)
    ranked_chunks = await rerank_evidence(
        search_query,
        evidence,
        max_chunks=active_max_chunks,
        word_ceiling=active_word_ceiling,
    )
    evidence_str = format_retrieved_evidence_str(ranked_chunks)

    if active_items and isinstance(active_items, list):
        clean_active = format_active_items(active_items, include_keys=False)
    elif active_items_str and active_items_str.strip():
        clean_active = re.sub(
            r"(\[(?:Attachment\s+|Note\s+)?Key:\s*[^\]]+\]|\*\*Item Key\*\*:\s*[^\n]+)\s*",
            "",
            active_items_str.strip(),
        )
    else:
        clean_active = ""

    # Synthesize grounded answer
    user_query = state.get(USER_PROMPT)
    if not user_query:
        # Session event inspection fallback
        if session and getattr(session, "events", None) and curr_inv_id:
            for ev in session.events:
                if (
                    getattr(ev, "invocation_id", None) == curr_inv_id
                    and getattr(ev, "author", None) == "user"
                    and getattr(ev, "content", None)
                ):
                    parts = getattr(ev.content, "parts", None) or []
                    for p in parts:
                        if getattr(p, "text", None) and not getattr(p, "thought", False):
                            user_query = p.text.strip()
                            break
                    if user_query:
                        break
    if not user_query:
        user_query = search_query

    # Record research turn entry for predictive prompt continuity
    research_entry = {
        "invocation_id": curr_inv_id,
        "pipeline": "library_search",
        "user_query": user_query,
        "search_goal": search_query,
        "conclude_output": {
            "summary": search_report.get("summary", ""),
            "gaps": search_report.get("gaps", []),
            "notes": search_report.get("notes"),
        },
    }
    history = list(state.get(RESEARCH_TURN_HISTORY) or [])
    if history and history[-1].get("invocation_id") == curr_inv_id:
        history[-1] = research_entry
    else:
        history.append(research_entry)
    state[RESEARCH_TURN_HISTORY] = history[-5:]

    synth_parts: list[str] = []

    profile_str = str(state.get(USER_RESEARCH_PROFILE_STR) or "").strip()
    if profile_str:
        synth_parts.append(profile_str)

    prior_audit = state.get(RETRIEVED_EVIDENCE) or []
    if prior_audit and isinstance(prior_audit, list):
        last_turn = prior_audit[-1]
        prior_active = last_turn.get("active_context_items")
        if _has_context_continuity(prior_active, active_items):
            report = last_turn.get("search_report") or {}
            if report.get("source") not in ("auto_generated", "budget_exhaustion"):
                last_query = last_turn.get("search_query") or last_turn.get("user_query")
                last_summary = report.get("summary")
                if last_summary and last_query:
                    synth_parts.append(
                        f"# Prior Turn Context (Read-Only Reference)\n"
                        f"Previous Inquiry: \"{last_query}\"\n"
                        f"Previous Findings Summary: {last_summary}"
                    )

    synth_parts.append(f"# User Query\n{user_query}")
    if clean_active:
        synth_parts.append(f"# Context: Active Items Selected in Zotero\n{clean_active}")
    if focal_snippets_str.strip():
        clean_synth_snippets = strip_snippet_keys(focal_snippets_str).strip()
        if clean_synth_snippets:
            if "# Context: User-Selected Focal Snippets from Attachment" not in clean_synth_snippets:
                synth_parts.append(
                    f"# Context: User-Selected Focal Snippets from Attachment\n{clean_synth_snippets}"
                )
            else:
                synth_parts.append(clean_synth_snippets)
    synth_parts.append(f"# Target Search Goal\n{search_query}")
    synth_parts.append(f"# Retrieved Literature Evidence\n{evidence_str}")

    synth_prompt = "\n\n".join(synth_parts)

    try:
        from zocrates.config import (
            get_considered_sources_disclosure,
            get_references_section_enabled,
        )
        zotero_dir = resolve_zotero_data_dir()
        cfg = load_zocrates_config(zotero_dir)
        citations_enabled = bool(cfg.get("citations_enabled", True))
        references_enabled = bool(cfg.get("references_section_enabled", True))
        considered_disclosure = bool(cfg.get("considered_sources_disclosure", True))
        verification_enabled = bool(cfg.get("citation_verification_enabled", False))
        verification_threshold = float(cfg.get("citation_verification_threshold", 0.60))
    except Exception:
        zotero_dir = None
        cfg = {}
        citations_enabled = True
        references_enabled = True
        considered_disclosure = True
        verification_enabled = False
        verification_threshold = 0.60
    if hasattr(synth_agent, "instruction"):
        try:
            synth_agent.instruction = get_synthesizer_instruction(citations_enabled)
        except (AttributeError, TypeError):
            pass

    synth_response = await ctx.run_node(
        synth_agent,
        node_input=synth_prompt,
        use_as_output=True,
        override_isolation_scope=active_isolation_scope,
    )
    raw_synth_text = coerce_to_text(synth_response)
    raw_synth_text = _rewrite_relative_markdown_images(raw_synth_text, ranked_chunks, active_items)
    response_text = _clean_user_facing_synthesis(raw_synth_text)

    # Optional citation support verification via fast decision models
    support_probabilities: dict[int, list[float | None]] | dict[int, float | None] | None = None
    verification_telemetry: dict[str, Any] = {}
    if verification_enabled and ranked_chunks:
        try:
            from zocrates.agents.library_search.citation_verifier import verify_citations_support

            support_probabilities = await verify_citations_support(
                text=response_text,
                ranked_chunks=ranked_chunks,
                config=cfg,
                telemetry_out=verification_telemetry,
            )
        except Exception as ver_err:
            logger.warning("Error running citation verification in search pipeline: %s", ver_err)
            support_probabilities = None

    # Resolve citations & references in-process without out-of-band SQL writes
    citations: list[dict[str, Any]] = []
    references: list[dict[str, Any]] = []
    cited_chunk_ids: list[str] = []
    enriched_text: str | None = response_text

    if (citations_enabled or references_enabled) and ranked_chunks:
        try:
            from zocrates.agents.library_search.citation_resolver import resolve_citations_and_references

            resolved = resolve_citations_and_references(
                text=response_text,
                ranked_chunks=ranked_chunks,
                zotero_dir=zotero_dir,
                references_enabled=references_enabled,
                considered_disclosure=considered_disclosure,
                citations_enabled=citations_enabled,
                support_probabilities=support_probabilities,
                verification_threshold=verification_threshold,
            )
            response_text = resolved["enriched_text"]
            citations = resolved["citations"]
            references = resolved["references"]
            cited_chunk_ids = resolved["cited_chunk_ids"]
            enriched_text = response_text
        except Exception as res_err:
            logger.warning("Error resolving citations in search pipeline: %s", res_err)
            enriched_text = None

    # Append turn audit log to persistent session state (capped to MAX_AUDIT_HISTORY_TURNS)
    new_audit_entry = {
        "invocation_id": curr_inv_id,
        "user_query": user_query,
        "search_query": search_query,
        "active_context_items": list(active_items) if (active_items and isinstance(active_items, list)) else [],
        "evidence_chunks": ranked_chunks,
        "cited_chunk_ids": cited_chunk_ids,
        "citations": citations,
        "references": references,
        "search_report": search_report,
        "support_probabilities": support_probabilities,
        "citation_verification_telemetry": verification_telemetry or None,
        "enriched_text": enriched_text,
    }

    audit_history = list(state.get(RETRIEVED_EVIDENCE) or [])
    # Deduplicate on resume/retry: if last entry matches the same search_query and user_query, update it in-place
    if (
        audit_history
        and audit_history[-1].get("search_query") == search_query
        and audit_history[-1].get("user_query") == user_query
    ):
        audit_history[-1] = new_audit_entry
    else:
        audit_history.append(new_audit_entry)

    if len(audit_history) > MAX_AUDIT_HISTORY_TURNS:
        audit_history = audit_history[-MAX_AUDIT_HISTORY_TURNS:]

    state[RETRIEVED_EVIDENCE] = audit_history

    # Clean up ephemeral turn-level snapshot so it never leaks into subsequent turns
    state[HITL_SNAPSHOT] = None
    state[TEMP_EXTENSION_REQUESTED] = False
    state[TEMP_EXTENSION_REQUEST_PASS] = None
    state[TEMP_EXTENSION_COUNT] = None
    state[TEMP_RETRIEVAL_TOOL_BUDGET] = None
    state[TEMP_WORD_CEILING] = None
    state[TEMP_MAX_RERANK_CHUNKS] = None

    # Build a clean Markdown view with in-text parenthetical author-year citations
    # and clean Markdown bibliography for the coordinator agent.
    clean_coordinator_text = build_coordinator_view(
        text=response_text,
        citations=citations,
        references=references,
    )

    return clean_coordinator_text

_strip_thinking_tags

_strip_thinking_tags(text)

Safely strip reasoning/thinking scratchpad tags without mangling code blocks or inline references.

Source code in zocrates/agents/library_search/search_pipeline.py

def _strip_thinking_tags(text: str) -> str:
    """Safely strip reasoning/thinking scratchpad tags without mangling code blocks or inline references."""
    if not text or ("<think>" not in text and "</think>" not in text):
        return text

    code_blocks: list[str] = []

    def _save_code(m: re.Match) -> str:
        code_blocks.append(m.group(0))
        return f"\x00CODE_{len(code_blocks)-1}\x00"

    # Protect code blocks and inline backticks
    protected = re.sub(r"```[\s\S]*?```|`[^`\n]+`", _save_code, text)

    # 1. Strip closed thinking blocks outside code
    protected = re.sub(r"<think>[\s\S]*?</think>", "", protected)

    # 2. Strip orphaned closing tag at start of output (pre-injected prompt template remnants)
    protected = re.sub(r"^\s*[\s\S]*?</think>\s*", "", protected)

    # 3. Only strip unclosed <think> if it starts at the beginning of the text (token truncation of scratchpad)
    protected = re.sub(r"^\s*<think>[\s\S]*$", "", protected)

    # Restore code blocks
    def _restore_code(m: re.Match) -> str:
        idx = int(m.group(1))
        return code_blocks[idx]

    return re.sub(r"\x00CODE_(\d+)\x00", _restore_code, protected)

build_coordinator_view

build_coordinator_view(text, citations, references)

Build a clean, human-readable Markdown representation for the coordinator agent.

Substitutes in-text numeric chunk citation tokens ([1], [1, 2], etc.) with natural parenthetical author-year citations, masks code blocks, strips trailing HTML/reference blocks, and appends a clean Markdown two-tier bibliography without HTML tags or internal item keys.

Parameters:

Name	Type	Description	Default
text	str	Response text (may contain numeric citation tokens or existing sources).	required
citations	list[dict[str, Any]]	List of citation dicts containing metadata (index, authors, year, title).	required
references	list[dict[str, Any]]	List of reference dicts containing document metadata (tier, authors, year, title, publication).	required

Returns:

Type	Description
str	Clean Markdown string with parenthetical citations and clean Markdown bibliography.

Source code in zocrates/agents/library_search/citation_resolver.py

def build_coordinator_view(
    text: str,
    citations: list[dict[str, Any]],
    references: list[dict[str, Any]],
) -> str:
    """Build a clean, human-readable Markdown representation for the coordinator agent.

    Substitutes in-text numeric chunk citation tokens ([1], [1, 2], etc.) with
    natural parenthetical author-year citations, masks code blocks, strips
    trailing HTML/reference blocks, and appends a clean Markdown two-tier
    bibliography without HTML tags or internal item keys.

    Args:
        text: Response text (may contain numeric citation tokens or existing sources).
        citations: List of citation dicts containing metadata (index, authors, year, title).
        references: List of reference dicts containing document metadata (tier, authors, year, title, publication).

    Returns:
        Clean Markdown string with parenthetical citations and clean Markdown bibliography.
    """
    clean_text = text or ""
    # a. Strip any trailing reference headers/sections from text
    clean_text = re.sub(
        r"(?:\n+|^\s*)<div\s+class=[\"']zocrates-sources-section[\"'][\s\S]*$",
        "",
        clean_text,
        flags=re.IGNORECASE,
    )
    clean_text = re.sub(
        r"(?:\n{1,3}|^\s*)#{1,3}\s*(?:Cited Sources|References|Sources consulted)[\s\S]*$",
        "",
        clean_text,
        flags=re.IGNORECASE,
    ).rstrip()

    # b. Protect fenced code blocks (```...``` or ~~~...~~~) and inline code (`...`)
    code_placeholders: list[str] = []

    def _mask_code(m: re.Match) -> str:
        """Replace code blocks with unique placeholders during academic citation formatting.

        Args:
            m: Regular expression match object for the matched code block.

        Returns:
            Placeholder string representing the masked code.
        """
        idx = len(code_placeholders)
        code_placeholders.append(m.group(0))
        return f"@@ZOCRATES_CODE_MASK_{idx}@@"

    masked_text = _CODE_BLOCK_RE.sub(_mask_code, clean_text)

    # c. Substitute in-text numeric chunk citation tokens
    citations_list = citations or []

    def _replace_citation_token(m: re.Match) -> str:
        """Replace numeric citation tokens with formatted author-year citations.

        Args:
            m: Regular expression match object containing numeric citation tokens.

        Returns:
            Formatted parenthetical author-year citation string.
        """
        raw_content = m.group(1) if m.group(1) is not None else m.group(2)
        if not raw_content:
            return ""
        parts = [p.strip() for p in re.split(r"[,;]", raw_content) if p.strip()]
        indices: list[int] = []
        for p in parts:
            if p.isdigit():
                indices.append(int(p))

        labels: list[str] = []
        for idx in indices:
            c = None
            for item in citations_list:
                c_idx = item.get("index")
                if c_idx is not None and (c_idx == idx or str(c_idx) == str(idx)):
                    c = item
                    break
            if not c:
                continue

            authors = _format_authors_clean(c.get("authors")) if c.get("authors") else ""
            year = _format_year_clean(c.get("year")) if c.get("year") else ""
            title = str(c.get("title") or "").strip()

            if authors and year:
                lbl = f"{authors}, {year}"
            elif authors:
                lbl = authors
            elif year:
                lbl = str(year)
            elif title:
                lbl = title
            else:
                lbl = ""

            if lbl and lbl not in labels:
                labels.append(lbl)

        if labels:
            joined = "; ".join(labels)
            return f"({joined})"
        return ""

    masked_text = _IN_TEXT_CITATION_TOKEN_RE.sub(_replace_citation_token, masked_text)
    masked_text = re.sub(r"[ \t]+([.,;:!?])", r"\1", masked_text)
    masked_text = re.sub(r"[ \t]{2,}", " ", masked_text)

    # d. Restore masked code blocks
    for i, code_str in enumerate(code_placeholders):
        masked_text = masked_text.replace(f"@@ZOCRATES_CODE_MASK_{i}@@", code_str)

    # e. If references is non-empty, assemble clean Markdown bibliography
    t1_lines: list[str] = []
    t2_lines: list[str] = []

    if references:
        tier1_refs: list[dict[str, Any]] = []
        tier2_refs: list[dict[str, Any]] = []

        for ref in references:
            t = ref.get("tier")
            if t == 2 or t == "2":
                tier2_refs.append(ref)
            else:
                tier1_refs.append(ref)

        if tier1_refs:
            t1_lines.append("# Cited Sources")
            for ref in tier1_refs:
                meta, rest = _format_doc_citation(ref)
                entry = f"{meta}. {rest}" if meta else rest
                t1_lines.append(f"- {entry}")

        if tier2_refs:
            t2_lines.append("## Other Considered Sources")
            for ref in tier2_refs:
                meta, rest = _format_doc_citation(ref)
                entry = f"{meta}. {rest}" if meta else rest
                t2_lines.append(f"- {entry}")

    sections: list[str] = []
    if masked_text.strip():
        sections.append(masked_text.strip())
    if t1_lines:
        sections.append("\n".join(t1_lines))
    if t2_lines:
        sections.append("\n".join(t2_lines))

    # f. Return resulting string
    return "\n\n".join(sections)

coerce_to_text

coerce_to_text(response)

Extract plain text string from string, Content, Part, dict, Event, or object.

Extracts text parts while strictly suppressing reasoning/thought parts (where thought=True) and placing parts extraction ahead of .text to prevent internal thought tokens or raw object reprs from leaking into user-facing answers.

Source code in zocrates/agents/common/pipeline_base.py

def coerce_to_text(response: Any) -> str:
    """Extract plain text string from string, Content, Part, dict, Event, or object.

    Extracts text parts while strictly suppressing reasoning/thought parts (where
    `thought=True`) and placing `parts` extraction ahead of `.text` to prevent
    internal thought tokens or raw object reprs from leaking into user-facing answers.
    """
    if response is None:
        return ""
    if isinstance(response, str):
        return response
    if getattr(response, "thought", False):
        return ""
    if hasattr(response, "output") and response.output is not None:
        return coerce_to_text(response.output)
    if hasattr(response, "parts") and response.parts:
        return "".join(
            getattr(p, "text", "")
            for p in response.parts
            if getattr(p, "text", None) and not getattr(p, "thought", False)
        )
    if hasattr(response, "text") and response.text:
        return str(response.text)
    if hasattr(response, "content"):
        return coerce_to_text(response.content)
    if isinstance(response, dict):
        if response.get("thought", False):
            return ""
        if "text" in response:
            return str(response["text"])
        if "content" in response:
            return coerce_to_text(response["content"])
    return str(response)

conclude_search

conclude_search(summary, notes=None, gaps=None, tool_context=None)

Conclude the retrieval phase and hand off accumulated evidence to the synthesizer.

Call this tool as soon as you have collected the necessary evidence, or immediately upon exhausting your retrieval tool budget. Do NOT write final answers or syntheses.

Parameters:

Name	Type	Description	Default
summary	str	1-2 sentence overview of the evidence gathered.	required
notes	str | None	Optional observations regarding caveats or overall synthesis guidance.	None
gaps	list[str] | None	Optional list of missing facets or literature gaps. If there are no gaps, leave this parameter unspecified.	None

Returns:

Type	Description
dict[str, Any]	Confirmation dictionary indicating search has concluded.

Source code in zocrates/agents/common/tools.py

def conclude_search(
    summary: str,
    notes: str | None = None,
    gaps: list[str] | None = None,
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Conclude the retrieval phase and hand off accumulated evidence to the synthesizer.

    Call this tool as soon as you have collected the necessary evidence, or immediately
    upon exhausting your retrieval tool budget. Do NOT write final answers or syntheses.

    Args:
        summary: 1-2 sentence overview of the evidence gathered.
        notes: Optional observations regarding caveats or overall synthesis guidance.
        gaps: Optional list of missing facets or literature gaps.
            If there are no gaps, leave this parameter unspecified.

    Returns:
        Confirmation dictionary indicating search has concluded.
    """
    if isinstance(gaps, str):
        gap_list = [g.strip() for g in re.split(r"[,;\n]+", gaps) if g.strip()]
    elif isinstance(gaps, (list, tuple, set)):
        gap_list = [str(g).strip() for g in gaps if g]
    else:
        gap_list = []

    report = {
        "summary": str(summary or "").strip(),
        "notes": str(notes or "").strip() if notes else None,
        "gaps": gap_list,
        "source": "subagent",
    }
    if tool_context is not None:
        tool_context.state[TEMP_SEARCH_REPORT] = report
        if getattr(tool_context, "actions", None) is not None:
            tool_context.actions.skip_summarization = True

    return {
        "status": "search_concluded",
        "message": "Retrieval phase concluded. Accumulated evidence will now be synthesized.",
        "report": report,
    }

create_library_search_agent

create_library_search_agent(model=None, thinking_level='low', generate_content_config=None)

Create the stateless retrieval subagent equipped with the consolidated retrieval tool suite.

Source code in zocrates/agents/library_search/search_pipeline.py

def create_library_search_agent(
    model: Any = None,
    thinking_level: str | None = "low",
    generate_content_config: types.GenerateContentConfig | None = None,
) -> Agent:
    """Create the stateless retrieval subagent equipped with the consolidated retrieval tool suite."""
    if model is None:
        from zocrates.llm import get_llm_model

        model = get_llm_model()

    if generate_content_config is None and thinking_level is not None:
        generate_content_config = get_agent_generate_content_config(thinking_level)

    return Agent(
        name="library_search_agent",
        model=model,
        instruction=LIBRARY_SEARCH_AGENT_INSTRUCTION,
        tools=[
            FunctionTool(search_corpus),
            FunctionTool(get_document_toc),
            FunctionTool(read_document),
            FunctionTool(get_related_items),
            FunctionTool(conclude_search),
            FunctionTool(request_search_extension),
        ],
        generate_content_config=generate_content_config,
        mode="single_turn",
        include_contents="default",
    )

create_library_synthesizer_agent

create_library_synthesizer_agent(model=None, thinking_level='high', generate_content_config=None, citations_enabled=None)

Create the grounded literature synthesizer agent.

Source code in zocrates/agents/library_search/search_pipeline.py

def create_library_synthesizer_agent(
    model: Any = None,
    thinking_level: str | None = "high",
    generate_content_config: types.GenerateContentConfig | None = None,
    citations_enabled: bool | None = None,
) -> Agent:
    """Create the grounded literature synthesizer agent."""
    if model is None:
        from zocrates.llm import get_llm_model

        model = get_llm_model()

    if generate_content_config is None and thinking_level is not None:
        generate_content_config = get_agent_generate_content_config(thinking_level)

    if citations_enabled is None:
        try:
            citations_enabled = get_citations_enabled()
        except Exception:
            citations_enabled = True

    return Agent(
        name="library_synthesizer_agent",
        model=model,
        instruction=get_synthesizer_instruction(citations_enabled),
        generate_content_config=generate_content_config,
        mode="single_turn",
        include_contents="none",
    )

enqueue_isolation_scope_closing_event async

enqueue_isolation_scope_closing_event(ctx, *, result, default_author='search_pipeline_node')

Emit a terminal finish_task event that closes ctx's isolation scope.

ADK's Runner._find_active_task_scope treats any non-empty isolation_scope as an unclosed in-flight task agent awaiting user input unless an event carrying a finish_task FunctionResponse (result "Task completed." or "Task failed.") is recorded for that scope. Stamping the pipeline's events with an isolation scope without explicit closure causes ADK to tag subsequent user turns with the stale scope, causing the coordinator (isolation_scope=None) to drop them. Emitting the closing event on the invocation context lets Runner._find_active_task_scope recognize the scope as finished.

Parameters:

Name	Type	Description	Default
ctx	Context	The dynamically executed search pipeline context.	required
result	str	Terminal finish_task result ("Task completed." or "Task failed.").	required
default_author	str	Fallback author name if ctx.agent_name is unset.	'search_pipeline_node'

Source code in zocrates/agents/common/pipeline_base.py

async def enqueue_isolation_scope_closing_event(
    ctx: Context,
    *,
    result: str,
    default_author: str = "search_pipeline_node",
) -> None:
    """Emit a terminal ``finish_task`` event that closes ``ctx``'s isolation scope.

    ADK's ``Runner._find_active_task_scope`` treats any non-empty
    ``isolation_scope`` as an unclosed in-flight task agent awaiting user input
    unless an event carrying a ``finish_task`` FunctionResponse (result
    ``"Task completed."`` or ``"Task failed."``) is recorded for that scope.
    Stamping the pipeline's events with an isolation scope without explicit closure
    causes ADK to tag subsequent user turns with the stale scope, causing the coordinator
    (``isolation_scope=None``) to drop them. Emitting the closing event on the
    invocation context lets ``Runner._find_active_task_scope`` recognize the scope as finished.

    Args:
        ctx: The dynamically executed search pipeline context.
        result: Terminal ``finish_task`` result ("Task completed." or "Task failed.").
        default_author: Fallback author name if ctx.agent_name is unset.
    """
    if not getattr(ctx, "isolation_scope", None):
        return
    invocation_ctx = getattr(ctx, "_invocation_context", None)
    if invocation_ctx is None:
        return
    closing_event = Event(
        author=getattr(ctx, "agent_name", None) or default_author,
        isolation_scope=ctx.isolation_scope,
        invocation_id=getattr(invocation_ctx, "invocation_id", ""),
        branch=getattr(invocation_ctx, "branch", None),
        content=types.Content(
            role="tool",
            parts=[
                types.Part.from_function_response(
                    name="finish_task",
                    response={"result": result},
                )
            ],
        ),
    )
    # Never let a queueing failure mask or replace the synthesized turn result
    # or the original exception: this event is a best-effort scope-close marker.
    try:
        await invocation_ctx._enqueue_event(closing_event)
    except Exception as e:
        logger.warning("Failed to enqueue isolation scope closing event: %s", e)

ensure_isolation_scope

ensure_isolation_scope(ctx, prefixes)

Recover and assign isolation scope to ctx and its invocation context if unset.

Parameters:

Name	Type	Description	Default
ctx	Context	Workflow execution context.	required
prefixes	tuple[str, ...] | list[str] | str	Required tuple, list, or string of acceptable isolation scope prefix(es).	required

Source code in zocrates/agents/common/pipeline_base.py

def ensure_isolation_scope(
    ctx: Context,
    prefixes: tuple[str, ...] | list[str] | str,
) -> str | None:
    """Recover and assign isolation scope to ctx and its invocation context if unset.

    Args:
        ctx: Workflow execution context.
        prefixes: Required tuple, list, or string of acceptable isolation scope prefix(es).
    """
    scope = getattr(ctx, "isolation_scope", None) or recover_isolation_scope(ctx, prefixes=prefixes)
    if scope:
        try:
            ctx.isolation_scope = scope
        except (AttributeError, TypeError):
            pass
        if hasattr(ctx, "_invocation_context") and ctx._invocation_context is not None:
            try:
                ctx._invocation_context.isolation_scope = scope
            except (AttributeError, TypeError):
                pass
    return scope

filter_substantive_evidence

filter_substantive_evidence(chunks)

Filter out non-substantive evidence chunks (frontmatter, TOC).

Fallback guard: If filtering yields 2 or fewer substantive chunks (regardless of whether any remain), returns the original chunks so the synthesizer is not starved of evidence. This preserves frontmatter chunks when there is not enough substantive content to stand on its own (e.g. the query only matched 1-2 real sections alongside a frontmatter hit).

Source code in zocrates/agents/common/reranker.py

def filter_substantive_evidence(
    chunks: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Filter out non-substantive evidence chunks (frontmatter, TOC).

    Fallback guard: If filtering yields 2 or fewer substantive chunks (regardless
    of whether any remain), returns the original chunks so the synthesizer is not
    starved of evidence.  This preserves frontmatter chunks when there is not
    enough substantive content to stand on its own (e.g. the query only matched
    1-2 real sections alongside a frontmatter hit).
    """
    if not chunks:
        return []

    substantive = [c for c in chunks if is_substantive_chunk(c)]
    if len(substantive) <= 2:
        if len(chunks) > len(substantive):
            logger.debug(
                "Substantive filter left only %d chunk(s) (<= 2); falling back to original %d chunks.",
                len(substantive),
                len(chunks),
            )
        return chunks
    return substantive

format_active_items

format_active_items(active_items, include_keys=True)

Format active context items from session state for insertion into agent prompts.

Supports a hierarchical format when a parent item has multiple converted attachments and/or child notes. When a parent has exactly one converted attachment and no notes the output is byte-identical to the legacy flat format so existing prompts and tests are unaffected.

Parameters:

Name	Type	Description	Default
active_items	list[dict[str, Any]] | None	List of item dictionaries representing active selections in Zotero UI. Each entry may optionally include a children list of attachment/note dicts (see fast_api_app.py).	required
include_keys	bool	Whether to include [Key: ...], [Attachment Key: ...], and [Note Key: ...] identifiers. Set to False for synthesizer/coordinator prompts to hide internal keys.	True

Returns:

Type	Description
str	Formatted multi-line Markdown string.

Source code in zocrates/agents/common/prompts.py

def format_active_items(
    active_items: list[dict[str, Any]] | None,
    include_keys: bool = True,
) -> str:
    """Format active context items from session state for insertion into agent prompts.

    Supports a hierarchical format when a parent item has multiple converted
    attachments and/or child notes.  When a parent has exactly one converted
    attachment and no notes the output is byte-identical to the legacy flat
    format so existing prompts and tests are unaffected.

    Args:
        active_items: List of item dictionaries representing active selections in
            Zotero UI.  Each entry may optionally include a ``children`` list
            of attachment/note dicts (see ``fast_api_app.py``).
        include_keys: Whether to include ``[Key: ...]``, ``[Attachment Key: ...]``,
            and ``[Note Key: ...]`` identifiers.  Set to ``False`` for
            synthesizer/coordinator prompts to hide internal keys.

    Returns:
        Formatted multi-line Markdown string.
    """
    if not active_items:
        return "No active paper is currently selected in the Zotero UI."

    lines: list[str] = []
    for item in active_items:
        key = item.get("item_key") or item.get("key", "unknown")
        title = item.get("title", "Untitled")

        year = item.get("year", "")
        year_str = f" ({year})" if year else ""
        children: list[dict[str, Any]] = item.get("children") or []
        has_notes = any(c.get("is_note") for c in children)
        use_hierarchy = len(children) > 1 or has_notes

        if use_hierarchy:
            # --- Parent line ---
            if include_keys:
                lines.append(f'- [Key: {key}] Title: "{title}"{year_str}')
            else:
                lines.append(f'- Title: "{title}"{year_str}')

            # --- Child lines ---
            for child in children:
                c_key = child.get("item_key", "unknown")
                c_title = child.get("title", "Untitled")
                c_type = child.get("content_type", "")

                if child.get("is_note"):
                    word_count = child.get("word_count", 0)
                    inline_text: str | None = child.get("inline_text")
                    if inline_text:
                        # Short note: inline content as blockquote.
                        if include_keys:
                            lines.append(
                                f'  - [Note Key: {c_key}] "{c_title}" ({word_count} words) [text/note]:'
                            )
                        else:
                            lines.append(f'  - Note: "{c_title}":')
                        for inline_line in inline_text.splitlines():
                            lines.append(f"    > {inline_line}")
                    else:
                        # Longer note: reference only.
                        if include_keys:
                            lines.append(
                                f'  - [Note Key: {c_key}] "{c_title}" (~{word_count} words) [text/note]'
                            )
                        else:
                            lines.append(f'  - Note: "{c_title}" [text/note]')
                else:
                    # Attachment child.
                    if include_keys:
                        lines.append(f'  - [Attachment Key: {c_key}] "{c_title}" [{c_type}]')
                    else:
                        lines.append(f'  - Attachment: "{c_title}" [{c_type}]')
        else:
            # --- Flat format (backward-compatible, single attachment or no children) ---
            if include_keys:
                lines.append(f'- [Key: {key}] Title: "{title}"{year_str}')
            else:
                lines.append(f'- Title: "{title}"{year_str}')

    return "\n".join(lines)

format_retrieved_evidence_str

format_retrieved_evidence_str(evidence_chunks)

Format accumulated evidence chunks into a clean XML-tagged block.

Suitable for passing into agent prompts or sufficiency evaluators. Encloses chunks inside and tags to prevent Markdown boundary collisions, eliminate duplicate titles/headings, and suppress internal chunk IDs and scores from the prompt representation.

Parameters:

Name	Type	Description	Default
evidence_chunks	list[dict[str, Any]] | None	List of serialized evidence chunk dictionaries.	required

Returns:

Type	Description
str	XML string summarizing all retrieved evidence chunks.

Source code in zocrates/agents/common/prompts.py

def format_retrieved_evidence_str(evidence_chunks: list[dict[str, Any]] | None) -> str:
    """Format accumulated evidence chunks into a clean XML-tagged block.

    Suitable for passing into agent prompts or sufficiency evaluators.
    Encloses chunks inside <retrieved_evidence> and <chunk index="..." ...>
    tags to prevent Markdown boundary collisions, eliminate duplicate titles/headings,
    and suppress internal chunk IDs and scores from the prompt representation.

    Args:
        evidence_chunks: List of serialized evidence chunk dictionaries.

    Returns:
        XML string summarizing all retrieved evidence chunks.
    """
    if not evidence_chunks:
        return "<retrieved_evidence>\nNo evidence retrieved.\n</retrieved_evidence>"

    chunk_blocks: list[str] = []

    for idx, c in enumerate(evidence_chunks, start=1):
        title = str(c.get("title") or "").strip()
        heading = str(c.get("heading") or "").strip()
        ctype = str(c.get("chunk_type") or "text").strip()
        text = str(c.get("text") or "").strip()

        # Attribute construction
        attr_parts = [f'index="{idx}"']

        doc_name = title if title else "Untitled Document"
        attr_parts.append(f'title="{html.escape(doc_name, quote=True)}"')
        auth_str = _format_authors_clean(c.get("authors"))
        if auth_str:
            attr_parts.append(f'authors="{html.escape(auth_str, quote=True)}"')
        year_str = _format_year_clean(c.get("year"))
        if year_str:
            attr_parts.append(f'year="{html.escape(year_str, quote=True)}"')
        if heading:
            attr_parts.append(f'section="{html.escape(heading, quote=True)}"')
        if ctype:
            attr_parts.append(f'type="{html.escape(ctype, quote=True)}"')

        attr_str = " " + " ".join(attr_parts)

        # Body construction
        body_parts = []
        item_key_val = str(c.get("item_key") or "").strip()
        if text:
            # Rewrite any local relative image links in chunk text (e.g. ![Fig](images/foo.jpg))
            # to fully qualified zocrates-img://{item_key}/images/foo.jpg URIs so the synthesizer
            # always copies an actionable URI even when referencing inline section figures.
            if item_key_val and ("images/" in text or "./images/" in text):
                text = re.sub(
                    r"(!\[[^\]]*\]\()\s*(?:\./)?(images/[^)\s]+)\)",
                    rf"\1zocrates-img://{item_key_val}/\2)",
                    text,
                )
            body_parts.append(text)
        if c.get("image_path"):
            raw_image_path = str(c["image_path"]).strip()
            if item_key_val:
                # URL-encode the path component for resilience against spaces
                # in PDF-extracted filenames (Markdown link syntax breaks on
                # unencoded spaces).  safe="/" preserves path separators.
                encoded_path = urllib.parse.quote(raw_image_path, safe="/")
                display_uri = f"zocrates-img://{item_key_val}/{encoded_path}"
            else:
                # item_key missing: fall back to raw relative path so the chunk
                # is still informative, but no displayable URI will be emitted.
                display_uri = raw_image_path
            body_parts.append(f'<figure_image path="{html.escape(display_uri, quote=True)}" />')
        if c.get("structured_content"):
            body_parts.append(f'<table_csv>\n{str(c["structured_content"]).strip()}\n</table_csv>')

        body = "\n".join(body_parts)
        if body:
            chunk_blocks.append(f'  <chunk{attr_str}>\n{body}\n  </chunk>')
        else:
            chunk_blocks.append(f'  <chunk{attr_str} />')

    return "<retrieved_evidence>\n" + "\n".join(chunk_blocks) + "\n</retrieved_evidence>"

get_agent_generate_content_config

get_agent_generate_content_config(thinking_level=DEFAULT_THINKING_LEVEL, include_thoughts=True, thinking_budget=None, num_retries=None, zotero_data_dir=None, model=None)

Construct an ADK GenerateContentConfig with thinking/reasoning parameters.

Sets ADK's native thinking_config (for Google Gemini models) as well as http_options.extra_body (for LiteLLM / OpenRouter / OpenAI reasoning models) and http_options.retry_options for baked-in network retries.

Parameters:

Name	Type	Description	Default
thinking_level	str | ThinkingLevel | None	Thinking level ('minimal', 'low', 'medium', 'high', or None/disabled).	DEFAULT_THINKING_LEVEL
include_thoughts	bool	Whether thoughts are returned in the response stream.	True
thinking_budget	int | None	Optional token budget for reasoning (int or None).	None
num_retries	int | None	Optional maximum retry attempts for HTTP requests (defaults to llm_num_retries in config or DEFAULT_LLM_NUM_RETRIES = 4).	None
zotero_data_dir	Path | str | None	Optional root Zotero data directory for loading config.	None
model	str | Any | None	Optional model identifier or LLM instance to tailor reasoning schema and Ollama thinking gating.	None

Returns:

Type	Description
GenerateContentConfig | None	Configured GenerateContentConfig, or None if thinking is disabled.

Source code in zocrates/agents/common/config.py

def get_agent_generate_content_config(
    thinking_level: str | types.ThinkingLevel | None = DEFAULT_THINKING_LEVEL,
    include_thoughts: bool = True,
    thinking_budget: int | None = None,
    num_retries: int | None = None,
    zotero_data_dir: Path | str | None = None,
    model: str | Any | None = None,
) -> types.GenerateContentConfig | None:
    """Construct an ADK GenerateContentConfig with thinking/reasoning parameters.

    Sets ADK's native thinking_config (for Google Gemini models) as well as
    http_options.extra_body (for LiteLLM / OpenRouter / OpenAI reasoning models)
    and http_options.retry_options for baked-in network retries.

    Args:
        thinking_level: Thinking level ('minimal', 'low', 'medium', 'high', or None/disabled).
        include_thoughts: Whether thoughts are returned in the response stream.
        thinking_budget: Optional token budget for reasoning (int or None).
        num_retries: Optional maximum retry attempts for HTTP requests (defaults to llm_num_retries in config or DEFAULT_LLM_NUM_RETRIES = 4).
        zotero_data_dir: Optional root Zotero data directory for loading config.
        model: Optional model identifier or LLM instance to tailor reasoning schema and Ollama thinking gating.

    Returns:
        Configured GenerateContentConfig, or None if thinking is disabled.
    """
    if thinking_level is None:
        return None

    level_str = str(thinking_level).lower().replace("thinkinglevel.", "").strip()
    if level_str in ("none", "off", "false", "disabled", ""):
        return None

    thinking_config = types.ThinkingConfig(
        thinking_level=level_str,
        include_thoughts=include_thoughts,
        thinking_budget=thinking_budget,
    )

    # Map "minimal" to "low" for OpenAI / vLLM reasoning_effort schemas (avoiding HTTP 422)
    openai_reasoning_effort = "low" if level_str == "minimal" else level_str

    extra_body: dict[str, Any] = {
        "reasoning": {"effort": level_str},
        "reasoning_effort": openai_reasoning_effort,
    }
    if thinking_budget is not None:
        extra_body["reasoning"]["max_tokens"] = thinking_budget

    # Inspect config and model to gate think: true on Ollama models
    try:
        cfg = load_zocrates_config(zotero_data_dir)
    except Exception:
        cfg = {}

    model_name_str = None
    if model is not None:
        model_name_str = getattr(model, "model", None) or str(model)
    elif cfg:
        model_name_str = cfg.get("llm_model")

    provider = str(cfg.get("llm_provider") or "").lower()
    is_ollama = (
        (model_name_str and (
            model_name_str.startswith("ollama/")
            or model_name_str.startswith("ollama_chat/")
        ))
        or provider == "ollama"
    )

    if is_ollama:
        base_url = cfg.get("llm_base_url")
        if is_ollama_thinking_supported(model_name_str, base_url=base_url, zotero_data_dir=zotero_data_dir):
            extra_body["think"] = True
        else:
            # Model does not support thinking; do not send think: true (avoiding Ollama HTTP 400)
            extra_body.pop("think", None)

    if num_retries is None:
        try:
            val = cfg.get("llm_num_retries", DEFAULT_LLM_NUM_RETRIES)
            effective_retries = int(val) if val is not None else DEFAULT_LLM_NUM_RETRIES
        except Exception:
            effective_retries = DEFAULT_LLM_NUM_RETRIES
    else:
        effective_retries = int(num_retries)

    return types.GenerateContentConfig(
        thinking_config=thinking_config,
        http_options=types.HttpOptions(
            extra_body=extra_body,
            retry_options=types.HttpRetryOptions(attempts=effective_retries),
        ),
    )

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

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_document_toc

get_document_toc(item_key, tool_context=None)

Retrieve the Title, Abstract, Table of Contents, Figures, and Tables for a given item.

Parameters:

Name	Type	Description	Default
item_key	str	Unique Zotero item key.	required

Returns:

Type	Description
dict[str, Any]	Dictionary with 'status', 'item_key', and 'toc_content'.

Source code in zocrates/agents/common/tools.py

def get_document_toc(
    item_key: str,
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Retrieve the Title, Abstract, Table of Contents, Figures, and Tables for a given item.

    Args:
        item_key: Unique Zotero item key.

    Returns:
        Dictionary with 'status', 'item_key', and 'toc_content'.
    """
    allowed, err_msg, notice = _check_and_increment_tool_budget(tool_context)
    if not allowed:
        return {"status": "budget_exhausted", "error": err_msg}

    # Cycle suppression check for item_key before disk reading
    if tool_context is not None:
        seen_tocs = tool_context.state.get(TEMP_SEEN_TOC_ITEM_KEYS, [])
        if item_key in seen_tocs:
            res: dict[str, Any] = {
                "status": "already_inspected",
                "item_key": item_key,
                "message": (
                    f"TOC for {item_key} was already retrieved in this turn. "
                    "Refer to prior context or use read_document for specific sections."
                ),
            }
            if notice:
                _ensure_preliminary_report(tool_context, force=True)
                res["budget_notice"] = notice
            return res

    # Resolve bundle_dir and canonical resolved_key
    resolved_key = item_key
    bundle_dir = None
    try:
        resolved = _resolve_bundle_dir(item_key)
        if resolved:
            bundle_dir, r_key, _ = resolved
            resolved_key = r_key or resolved_key
    except Exception as exc:
        logger.debug("Failed to resolve bundle dir for TOC: %s", exc)

    # Check cached_tocs liveness: evict if item was deleted in Zotero
    if tool_context is not None:
        # ``or {}`` guards against an explicit None persisted in session state
        # (``.get(key, {})`` returns None when the key exists with a None value).
        cached_tocs = dict(tool_context.state.get(CACHED_TOCS) or {})
        if item_key in cached_tocs or resolved_key in cached_tocs:
            try:
                from zocrates.agents.common.toolkit import get_syncer
                syncer = get_syncer()
                is_active = syncer.db.is_item_active(resolved_key) if (syncer and syncer.db) else True
            except Exception:
                is_active = True
            if not is_active:
                cached_tocs.pop(item_key, None)
                cached_tocs.pop(resolved_key, None)
                tool_context.state[CACHED_TOCS] = cached_tocs
                return {
                    "status": "error",
                    "item_key": item_key,
                    "error": f"Document '{item_key}' has been deleted from the library.",
                }

    # Cycle suppression check for canonical resolved_key before disk reading
    if tool_context is not None:
        seen_tocs = tool_context.state.get(TEMP_SEEN_TOC_ITEM_KEYS, [])
        if resolved_key in seen_tocs:
            res = {
                "status": "already_inspected",
                "item_key": item_key,
                "message": (
                    f"TOC for {item_key} was already retrieved in this turn. "
                    "Refer to prior context or use read_document for specific sections."
                ),
            }
            if notice:
                _ensure_preliminary_report(tool_context, force=True)
                res["budget_notice"] = notice
            return res

    res = get_document_toc_core(item_key=item_key)

    if isinstance(res, dict) and "item_key" in res:
        resolved_key = res.get("item_key") or resolved_key

    if res.get("status") == "success" and tool_context is not None:
        # Track seen TOC keys for cycle suppression
        seen_tocs = list(tool_context.state.get(TEMP_SEEN_TOC_ITEM_KEYS, []))
        for k in (item_key, resolved_key):
            if k and k not in seen_tocs:
                seen_tocs.append(k)
        abstract_text = ""
        if bundle_dir:
            toc_json_path = bundle_dir / "toc.json"
            if toc_json_path.exists():
                try:
                    import json
                    with open(toc_json_path, encoding="utf-8") as f:
                        toc_data = json.load(f)
                        pk = toc_data.get("parent_item_key")
                        if pk and pk not in seen_tocs:
                            seen_tocs.append(pk)
                        raw_abs = (toc_data.get("abstract") or "").strip()
                        if raw_abs and raw_abs != "No abstract or summary available." and toc_data.get("abstract_source") != "none":
                            abstract_text = raw_abs
                except Exception:
                    pass
        tool_context.state[TEMP_SEEN_TOC_ITEM_KEYS] = seen_tocs

        input_key = item_key
        doc_title = res.get("title") or ""
        parent_key = input_key if input_key != resolved_key else ""

        # Accumulate retrieved abstract into TEMP_RETRIEVED_EVIDENCE so the synthesizer
        # is never starved of evidence when the search agent inspects TOC / abstract.
        if not abstract_text and res.get("toc_content"):
            match = re.search(r"## Abstract\s*\n(.*?)(?=\n## |\Z)", str(res["toc_content"]), re.DOTALL)
            if match:
                raw_abs = match.group(1).strip()
                if raw_abs and raw_abs != "No abstract or summary available.":
                    abstract_text = raw_abs

        if abstract_text:
            doc_authors = None
            doc_year = None
            if bundle_dir:
                try:
                    from zocrates.agents.common.toolkit import get_syncer
                    syncer = get_syncer()
                    if syncer and getattr(syncer, "processor", None):
                        meta = syncer.processor.get_metadata(resolved_key) or {}
                        doc_authors = meta.get("authors") or meta.get("creators")
                        doc_year = meta.get("year")
                except Exception:
                    pass
            if doc_authors is None or doc_year is None:
                active_items = tool_context.state.get(ACTIVE_CONTEXT_ITEMS) or []
                for item in active_items:
                    if isinstance(item, dict):
                        keys = [item.get("key"), item.get("item_key")]
                        for att in item.get("attachments", []):
                            keys.append(att.get("key"))
                        if resolved_key in keys or item_key in keys:
                            if doc_authors is None:
                                doc_authors = item.get("authors") or item.get("creators")
                            if doc_year is None:
                                doc_year = item.get("year")
                            break

            current_evidence = list(tool_context.state.get(TEMP_RETRIEVED_EVIDENCE, []))
            synthetic_chunk_id = f"{resolved_key}:abstract"
            if not any(c.get("chunk_id") == synthetic_chunk_id for c in current_evidence):
                chunk_dict: dict[str, Any] = {
                    "chunk_id": synthetic_chunk_id,
                    "item_key": resolved_key,
                    "title": doc_title or res.get("title", ""),
                    "chunk_type": "text",
                    "heading": "Abstract",
                    "text": abstract_text,
                    "word_count": len(abstract_text.split()),
                    "retrieval_method": "direct_read",
                    "cosine_similarity": None,
                    "rrf_score": None,
                    "rerank_score": None,
                    "image_path": None,
                    "structured_content": None,
                }
                if doc_authors is not None:
                    chunk_dict["authors"] = doc_authors
                if doc_year is not None:
                    chunk_dict["year"] = doc_year
                current_evidence.append(chunk_dict)
                tool_context.state[TEMP_RETRIEVED_EVIDENCE] = current_evidence
        outline_text = ""
        if bundle_dir:
            try:
                outline_text = extract_compact_toc_outline(
                    bundle_dir,
                    doc_title,
                    resolved_key,
                    parent_key=parent_key,
                )
            except Exception as exc:
                logger.debug("Failed to extract compact toc outline: %s", exc)

        if not outline_text and res.get("toc_content"):
            eff_pk = parent_key if parent_key != resolved_key else ""
            key_bracket = f"[Key: {resolved_key}, Parent Key: {eff_pk}]" if eff_pk else f"[Key: {resolved_key}]"
            title_part = f' "{doc_title}"' if doc_title else ""
            lines = [f"### {key_bracket}{title_part}"]
            for raw_line in str(res["toc_content"]).splitlines():
                s = raw_line.strip()
                if s.startswith("- [") or s.startswith("## Figures") or s.startswith("## Tables"):
                    lines.append(s)
            outline_text = "\n".join(lines)

        if outline_text:
            cached_entry = {
                "input_key": input_key,
                "resolved_key": resolved_key,
                "title": doc_title,
                "outline_text": outline_text,
                "updated_at": time.time(),
            }

            # ``or {}`` guards against an explicit None persisted in session state
            # (see matching comment above).
            cached_tocs = dict(tool_context.state.get(CACHED_TOCS) or {})
            cached_tocs[input_key] = cached_entry
            if resolved_key != input_key:
                cached_tocs[resolved_key] = cached_entry

            pruned_tocs = _prune_cached_tocs(cached_tocs, max_entries=MAX_UNIQUE_CACHED_TOCS)
            tool_context.state[CACHED_TOCS] = pruned_tocs

    # Dynamic TOC relation count injection (P1-3)
    if isinstance(res, dict) and res.get("status") == "success" and "toc_content" in res:
        rel_count = 0
        cache_dir = bundle_dir.parent if bundle_dir else None
        if not cache_dir:
            try:
                from zocrates.agents.common.toolkit import get_syncer
                syncer = get_syncer()
                if syncer and getattr(syncer, "processor", None):
                    cache_dir = getattr(syncer.processor, "_cache_dir", None)
            except Exception:
                pass

        if cache_dir:
            meta_db_path = cache_dir / "conversion_meta.db"
            if meta_db_path.exists():
                try:
                    with contextlib.closing(sqlite3.connect(str(meta_db_path), timeout=5.0)) as meta_conn:
                        meta_conn.execute("PRAGMA busy_timeout = 5000;")
                        row = meta_conn.execute(
                            "SELECT COUNT(DISTINCT target_key) FROM item_relations WHERE source_key = ?",
                            (resolved_key,),
                        ).fetchone()
                        if row:
                            rel_count = row[0]
                except Exception as exc:
                    logger.debug("Failed to query item_relations count for %s: %s", resolved_key, exc)

        toc_str = str(res["toc_content"])
        rel_line = f"**Related Items**: {rel_count}"
        if "## Table of Contents" in toc_str:
            idx = toc_str.find("## Table of Contents")
            prefix = toc_str[:idx].rstrip()
            suffix = toc_str[idx:]
            res["toc_content"] = (prefix + "\n\n" if prefix else "") + rel_line + "\n\n" + suffix
        elif "## Sections" in toc_str:
            idx = toc_str.find("## Sections")
            prefix = toc_str[:idx].rstrip()
            suffix = toc_str[idx:]
            res["toc_content"] = (prefix + "\n\n" if prefix else "") + rel_line + "\n\n" + suffix
        else:
            res["toc_content"] = toc_str.rstrip() + f"\n\n{rel_line}\n"

    if notice and tool_context is not None:
        _ensure_preliminary_report(tool_context, force=True)
        res["budget_notice"] = notice
    return res

get_pipeline_interrupt_ids

get_pipeline_interrupt_ids(ctx)

Scan session events and ctx.resume_inputs to return set of all pipeline interrupt IDs.

Source code in zocrates/agents/common/pipeline_base.py

def get_pipeline_interrupt_ids(ctx: Context) -> set[str]:
    """Scan session events and ctx.resume_inputs to return set of all pipeline interrupt IDs."""
    interrupts: set[str] = set()
    resume_inputs = getattr(ctx, "resume_inputs", None)
    if callable(resume_inputs):
        try:
            resume_inputs = resume_inputs()
        except Exception:
            resume_inputs = None
    if isinstance(resume_inputs, dict):
        for k in resume_inputs:
            if isinstance(k, str) and k.startswith("interrupt_ext_"):
                interrupts.add(k)

    curr_inv_id = getattr(ctx, "invocation_id", None) or (
        getattr(ctx._invocation_context, "invocation_id", None)
        if hasattr(ctx, "_invocation_context")
        else None
    )
    session = getattr(ctx, "session", None)
    if not session and hasattr(ctx, "_invocation_context"):
        session = getattr(ctx._invocation_context, "session", None)
    if session and getattr(session, "events", None):
        for ev in session.events:
            if curr_inv_id and getattr(ev, "invocation_id", None) != curr_inv_id:
                continue
            if getattr(ev, "long_running_tool_ids", None):
                for int_id in ev.long_running_tool_ids:
                    if str(int_id).startswith("interrupt_ext_"):
                        interrupts.add(str(int_id))
            ev_content = getattr(ev, "content", None)
            if ev_content and getattr(ev_content, "parts", None):
                for p in ev_content.parts:
                    fc = getattr(p, "function_call", None)
                    if fc and getattr(fc, "id", None) and str(fc.id).startswith("interrupt_ext_"):
                        interrupts.add(str(fc.id))
    return interrupts

get_related_items

get_related_items(item_key, tool_context=None)

Retrieve all verified related papers for a given document from the knowledge graph.

Parameters:

Name	Type	Description	Default
item_key	str	Unique Zotero item or attachment key.	required

Returns:

Type	Description
dict[str, Any]	Dictionary with status, item_key, total_count, and list of related items.

Source code in zocrates/agents/common/tools.py

def get_related_items(
    item_key: str,
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Retrieve all verified related papers for a given document from the knowledge graph.

    Args:
        item_key: Unique Zotero item or attachment key.

    Returns:
        Dictionary with status, item_key, total_count, and list of related items.
    """
    allowed, err_msg, notice = _check_and_increment_tool_budget(tool_context)
    if not allowed:
        return {"status": "budget_exhausted", "error": err_msg}

    res = get_related_items_core(item_key=item_key, ceiling=25)

    if notice and tool_context is not None:
        _ensure_preliminary_report(tool_context, force=True)
        res["budget_notice"] = notice

    return res

get_synthesizer_instruction

get_synthesizer_instruction(citations_enabled)

Return the synthesizer instruction tailored to citation configuration.

Source code in zocrates/agents/library_search/search_pipeline.py

def get_synthesizer_instruction(citations_enabled: bool) -> str:
    """Return the synthesizer instruction tailored to citation configuration."""
    if citations_enabled:
        return LIBRARY_SYNTHESIZER_AGENT_INSTRUCTION
    return LIBRARY_SYNTHESIZER_AGENT_INSTRUCTION_STANDARD

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

read_document

read_document(item_key, asset_id=None, page=None, tool_context=None)

Read full document Markdown or targeted sections, figures, or tables from the OKF bundle.

Accepts exact asset IDs from an item's TOC (e.g., 's1', 's2', 'f1', 't1'), 'abstract' (or 'summary') to read the document's abstract, a list of IDs, or fuzzy heading names.

Parameters:

Name	Type	Description	Default
item_key	str	Unique Zotero item key.	required
asset_id	str | list[str] | None	Asset ID from TOC, list of asset IDs, or fuzzy heading string. If None (default), reads whole document paginated.	None
page	int | None	1-indexed page number if content exceeds ~3,000 words.	None

Returns:

Type	Description
dict[str, Any]	Dictionary with 'status', 'item_key', and list of 'assets'.

Source code in zocrates/agents/common/tools.py

def read_document(
    item_key: str,
    asset_id: str | list[str] | None = None,
    page: int | None = None,
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Read full document Markdown or targeted sections, figures, or tables from the OKF bundle.

    Accepts exact asset IDs from an item's TOC (e.g., 's1', 's2', 'f1', 't1'), 'abstract' (or 'summary') to read the document's abstract, a list of IDs, or fuzzy heading names.

    Args:
        item_key: Unique Zotero item key.
        asset_id: Asset ID from TOC, list of asset IDs, or fuzzy heading string.
            If None (default), reads whole document paginated.
        page: 1-indexed page number if content exceeds ~3,000 words.

    Returns:
        Dictionary with 'status', 'item_key', and list of 'assets'.
    """
    allowed, err_msg, notice = _check_and_increment_tool_budget(tool_context)
    if not allowed:
        return {"status": "budget_exhausted", "error": err_msg}

    norm_asset_ids = normalize_asset_ids(asset_id)

    # Intra-tool deduplication check using canonical keys only
    if tool_context is not None:
        seen_assets = set(tool_context.state.get(TEMP_SEEN_ASSETS, []))
        if norm_asset_ids is None:
            full_key = f"{item_key}:full"
            if full_key in seen_assets:
                res = {
                    "status": "success",
                    "item_key": item_key,
                    "assets": [],
                    "notice": f"[Full document {item_key} was already retrieved in this session.]",
                }
                if notice:
                    _ensure_preliminary_report(tool_context, force=True)
                    res["budget_notice"] = notice
                return res
        else:
            unseen = [aid for aid in norm_asset_ids if f"{item_key}:{aid}" not in seen_assets]
            if not unseen:
                res = {
                    "status": "success",
                    "item_key": item_key,
                    "assets": [],
                    "notice": f"[All requested assets for {item_key} ({', '.join(norm_asset_ids)}) were already retrieved in this session.]",
                }
                if notice:
                    _ensure_preliminary_report(tool_context, force=True)
                    res["budget_notice"] = notice
                return res
            norm_asset_ids = unseen

    call_aid: str | list[str] | None = (
        norm_asset_ids[0]
        if (norm_asset_ids is not None and len(norm_asset_ids) == 1)
        else norm_asset_ids
    )
    res = read_document_core(item_key=item_key, asset_id=call_aid, page=page)

    if res.get("status") == "success" and tool_context is not None:
        assets = res.get("assets", [])
        intersecting_ids = res.get("intersecting_chunk_ids", [])

        # Update temp:seen_assets with CANONICAL keys (and single requested alias if resolved)
        seen_assets = set(tool_context.state.get(TEMP_SEEN_ASSETS, []))
        if norm_asset_ids is None:
            seen_assets.add(f"{item_key}:full")
        elif len(norm_asset_ids) == 1 and assets:
            seen_assets.add(f"{item_key}:{norm_asset_ids[0]}")

        for asset in assets:
            resolved_aid = asset.get("asset_id", "full")
            seen_assets.add(f"{item_key}:{resolved_aid}")
        tool_context.state[TEMP_SEEN_ASSETS] = list(seen_assets)

        # Range-Aware Pre-Exclusion: Add intersecting chunk IDs to temp:seen_chunk_ids
        seen_chunk_ids = set(tool_context.state.get(TEMP_SEEN_CHUNK_IDS, []))
        seen_chunk_ids.update(intersecting_ids)
        tool_context.state[TEMP_SEEN_CHUNK_IDS] = list(seen_chunk_ids)

        # Convert read assets to synthetic evidence chunks in temp:retrieved_evidence (deduping by chunk_id)
        doc_title = res.get("title") or ""
        doc_authors = res.get("authors")
        doc_year = res.get("year")
        if tool_context is not None and (not doc_title or doc_authors is None or doc_year is None):
            active_items = tool_context.state.get(ACTIVE_CONTEXT_ITEMS, [])
            if isinstance(active_items, list):
                for item in active_items:
                    if isinstance(item, dict) and (item.get("item_key") or item.get("key")) == item_key:
                        if not doc_title:
                            doc_title = item.get("title", "")
                        if doc_authors is None:
                            doc_authors = item.get("authors") or item.get("creators")
                        if doc_year is None:
                            doc_year = item.get("year")
                        break

        # Canonical bundle key: read_document_core resolves scope aliases (e.g.
        # a parent item key whose PDF bundle lives under an attachment key) to
        # the key the bundle is actually stored under.  Evidence chunks must
        # carry the RESOLVED key — the /assets endpoint serves files from
        # cache_dir/<item_key>, so a requested-but-unresolved key would 404
        # the figure image in the sidebar.
        resolved_bundle_key = str(res.get("item_key") or item_key).strip() or item_key

        current_evidence = list(tool_context.state.get(TEMP_RETRIEVED_EVIDENCE, []))
        existing_evidence_ids = {
            e.get("chunk_id") for e in current_evidence if e.get("chunk_id")
        }
        for asset in assets:
            synthetic_chunk_id = f"{resolved_bundle_key}:{asset.get('asset_id', 'full')}"
            if synthetic_chunk_id in existing_evidence_ids:
                continue
            existing_evidence_ids.add(synthetic_chunk_id)
            chunk_dict: dict[str, Any] = {
                "chunk_id": synthetic_chunk_id,
                "item_key": resolved_bundle_key,
                "title": doc_title,
                "chunk_type": asset.get("asset_type", "text"),
                "heading": asset.get("heading", ""),
                "text": asset.get("content", ""),
                "word_count": asset.get("word_count", 0),
                "retrieval_method": "direct_read",
                "cosine_similarity": None,
                "rrf_score": None,
                "rerank_score": None,
                "image_path": asset.get("path") if asset.get("asset_type") == "figure" else None,
                "structured_content": asset.get("structured_content"),
            }
            if doc_authors is not None:
                chunk_dict["authors"] = doc_authors
            if doc_year is not None:
                chunk_dict["year"] = doc_year
            current_evidence.append(chunk_dict)
        tool_context.state[TEMP_RETRIEVED_EVIDENCE] = current_evidence

        total_words = sum(c.get("word_count", 0) for c in current_evidence)
        res["retrieval_status"] = (
            f"[Retrieval status: ~{total_words} words accumulated in evidence set; "
            f"excluded {len(intersecting_ids)} overlapping vector chunks from future searches]"
        )

    if notice and tool_context is not None:
        _ensure_preliminary_report(tool_context, force=True)
        res["budget_notice"] = notice

    return res

recover_isolation_scope

recover_isolation_scope(ctx, prefixes)

Recover the search pipeline's isolation scope from snapshot or session events if unset.

Parameters:

Name	Type	Description	Default
ctx	Context	Workflow execution context.	required
prefixes	tuple[str, ...] | list[str] | str	Required tuple, list, or string of acceptable isolation scope prefix(es) (e.g. ("search_library_pipeline:", "search_pipeline:") or ("search_online_pipeline:",)).	required

Source code in zocrates/agents/common/pipeline_base.py

def recover_isolation_scope(
    ctx: Context,
    prefixes: tuple[str, ...] | list[str] | str,
) -> str | None:
    """Recover the search pipeline's isolation scope from snapshot or session events if unset.

    Args:
        ctx: Workflow execution context.
        prefixes: Required tuple, list, or string of acceptable isolation scope prefix(es)
            (e.g. `("search_library_pipeline:", "search_pipeline:")` or `("search_online_pipeline:",)`).
    """
    if getattr(ctx, "isolation_scope", None):
        return ctx.isolation_scope

    state = getattr(ctx, "state", None)
    if state:
        snapshot = state.get(HITL_SNAPSHOT)
        if isinstance(snapshot, dict) and snapshot.get("isolation_scope"):
            return snapshot["isolation_scope"]

    if isinstance(prefixes, str):
        prefixes_tuple = (prefixes,)
    else:
        prefixes_tuple = tuple(prefixes)

    # Fall back to scanning session events in reverse
    curr_inv_id = getattr(ctx, "invocation_id", None) or (
        getattr(ctx._invocation_context, "invocation_id", None)
        if hasattr(ctx, "_invocation_context")
        else None
    )
    session = getattr(ctx, "session", None)
    if not session and hasattr(ctx, "_invocation_context"):
        session = getattr(ctx._invocation_context, "session", None)
    if session and getattr(session, "events", None):
        if curr_inv_id:
            for event in reversed(session.events):
                if getattr(event, "invocation_id", None) == curr_inv_id:
                    scope = getattr(event, "isolation_scope", None)
                    if scope and str(scope).startswith(prefixes_tuple):
                        return str(scope)
        for event in reversed(session.events):
            scope = getattr(event, "isolation_scope", None)
            if scope and str(scope).startswith(prefixes_tuple):
                return str(scope)
    return None

request_search_extension

request_search_extension(justification, covered_summary, missing_aspects, tool_context=None)

Request permission from the user to extend the retrieval budget beyond the current limit.

Use this tool when evidence volume or retrieval tool budget is approaching its ceiling and crucial literature aspects or candidate papers remain unaddressed.

Parameters:

Name	Type	Description	Default
justification	str	Clear reason why additional retrieval calls or evidence capacity are needed.	required
covered_summary	str	Brief summary of the findings and evidence collected so far.	required
missing_aspects	list[str]	List of specific unaddressed aspects, papers, or topics still needed.	required

Returns:

Type	Description
dict[str, Any]	Dictionary confirming extension request registration and turn conclusion.

Source code in zocrates/agents/common/tools.py

def request_search_extension(
    justification: str,
    covered_summary: str,
    missing_aspects: list[str],
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Request permission from the user to extend the retrieval budget beyond the current limit.

    Use this tool when evidence volume or retrieval tool budget is approaching its ceiling
    and crucial literature aspects or candidate papers remain unaddressed.

    Args:
        justification: Clear reason why additional retrieval calls or evidence capacity are needed.
        covered_summary: Brief summary of the findings and evidence collected so far.
        missing_aspects: List of specific unaddressed aspects, papers, or topics still needed.

    Returns:
        Dictionary confirming extension request registration and turn conclusion.
    """
    if tool_context is not None:
        tool_context.state[TEMP_SEARCH_REPORT] = {
            "summary": covered_summary,
            "gaps": missing_aspects if isinstance(missing_aspects, list) else [str(missing_aspects)],
            "notes": justification,
            "source": "subagent_request",
        }
        tool_context.state[TEMP_EXTENSION_REQUESTED] = True
        ext_num = int(tool_context.state.get(TEMP_EXTENSION_COUNT, 0)) + 1
        tool_context.state[TEMP_EXTENSION_REQUEST_PASS] = ext_num
        tool_context.actions.skip_summarization = True

    return {
        "status": "extension_requested",
        "message": "Extension request registered. Concluding retrieval turn for user approval.",
    }

rerank_evidence async

rerank_evidence(query, chunks, max_chunks=None, word_ceiling=None)

Rerank candidate chunks using OpenRouter cross-encoder with budget trimming and fallback.

Source code in zocrates/agents/common/reranker.py

async def rerank_evidence(
    query: str,
    chunks: list[dict[str, Any]],
    max_chunks: int | None = None,
    word_ceiling: int | None = None,
) -> list[dict[str, Any]]:
    """Rerank candidate chunks using OpenRouter cross-encoder with budget trimming and fallback."""
    chunks = filter_substantive_evidence(chunks)

    effective_max_chunks = max_chunks if max_chunks is not None else MAX_RERANK_CHUNKS
    effective_word_ceiling = word_ceiling if word_ceiling is not None else HARD_WORD_CEILING

    if not chunks:
        return []

    if len(chunks) == 1:
        single = dict(chunks[0])
        single["rerank_score"] = None
        return _trim_to_budget([single], effective_word_ceiling)

    api_key = get_secure_api_key()

    # Fail-open enforcement (Phase 7 onboarding, plan §5.1/§8): a missing or
    # invalid reranker model config must degrade to un-reranked retrieval with
    # a logged warning — never fail the turn or the search run.
    try:
        reranker_model_raw = load_zocrates_config().get("reranker_model")
    except Exception as cfg_exc:
        logger.warning(
            "Failed to read reranker_model from config (%s). Skipping cross-encoder rerank.",
            cfg_exc,
        )
        trimmed = [dict(c, rerank_score=None) for c in chunks][:effective_max_chunks]
        return _trim_to_budget(trimmed, effective_word_ceiling)

    if not isinstance(reranker_model_raw, str) or not reranker_model_raw.strip():
        logger.warning(
            "No reranker model configured (reranker_model unset or invalid). "
            "Skipping cross-encoder rerank; retrieval proceeds un-reranked."
        )
        trimmed = [dict(c, rerank_score=None) for c in chunks][:effective_max_chunks]
        return _trim_to_budget(trimmed, effective_word_ceiling)

    reranker_model = reranker_model_raw.strip()

    if not api_key:
        logger.debug(
            "No OpenRouter API key found. Skipping cross-encoder rerank."
        )
        trimmed = [dict(c, rerank_score=None) for c in chunks][:effective_max_chunks]
        return _trim_to_budget(trimmed, effective_word_ceiling)

    documents = [c.get("text", "") for c in chunks]
    payload = {
        "model": reranker_model,
        "query": query,
        "documents": documents,
        "top_n": len(documents),
    }

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.post(
                "https://openrouter.ai/api/v1/rerank",
                headers={
                    "Authorization": f"Bearer {api_key}",
                    "Content-Type": "application/json",
                },
                json=payload,
            )
            resp.raise_for_status()
            data = resp.json()

        results = data.get("results", [])
        score_by_idx = {
            r["index"]: r.get("relevance_score", 0.0) for r in results
        }

        ranked_chunks = []
        for idx, chunk in enumerate(chunks):
            chunk_copy = dict(chunk)
            chunk_copy["rerank_score"] = float(score_by_idx.get(idx, 0.0))
            ranked_chunks.append(chunk_copy)

        ranked_chunks.sort(
            key=lambda c: c.get("rerank_score", 0.0), reverse=True
        )
        top_candidates = ranked_chunks[:effective_max_chunks]
        return _trim_to_budget(top_candidates, effective_word_ceiling)

    except Exception as e:
        logger.warning(
            "OpenRouter rerank call failed (%s: %s). Falling back to original chunk order.",
            type(e).__name__,
            e,
        )
        trimmed = [dict(c, rerank_score=None) for c in chunks][:effective_max_chunks]
        return _trim_to_budget(trimmed, effective_word_ceiling)

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

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

search_budget_checkpoint async

search_budget_checkpoint(ctx, node_input=None)

Dynamic workflow checkpoint node yielding RequestInput to pause for human extension.

Source code in zocrates/agents/library_search/search_pipeline.py

@node(name="search_budget_checkpoint", rerun_on_resume=False)
async def search_budget_checkpoint(
    ctx: Context,
    node_input: dict[str, Any] | None = None,
) -> Any:
    """Dynamic workflow checkpoint node yielding RequestInput to pause for human extension."""
    payload = node_input or {}
    ext_num = payload.get("extension_number", 1)
    int_id = f"interrupt_ext_{ext_num}"
    yield RequestInput(
        interrupt_id=int_id,
        message="Deep research limit reached. Extend retrieval or synthesize current findings?",
        payload=payload,
    )

search_corpus

search_corpus(query, item_key='active', chunk_type=None, limit=10, tool_context=None)

Hybrid (dense vector + FTS5 BM25) search over indexed library chunks with compound pre-filtering.

Parameters:

Name	Type	Description	Default
query	str	The search query string. If empty, returns immediately.	required
item_key	str | list[str] | None	Scope of the search: - "active" (default): searches active context items selected in Zotero (or entire library if none selected). - None, "all", or "*": searches across the ENTIRE library (unfiltered). - Specific item key(s) (e.g. "HCZLU22L" or ["HCZLU22L", "M5Z7SFWM"]): filters strictly to those items.	'active'
chunk_type	str | list[str] | None	Chunk type filter ('frontmatter', 'figure', 'table', 'equation', 'code', 'text'). Omit, None, or "all" to search across all chunk types.	None
limit	int	Maximum number of chunks to return (default 10).	10

Returns:

Type	Description
dict[str, Any]	Dictionary with 'status' and list of matching 'results'.

Source code in zocrates/agents/common/tools.py

def search_corpus(
    query: str,
    item_key: str | list[str] | None = "active",
    chunk_type: str | list[str] | None = None,
    limit: int = 10,
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Hybrid (dense vector + FTS5 BM25) search over indexed library chunks with compound pre-filtering.

    Args:
        query: The search query string. If empty, returns immediately.
        item_key: Scope of the search:
            - "active" (default): searches active context items selected in Zotero (or entire library if none selected).
            - None, "all", or "*": searches across the ENTIRE library (unfiltered).
            - Specific item key(s) (e.g. "HCZLU22L" or ["HCZLU22L", "M5Z7SFWM"]): filters strictly to those items.
        chunk_type: Chunk type filter ('frontmatter', 'figure', 'table', 'equation', 'code', 'text').
            Omit, None, or "all" to search across all chunk types.
        limit: Maximum number of chunks to return (default 10).

    Returns:
        Dictionary with 'status' and list of matching 'results'.
    """
    if not query or not query.strip():
        return {
            "status": "error",
            "message": "Query string cannot be empty.",
            "results": [],
        }

    allowed, err_msg, notice = _check_and_increment_tool_budget(tool_context)
    if not allowed:
        return {"status": "budget_exhausted", "error": err_msg}

    filter_keys = item_key
    seen_chunk_ids: list[str] | None = None

    if tool_context is not None:
        # Resolve active selection scope only when explicitly requested via "active", "selected", or "context"
        if isinstance(filter_keys, str) and filter_keys.strip().lower() in ("active", "selected", "context"):
            active_items = tool_context.state.get(ACTIVE_CONTEXT_ITEMS)
            if active_items and isinstance(active_items, list):
                extracted = [
                    item.get("item_key") or item.get("key")
                    for item in active_items
                    if isinstance(item, dict) and (item.get("item_key") or item.get("key"))
                ]
                filter_keys = extracted if extracted else None
            else:
                filter_keys = None

        raw_seen = tool_context.state.get(TEMP_SEEN_CHUNK_IDS, [])
        if isinstance(raw_seen, (list, set)):
            seen_chunk_ids = list(raw_seen)

    res = search_corpus_core(
        query=query,
        item_key=filter_keys,
        chunk_type=chunk_type,
        limit=limit,
        exclude_chunk_ids=seen_chunk_ids,
    )

    if res.get("status") == "success" and tool_context is not None:
        results = res.get("results", [])
        new_ids = [r["chunk_id"] for r in results if "chunk_id" in r]

        # Update temp:seen_chunk_ids
        current_seen = list(tool_context.state.get(TEMP_SEEN_CHUNK_IDS, []))
        seen_set = set(current_seen)
        seen_set.update(new_ids)
        tool_context.state[TEMP_SEEN_CHUNK_IDS] = list(seen_set)

        # Update temp:retrieved_evidence
        current_evidence = list(tool_context.state.get(TEMP_RETRIEVED_EVIDENCE, []))
        for r in results:
            chunk_entry: dict[str, Any] = {
                "chunk_id": r["chunk_id"],
                "item_key": r["item_key"],
                "title": r.get("title", ""),
                "chunk_type": r.get("chunk_type", "text"),
                "heading": r.get("heading", ""),
                "text": r.get("text", ""),
                "word_count": r.get("word_count", 0),
                "retrieval_method": "hybrid",
                "cosine_similarity": r.get("cosine_similarity"),
                "rrf_score": r.get("rrf_score"),
                "rerank_score": None,
                "image_path": r.get("image_path"),
                "structured_content": None,
            }
            if "authors" in r:
                chunk_entry["authors"] = r["authors"]
            if "year" in r:
                chunk_entry["year"] = r["year"]
            current_evidence.append(chunk_entry)
        tool_context.state[TEMP_RETRIEVED_EVIDENCE] = current_evidence

        total_words = sum(c.get("word_count", 0) for c in current_evidence)
        res["retrieval_status"] = (
            f"[Retrieval status: {len(new_ids)} new chunks retrieved, "
            f"~{total_words} words accumulated in evidence set]"
        )

    if notice and tool_context is not None:
        _ensure_preliminary_report(tool_context, force=True)
        res["budget_notice"] = notice

    return res

search_library async

search_library(request, tool_context)

Autonomous search tool executing the local library search and synthesis pipeline.

Use this tool when the user asks questions about papers, notes, or topics contained in their existing Zotero library.

Source code in zocrates/agents/library_search/search_pipeline.py

async def search_library(request: str, tool_context: ToolContext) -> str:
    """Autonomous search tool executing the local library search and synthesis pipeline.

    Use this tool when the user asks questions about papers, notes, or topics
    contained in their existing Zotero library.
    """
    tool_context.actions.skip_summarization = True
    fc_id = (
        getattr(tool_context, "function_call_id", None)
        or getattr(tool_context, "invocation_id", None)
        or getattr(getattr(tool_context, "_invocation_context", None), "invocation_id", None)
        or "default"
    )
    result = await tool_context.run_node(
        search_library_pipeline_node,
        node_input=request,
        override_isolation_scope=f"search_library_pipeline:{fc_id}",
    )
    return coerce_to_text(result)

search_library_pipeline_node async

search_library_pipeline_node(ctx, node_input='')

Dynamic workflow function orchestrating autonomous search, reranking, and synthesis.

Wraps _run_search_pipeline in a try/finally so the node always closes its isolation scope (search_library_pipeline:{fc_id}) before returning, whether it succeeds ("Task completed.") or raises ("Task failed."). Closing the scope prevents ADK's Runner from treating the finished pipeline as an unclosed in-flight task agent and hijacking subsequent user turns.

Source code in zocrates/agents/library_search/search_pipeline.py

@node(name="search_library_pipeline_node", rerun_on_resume=True)
async def search_library_pipeline_node(
    ctx: Context,
    node_input: str = "",
) -> str:
    """Dynamic workflow function orchestrating autonomous search, reranking, and synthesis.

    Wraps ``_run_search_pipeline`` in a try/finally so the node always closes
    its isolation scope (``search_library_pipeline:{fc_id}``) before returning, whether
    it succeeds ("Task completed.") or raises ("Task failed."). Closing the
    scope prevents ADK's Runner from treating the finished pipeline as an
    unclosed in-flight task agent and hijacking subsequent user turns.
    """
    _ensure_isolation_scope(ctx)

    completed = False
    interrupted = False
    try:
        response_text = await _run_search_pipeline(ctx, node_input)
        completed = True
        return response_text
    except NodeInterruptedError:
        interrupted = True
        raise
    finally:
        if not interrupted:
            await _enqueue_isolation_scope_closing_event(
                ctx,
                result="Task completed." if completed else "Task failed.",
            )

strip_snippet_keys

strip_snippet_keys(snippet_xml)

Programmatically strip item key attributes from snippet XML tags.

Leaves all other XML attributes (e.g. paper, page, authors) and snippet text content intact.

Parameters:

Name	Type	Description	Default
snippet_xml	str	XML-formatted snippet string.	required

Returns:

Type	Description
str	Snippet XML string with all ' key="..."' attributes removed.

Source code in zocrates/agents/common/prompts.py

def strip_snippet_keys(snippet_xml: str) -> str:
    """Programmatically strip item key attributes from snippet XML tags.

    Leaves all other XML attributes (e.g. paper, page, authors)
    and snippet text content intact.

    Args:
        snippet_xml: XML-formatted snippet string.

    Returns:
        Snippet XML string with all ' key="..."' attributes removed.
    """
    if not snippet_xml:
        return ""
    return re.sub(r'\s+key="[^"]*"', "", snippet_xml)