search_pipeline

zocrates.agents.online_search.search_pipeline

Dynamic online literature discovery pipeline: search loop and synthesis workflow node.

CANDIDATE_PAPERS module-attribute

CANDIDATE_PAPERS = 'candidates'

HITL_SNAPSHOT module-attribute

HITL_SNAPSHOT = 'hitl_snapshot'

INITIAL_ONLINE_TOOL_BUDGET module-attribute

INITIAL_ONLINE_TOOL_BUDGET = 8

MAX_ONLINE_EXTENSIONS module-attribute

MAX_ONLINE_EXTENSIONS = 2

ONLINE_BUDGET_EXTENSION_INCREMENT module-attribute

ONLINE_BUDGET_EXTENSION_INCREMENT = 5

RESEARCH_TURN_HISTORY module-attribute

RESEARCH_TURN_HISTORY = 'research_turn_history'

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_ONLINE_CANDIDATES module-attribute

TEMP_ONLINE_CANDIDATES = 'temp_online_candidates'

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_SEARCH_QUERY module-attribute

TEMP_SEARCH_QUERY = 'temp:search_query'

TEMP_SEARCH_REPORT module-attribute

TEMP_SEARCH_REPORT = 'temp:search_report'

USER_PROMPT module-attribute

USER_PROMPT = 'user_prompt'

USER_RESEARCH_PROFILE_STR module-attribute

USER_RESEARCH_PROFILE_STR = 'user_research_profile_str'

__all__ module-attribute

__all__ = ['build_online_coordinator_view', 'format_candidates_for_synthesizer', 'online_search_budget_checkpoint', 'search_online_literature', 'search_online_pipeline_node']

logger module-attribute

logger = logging.getLogger(__name__)

CandidatePaper dataclass

CandidatePaper(candidate_id, title, authors=list(), year=None, venue=None, doi=None, arxiv_id=None, openalex_id='', abstract=None, tldr=None, is_oa=False, oa_pdf_url=None, oa_pdf_urls=list(), citation_count=0, primary_topic=None, keywords=list(), in_library=False)

Represents a discovered scholarly paper candidate.

from_dict classmethod

from_dict(data)

Deserialize CandidatePaper from a dictionary.

Source code in zocrates/discovery/models.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "CandidatePaper":
    """Deserialize CandidatePaper from a dictionary."""
    citation_count_val = data.get("citation_count", 0)
    try:
        citation_count = int(citation_count_val) if citation_count_val is not None else 0
    except (ValueError, TypeError):
        citation_count = 0

    year_val = data.get("year")
    year: int | None = None
    if year_val is not None:
        try:
            year = int(year_val)
        except (ValueError, TypeError):
            year = None

    raw_authors = data.get("authors")
    if isinstance(raw_authors, list):
        authors = [str(a).strip() for a in raw_authors if a and str(a).strip()]
    elif isinstance(raw_authors, str) and raw_authors.strip():
        authors = [raw_authors.strip()]
    else:
        authors = []

    raw_keywords = data.get("keywords")
    if isinstance(raw_keywords, list):
        keywords = [str(k).strip() for k in raw_keywords if k and str(k).strip()]
    elif isinstance(raw_keywords, str) and raw_keywords.strip():
        keywords = [raw_keywords.strip()]
    else:
        keywords = []

    def _clean_opt_str(val: Any) -> str | None:
        """Clean and trim an optional string or scalar value.

        Args:
            val: Input value to convert to a stripped string.

        Returns:
            Stripped string, or None if empty or invalid.
        """
        if val is not None and isinstance(val, (str, int, float)):
            s = str(val).strip()
            return s if s else None
        return None

    raw_urls = data.get("oa_pdf_urls")
    oa_pdf_urls: list[str] = []
    if isinstance(raw_urls, list):
        oa_pdf_urls = [str(u).strip() for u in raw_urls if u and str(u).strip()]
    primary_pdf = _clean_opt_str(data.get("oa_pdf_url"))
    if primary_pdf and primary_pdf not in oa_pdf_urls:
        oa_pdf_urls.insert(0, primary_pdf)

    return cls(
        candidate_id=str(data.get("candidate_id") or ""),
        title=str(data.get("title") or "").strip(),
        authors=authors,
        year=year,
        venue=_clean_opt_str(data.get("venue")),
        doi=_clean_opt_str(data.get("doi")),
        arxiv_id=_clean_opt_str(data.get("arxiv_id")),
        openalex_id=str(data.get("openalex_id") or ""),
        abstract=_clean_opt_str(data.get("abstract")),
        tldr=_clean_opt_str(data.get("tldr")),
        is_oa=bool(data.get("is_oa", False)),
        oa_pdf_url=primary_pdf,
        oa_pdf_urls=oa_pdf_urls,
        citation_count=citation_count,
        primary_topic=_clean_opt_str(data.get("primary_topic")),
        keywords=keywords,
        in_library=bool(data.get("in_library", False)),
    )

to_dict

to_dict()

Serialize CandidatePaper to a plain dictionary.

Source code in zocrates/discovery/models.py

def to_dict(self) -> dict[str, Any]:
    """Serialize CandidatePaper to a plain dictionary."""
    return {
        "candidate_id": self.candidate_id,
        "title": self.title,
        "authors": list(self.authors),
        "year": self.year,
        "venue": self.venue,
        "doi": self.doi,
        "arxiv_id": self.arxiv_id,
        "openalex_id": self.openalex_id,
        "abstract": self.abstract,
        "tldr": self.tldr,
        "is_oa": self.is_oa,
        "oa_pdf_url": self.oa_pdf_url,
        "oa_pdf_urls": list(self.oa_pdf_urls),
        "citation_count": self.citation_count,
        "primary_topic": self.primary_topic,
        "keywords": list(self.keywords),
        "in_library": self.in_library,
    }

_clean_user_facing_synthesis

_clean_user_facing_synthesis(text)

Post-process online synthesizer output to deterministically strip thinking tags and bibliography headings.

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

def _clean_user_facing_synthesis(text: str) -> str:
    """Post-process online synthesizer output to deterministically strip thinking tags and bibliography headings."""
    if not text:
        return text

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

    # Strip hallucinated trailing references / candidate paper headings
    cleaned = re.sub(
        r"(?:\n{1,3}|^\s*)#{1,3}\s*(?:References|Cited Sources|Sources Consulted|Recommended Papers for Import|Discovered Candidate Papers)[\s\S]*$",
        "",
        text,
        flags=re.IGNORECASE,
    ).rstrip()

    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.

Source code in zocrates/agents/online_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."""
    await enqueue_isolation_scope_closing_event(
        ctx,
        result=result,
        default_author="search_online_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/online_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_online_pipeline:",),
    )

_format_candidate_authors

_format_candidate_authors(raw_authors)

Format a candidate paper's authors into a clean academic citation string.

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

def _format_candidate_authors(raw_authors: Any) -> str:
    """Format a candidate paper's authors into a clean academic citation string."""
    if isinstance(raw_authors, list):
        cleaned = [str(a).strip() for a in raw_authors if a and str(a).strip()]
    elif isinstance(raw_authors, str) and raw_authors.strip():
        cleaned = [raw_authors.strip()]
    else:
        cleaned = []

    if not cleaned:
        return "Unknown Authors"
    if len(cleaned) == 1:
        return cleaned[0]
    if len(cleaned) == 2:
        return f"{cleaned[0]} & {cleaned[1]}"
    if len(cleaned) == 3:
        return f"{cleaned[0]}, {cleaned[1]}, & {cleaned[2]}"
    return f"{cleaned[0]} et al."

_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/online_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)

_reconcile_state_after_subagent

_reconcile_state_after_subagent(state)

Reconcile uncommitted parent State._delta with child subagent updates in State._value.

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

def _reconcile_state_after_subagent(state: Any) -> None:
    """Reconcile uncommitted parent State._delta with child subagent updates in State._value."""
    delta = getattr(state, "_delta", None)
    value = getattr(state, "_value", None)
    if not isinstance(delta, dict) or not isinstance(value, dict):
        return
    for key in list(delta.keys()):
        if key in value and delta[key] is not value[key] and delta[key] != value[key]:
            delta.pop(key, None)

_recover_isolation_scope

_recover_isolation_scope(ctx)

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

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

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

_run_online_search_pipeline async

_run_online_search_pipeline(ctx, node_input)

Core online search pipeline: subagent execution, HITL extensions, and synthesis.

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

async def _run_online_search_pipeline(ctx: Context, node_input: str) -> str:
    """Core online search pipeline: subagent execution, HITL extensions, 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, "")
        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))
    initial_online_budget = int(
        h_cfg.get("online_initial_retrieval_tools", INITIAL_ONLINE_TOOL_BUDGET)
    )
    max_extensions = int(h_cfg.get("max_online_extensions", MAX_ONLINE_EXTENSIONS))
    budget_increment = int(h_cfg.get("online_budget_increment", ONLINE_BUDGET_EXTENSION_INCREMENT))

    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 not snapshot:
        state[TEMP_RETRIEVAL_TOOL_BUDGET] = initial_online_budget
        state["temp_retrieval_tool_budget"] = initial_online_budget
        state[TEMP_RETRIEVAL_TOOL_COUNT] = 0
        state["temp_retrieval_tool_count"] = 0
        state[TEMP_EXTENSION_COUNT] = 0
        state["temp_extension_count"] = 0
        state[TEMP_ONLINE_CANDIDATES] = {}
        state["temp_online_candidates"] = {}
        state["selected_candidates"] = []
        state["candidates"] = []
        state["candidates_invocation_id"] = None
        state[CANDIDATE_PAPERS] = []
        state[TEMP_SEARCH_REPORT] = None
        state["temp_online_search_report"] = None
        state[TEMP_EXTENSION_REQUESTED] = False
        state["temp_extension_requested"] = False
        state[TEMP_EXTENSION_REQUEST_PASS] = None
        state["extension_rationale"] = None
        raw_engine = state.get("discovery_engine")
        if not (hasattr(raw_engine, "search") and callable(getattr(raw_engine, "search"))):
            if hasattr(state, "pop"):
                state.pop("discovery_engine", None)
            elif raw_engine is not None:
                state["discovery_engine"] = None
    else:
        ceilings = snapshot.get("ceilings", {})
        if state.get(TEMP_RETRIEVAL_TOOL_BUDGET) is None:
            state[TEMP_RETRIEVAL_TOOL_BUDGET] = ceilings.get("budget", INITIAL_ONLINE_TOOL_BUDGET)
        if not state.get(TEMP_ONLINE_CANDIDATES) and snapshot.get("candidates_pool"):
            state[TEMP_ONLINE_CANDIDATES] = snapshot["candidates_pool"]
            state["temp_online_candidates"] = snapshot["candidates_pool"]
        if not state.get("selected_candidates") and snapshot.get("selected_candidates"):
            restored = [
                CandidatePaper.from_dict(c) if isinstance(c, dict) else c
                for c in snapshot["selected_candidates"]
            ]
            state["selected_candidates"] = restored
            state[CANDIDATE_PAPERS] = [
                c.to_dict() if hasattr(c, "to_dict") else c for c in restored
            ]

    if state.get(TEMP_RETRIEVAL_TOOL_BUDGET) is None:
        state[TEMP_RETRIEVAL_TOOL_BUDGET] = initial_online_budget
    if state.get(TEMP_EXTENSION_COUNT) is None:
        state[TEMP_EXTENSION_COUNT] = 0

    # Clear invocation user_content so subagents receive only dedicated input
    if hasattr(ctx, "_invocation_context") and ctx._invocation_context is not None:
        ctx._invocation_context.user_content = None

    # Subagent resolution
    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("online_search")
    synth_agent = subagents.get("online_synthesizer")

    if (search_agent is None or synth_agent is None) and getattr(parent_agent, "name", None) in (
        "search_online_pipeline_node",
    ):
        try:
            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("online_search")
                if synth_agent is None:
                    synth_agent = coord_subagents.get("online_synthesizer")
        except Exception:
            pass

    if search_agent is None or synth_agent is None:
        from zocrates.llm import get_llm_model

        default_model = get_llm_model()
        if search_agent is None:
            search_agent = create_online_search_agent(default_model)
        if synth_agent is None:
            synth_agent = create_online_synthesizer_agent(default_model)

    search_node_input = f"# Target Literature Discovery Goal\n{search_query}"
    active_isolation_scope = _ensure_isolation_scope(ctx)

    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,
        )
        _reconcile_state_after_subagent(state)

    # Extension loop
    extension_count = 0
    if snapshot and snapshot.get("extension_count") is not None:
        extension_count = int(snapshot["extension_count"])

    while True:
        pool = (
            state.get(TEMP_ONLINE_CANDIDATES)
            if state.get(TEMP_ONLINE_CANDIDATES) is not None
            else state.get("temp_online_candidates")
        ) or (snapshot.get("candidates_pool") if snapshot else {}) or {}

        search_report = state.get(TEMP_SEARCH_REPORT) or (snapshot.get("search_report") if snapshot else None)
        if search_report is None:
            search_report = {
                "summary": f"Discovered {len(pool)} candidate papers." if pool else "No candidate papers discovered.",
                "gaps": [] if pool else ["Need to formulate alternative queries to discover relevant papers."],
                "notes": "",
                "source": "auto_generated",
            }
            state[TEMP_SEARCH_REPORT] = search_report

        current_calls = int(
            state.get(TEMP_RETRIEVAL_TOOL_COUNT)
            if state.get(TEMP_RETRIEVAL_TOOL_COUNT) is not None
            else 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)
            if state.get(TEMP_RETRIEVAL_TOOL_BUDGET) is not None
            else state.get("temp_retrieval_tool_budget") or INITIAL_ONLINE_TOOL_BUDGET
        )

        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(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))
            )
        )

        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"
        else:
            trigger_reason = "unknown"

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

        # Save snapshot
        serialized_pool = {
            k: (v.to_dict() if hasattr(v, "to_dict") else v)
            for k, v in pool.items()
        }
        state[HITL_SNAPSHOT] = {
            "search_query": search_query,
            "candidates_pool": serialized_pool,
            "selected_candidates": [
                c.to_dict() if hasattr(c, "to_dict") else c
                for c in state.get("selected_candidates") or []
            ],
            "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,
            },
            "pipeline_type": "online_search",
            "trigger_reason": trigger_reason,
            "trigger_details": {
                "tool_calls_used": current_calls,
                "tool_budget": current_budget,
            },
            "agent_justification": agent_justification,
        }

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

        decision = await ctx.run_node(
            online_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":
            break

        new_budget = current_budget + budget_increment
        new_extension_count = extension_count + 1

        state[TEMP_RETRIEVAL_TOOL_BUDGET] = new_budget
        state["temp_retrieval_tool_budget"] = new_budget
        state[TEMP_EXTENSION_COUNT] = new_extension_count
        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 discovery"
        continuation_prompt = (
            f"# Deep Research Extension Granted (Extension {new_extension_count} of {max_extensions})\n"
            "The user granted additional retrieval tool calls for online discovery.\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"online_search_extension_{new_extension_count}",
                override_isolation_scope=active_isolation_scope,
            )
            _reconcile_state_after_subagent(state)
            completed_extensions.append(new_extension_count)

        latest_pool = (
            state.get(TEMP_ONLINE_CANDIDATES)
            or state.get("temp_online_candidates")
            or pool
            or {}
        )
        state[HITL_SNAPSHOT] = {
            "search_query": search_query,
            "candidates_pool": {
                k: (v.to_dict() if hasattr(v, "to_dict") else v)
                for k, v in latest_pool.items()
            },
            "selected_candidates": [
                c.to_dict() if hasattr(c, "to_dict") else c
                for c in state.get("selected_candidates") or []
            ],
            "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 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,
            },
            "pipeline_type": "online_search",
            "trigger_reason": trigger_reason,
            "trigger_details": {
                "tool_calls_used": current_calls,
                "tool_budget": current_budget,
            },
            "agent_justification": agent_justification,
        }
        extension_count = new_extension_count

    # Determine candidate papers to present to synthesizer
    raw_selected = (
        state.get("selected_candidates")
        or (snapshot.get("selected_candidates") if snapshot else [])
        or []
    )
    synth_candidates: list[CandidatePaper] = [
        CandidatePaper.from_dict(c) if isinstance(c, dict) else c
        for c in raw_selected
    ]

    if not synth_candidates and pool:
        for cand in pool.values():
            c = CandidatePaper.from_dict(cand) if isinstance(cand, dict) else cand
            synth_candidates.append(c)
        synth_candidates = synth_candidates[:10]

    formatted_candidates = format_candidates_for_synthesizer(synth_candidates)

    user_query = state.get(USER_PROMPT)
    if not user_query:
        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": "online_search",
        "user_query": user_query,
        "search_goal": search_query,
        "conclude_output": {
            "summary": search_report.get("summary", ""),
            "gaps": search_report.get("gaps", []),
            "selected_candidates": [
                {"title": getattr(c, "title", ""), "tldr": getattr(c, "tldr", "")}
                for c in synth_candidates
                if getattr(c, "title", None)
            ],
        },
    }
    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:]

    profile_str = str(state.get(USER_RESEARCH_PROFILE_STR) or "").strip()
    synth_prompt_parts = []
    if profile_str:
        synth_prompt_parts.append(profile_str)
    synth_prompt_parts.extend([
        f"# User Research Inquiry\n{user_query}",
        f"# Target Search Goal\n{search_query}",
        f"# Discovered Candidate Papers\n{formatted_candidates}",
    ])
    synth_prompt = "\n\n".join(synth_prompt_parts)

    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)

    # Programmatic Candidate Stamping
    raw_candidates = (
        state.get("selected_candidates")
        or (snapshot.get("selected_candidates") if snapshot else [])
        or []
    )
    if not raw_candidates and pool:
        raw_candidates = list(pool.values())[:10]

    candidate_dicts = [
        c.to_dict() if hasattr(c, "to_dict") else c
        for c in raw_candidates
    ]

    state["candidates"] = candidate_dicts
    state["selected_candidates"] = candidate_dicts
    state[CANDIDATE_PAPERS] = candidate_dicts
    state["candidates_invocation_id"] = curr_inv_id
    meta = dict(state.get("metadata") or {})
    meta["candidates"] = candidate_dicts
    state["metadata"] = meta

    # Turn cleanup
    state[HITL_SNAPSHOT] = None
    state[TEMP_EXTENSION_REQUESTED] = False
    state["temp_extension_requested"] = False
    state[TEMP_EXTENSION_REQUEST_PASS] = None
    state[TEMP_EXTENSION_COUNT] = None
    state["temp_extension_count"] = None
    state[TEMP_RETRIEVAL_TOOL_BUDGET] = None
    state["temp_retrieval_tool_budget"] = None
    state["extension_rationale"] = None

    cleaned_synth_text = _clean_user_facing_synthesis(raw_synth_text)
    coordinator_view = build_online_coordinator_view(cleaned_synth_text, synth_candidates)
    return coordinator_view

_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/online_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_online_coordinator_view

build_online_coordinator_view(text, candidates)

Build a clean Markdown coordinator view with a trailing candidate papers mapping.

Appends a # Discovered Candidate Papers bibliography mapping 1-based candidate indices ([1], [2], ...) to bibliographic details and import rationales while omitting full abstracts to bound coordinator context token usage.

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

def build_online_coordinator_view(
    text: str,
    candidates: list[CandidatePaper] | None,
) -> str:
    """Build a clean Markdown coordinator view with a trailing candidate papers mapping.

    Appends a `# Discovered Candidate Papers` bibliography mapping 1-based candidate
    indices (`[1]`, `[2]`, ...) to bibliographic details and import rationales while
    omitting full abstracts to bound coordinator context token usage.
    """
    # Strip trailing candidate paper / reference headings for idempotency
    clean_text = re.sub(
        r"(?:\n{1,3}|^\s*)#{1,3}\s*(?:References|Cited Sources|Sources Consulted|Recommended Papers for Import|Discovered Candidate Papers)[\s\S]*$",
        "",
        text or "",
        flags=re.IGNORECASE,
    ).strip()
    if not candidates:
        return clean_text

    lines = ["# Discovered Candidate Papers"]
    for i, raw_c in enumerate(candidates, 1):
        extra_pdf_url = raw_c.get("pdf_url") if isinstance(raw_c, dict) else None
        c = CandidatePaper.from_dict(raw_c) if isinstance(raw_c, dict) else raw_c

        authors_str = _format_candidate_authors(getattr(c, "authors", None))
        year_str = str(c.year).strip() if getattr(c, "year", None) else "n.d."
        raw_title = str(getattr(c, "title", None) or "").strip()
        title_str = raw_title.rstrip(".") if raw_title else "Untitled"

        raw_venue = str(getattr(c, "venue", None) or "").strip().rstrip(".")
        venue_str = f" In {raw_venue}." if raw_venue else ""

        raw_doi = str(getattr(c, "doi", None) or "").strip()
        raw_url = str(
            getattr(c, "oa_pdf_url", None)
            or getattr(c, "pdf_url", None)
            or extra_pdf_url
            or ""
        ).strip()
        if raw_doi:
            doi_str = f" DOI: {raw_doi}"
        elif raw_url:
            doi_str = f" URL: {raw_url}"
        else:
            doi_str = ""

        entry = f"- [{i}] {authors_str} ({year_str}). {title_str}.{venue_str}{doi_str}"
        raw_tldr = str(getattr(c, "tldr", None) or "").strip()
        if raw_tldr:
            entry += f"\n  - Rationale: {raw_tldr}"

        lines.append(entry)

    candidate_section = "\n".join(lines)
    if clean_text:
        return f"{clean_text}\n\n{candidate_section}"
    return candidate_section

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)

create_online_search_agent

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

Create the stateless online literature search subagent.

Equipped with OpenAlex retrieval, candidate conclusion, and extension request tools.

Source code in zocrates/agents/online_search/agent.py

def create_online_search_agent(
    model: Any = None,
    thinking_level: str | None = "low",
    generate_content_config: types.GenerateContentConfig | None = None,
) -> Agent:
    """Create the stateless online literature search subagent.

    Equipped with OpenAlex retrieval, candidate conclusion, and extension request tools.
    """
    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="online_search_agent",
        model=model,
        instruction=ONLINE_SEARCH_AGENT_INSTRUCTION,
        tools=[
            FunctionTool(search_openalex),
            FunctionTool(conclude_online_search),
            FunctionTool(request_search_extension),
        ],
        generate_content_config=generate_content_config,
        mode="single_turn",
        include_contents="default",
    )

create_online_synthesizer_agent

create_online_synthesizer_agent(model=None, thinking_level='high', generate_content_config=None)

Create the online synthesizer subagent.

Produces an inquiry overview and structured candidate recommendations with numbered references.

Source code in zocrates/agents/online_search/agent.py

def create_online_synthesizer_agent(
    model: Any = None,
    thinking_level: str | None = "high",
    generate_content_config: types.GenerateContentConfig | None = None,
) -> Agent:
    """Create the online synthesizer subagent.

    Produces an inquiry overview and structured candidate recommendations with numbered references.
    """
    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="online_synthesizer_agent",
        model=model,
        instruction=ONLINE_SYNTHESIZER_AGENT_INSTRUCTION,
        tools=[],
        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

format_candidates_for_synthesizer

format_candidates_for_synthesizer(candidates)

Format candidate papers into an ordered list with sequential 1-based indices.

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

def format_candidates_for_synthesizer(candidates: list[CandidatePaper]) -> str:
    """Format candidate papers into an ordered list with sequential 1-based indices."""
    if not candidates:
        return "No candidate papers found."

    blocks = []
    for i, c in enumerate(candidates, 1):
        authors_str = ", ".join(c.authors) if c.authors else "Unknown"
        venue_str = c.venue or "Unknown Venue"
        oa_str = "Yes" if c.is_oa else "No"
        abstract_str = c.abstract or c.tldr or "No abstract or summary available."
        keywords_str = ", ".join(c.keywords) if c.keywords else "None"
        topic_str = c.primary_topic or "N/A"

        block = (
            f"Candidate [{i}]\n"
            f"- Candidate ID: {c.candidate_id}\n"
            f"- Title: {c.title}\n"
            f"- Authors: {authors_str}\n"
            f"- Year: {c.year or 'Unknown'}\n"
            f"- Venue: {venue_str} | Citations: {c.citation_count} | Open Access: {oa_str}\n"
            f"- DOI: {c.doi or 'None'} | arXiv ID: {c.arxiv_id or 'None'}\n"
            f"- Primary Topic: {topic_str}\n"
            f"- Keywords: {keywords_str}\n"
            f"- Abstract: {abstract_str}"
        )
        blocks.append(block)

    return "\n\n".join(blocks)

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

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

online_search_budget_checkpoint async

online_search_budget_checkpoint(ctx, node_input=None)

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

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

@node(name="online_search_budget_checkpoint", rerun_on_resume=False)
async def online_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="Online literature discovery limit reached. Extend search or synthesize current findings?",
        payload=payload,
    )

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

search_online_literature async

search_online_literature(request, tool_context)

Autonomous discovery tool executing the online scholarly literature search pipeline.

Use this tool when the user wants to discover new papers, preprints, external literature, or resolve external DOIs/arXiv links not in their library.

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

async def search_online_literature(request: str, tool_context: ToolContext) -> str:
    """Autonomous discovery tool executing the online scholarly literature search pipeline.

    Use this tool when the user wants to discover new papers, preprints, external literature,
    or resolve external DOIs/arXiv links not in their 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_online_pipeline_node,
        node_input=request,
        override_isolation_scope=f"search_online_pipeline:{fc_id}",
    )
    return coerce_to_text(result)

search_online_pipeline_node async

search_online_pipeline_node(ctx, node_input='')

Dynamic workflow function orchestrating online literature search and synthesis.

Wraps _run_online_search_pipeline in a try/finally so the node always closes its isolation scope (search_online_pipeline:{fc_id}) before returning, whether it succeeds ("Task completed.") or raises ("Task failed.").

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

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

    Wraps ``_run_online_search_pipeline`` in a try/finally so the node always closes
    its isolation scope (``search_online_pipeline:{fc_id}``) before returning, whether
    it succeeds ("Task completed.") or raises ("Task failed.").
    """
    _ensure_isolation_scope(ctx)

    completed = False
    interrupted = False
    try:
        response_text = await _run_online_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.",
            )