app

zocrates.agents.app

Application assembly and lifecycle management for Zocrates multi-agent systems.

ROLE_DEFAULT_THINKING module-attribute

ROLE_DEFAULT_THINKING = {'coordinator': 'medium', 'library_search': 'low', 'library_synthesizer': 'high', 'online_search': 'low', 'online_synthesizer': 'high'}

ZOCRATES_COMPACTION_PROMPT_TEMPLATE module-attribute

ZOCRATES_COMPACTION_PROMPT_TEMPLATE = '## OBJECTIVE\nYou are compacting the prior research dialogue history between a researcher and Zocrates (an AI academic research assistant).\n\nYour objective is to generate an accurate, dense, and grounded synthesis/summary of past dialogue turns so future research turns maintain complete context.\n\n## INSTRUCTIONS\n\n### Sections\nYou summary must include the following sections:\n- Sources: List all papers, studies, and documents discussed in the conversation using unnumbered bullet points with authors, publication year, paper title, and journal/source (e.g. "- Smith et al. (2021) — Mechanisms of BRCA-mediated repair (Nature Genetics)"). Do NOT use numeric citation brackets ([1], [2]) or numbered lists for sources.\n- Key Findings & Established Decisions: Summarize core findings, established facts, methodologies, and user decisions in dense bullet points. Attribute findings directly to the authors (e.g. "Petersen et al. (2022) showed...").\n- Unresolved Questions & Next Steps: Note any explicit knowledge gaps, ongoing literature inquiries, user preferences, or next steps identified in the conversation.\n\n### Critical Guidelines:\n- Exclude Internal Mechanics: Do not describe internal tool calls, retrieval loops, or intermediate system steps; focus strictly on final evidence obtained and conceptual conclusions.\n- Conciseness & Dense Structure: Keep the summary dense, direct, and under 400 words. Use structured bullet points rather than verbose paragraphs.\n- Do NOT include conversational pleasantries, greeting filler, or language declarations. \n- Do NOT attempt to answer any user questions found in the history. Your task is strictly to summarize the past dialogue into the sections above.\n\n## CONVERSATION HISTORY TO SUMMARIZE\n\n<dialogue_history>\n\n{conversation_history}\n\n</dialogue_history>\n\n## SUMMARY\nNow, provide the dense, structured academic compaction summary of the dialogue history above:\n'

__all__ module-attribute

__all__ = ['app', 'coordinator_agent', 'root_agent', 'set_agent_models', 'standalone_summarizer', 'get_compaction_summarizer', 'FailOpenLlmEventSummarizer', 'build_events_compaction_config', 'build_compaction_llm', 'ZOCRATES_COMPACTION_PROMPT_TEMPLATE']

app module-attribute

app = App(name='zocrates', root_agent=root_agent, events_compaction_config=None)

coordinator_agent module-attribute

coordinator_agent = create_coordinator_agent()

logger module-attribute

logger = logging.getLogger(__name__)

root_agent module-attribute

root_agent = coordinator_agent

standalone_summarizer module-attribute

standalone_summarizer = FailOpenLlmEventSummarizer(llm=build_compaction_llm(load_zocrates_config()), prompt_template=ZOCRATES_COMPACTION_PROMPT_TEMPLATE)

FailOpenLlmEventSummarizer

FailOpenLlmEventSummarizer(*args, **kwargs)

Bases: LlmEventSummarizer

Event summarizer with fail-open resilience and pipeline isolation scope filtering.

Initialize fail-open LLM event summarizer.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments forwarded to LlmEventSummarizer.	()
**kwargs	Any	Keyword arguments forwarded to LlmEventSummarizer.	{}

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

def __init__(
    self,
    *args: Any,
    **kwargs: Any,
) -> None:
    """Initialize fail-open LLM event summarizer.

    Args:
        *args: Positional arguments forwarded to `LlmEventSummarizer`.
        **kwargs: Keyword arguments forwarded to `LlmEventSummarizer`.
    """
    super().__init__(*args, **kwargs)
    self._last_compacted_turn_inv_id: str | None = None

maybe_summarize_events async

maybe_summarize_events(*, events)

Summarize conversation events with scope filtering and fail-open resilience.

Parameters:

Name	Type	Description	Default
events	list[Event]	List of session events to consider for compaction summarization.	required

Returns:

Type	Description
Event | None	Compaction event containing the generated summary, or None if skipped or on failure.

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

async def maybe_summarize_events(self, *, events: list[Event]) -> Event | None:
    """Summarize conversation events with scope filtering and fail-open resilience.

    Args:
        events: List of session events to consider for compaction summarization.

    Returns:
        Compaction event containing the generated summary, or None if skipped or on failure.
    """
    if not events:
        return None

    # 0. Single-turn idempotency: prevent running duplicate compaction within the exact same turn
    active_inv_id = _get_active_turn_invocation_id(events)
    if active_inv_id and active_inv_id == self._last_compacted_turn_inv_id:
        logger.debug(
            "Skipping duplicate compaction within the same invocation (%s).",
            active_inv_id,
        )
        return None

    # 1. Scope Filtering: Exclude scoped pipeline events (search_pipeline:{fc_id})
    # so compaction summarizes only high-level dialogue and clean tool returns,
    # without leaking internal ReAct retrieval loops or raw chunk payloads.
    unscoped_events = [
        e for e in events
        if getattr(e, "isolation_scope", None) is None
    ]
    if not unscoped_events:
        return None

    # Trim incomplete turns (e.g. trailing unanswered user questions or open tool calls)
    unscoped_events = _trim_to_completed_turns(unscoped_events)
    if not unscoped_events:
        return None

    with _start_compaction_span() as span:
        try:
            # Format prompt exclusively from unscoped high-level dialogue events
            conversation_history = self._format_events_for_prompt(unscoped_events)
            prompt = self._prompt_template.format(
                conversation_history=conversation_history
            )

            model_name = getattr(self._llm, "model", "unknown")
            _safe_set(span, "openinference.span.kind", "LLM")
            _safe_set(span, "llm.model_name", model_name)
            _safe_set(span, "gen_ai.request.model", model_name)
            _safe_set(span, "input.value", prompt)
            _safe_set(span, "input.mime_type", "text/plain")
            _safe_set(span, "compaction.total_events", len(events))
            _safe_set(span, "compaction.unscoped_events", len(unscoped_events))

            llm_request = LlmRequest(
                model=self._llm.model,
                contents=[Content(role="user", parts=[Part(text=prompt)])],
            )
            summary_content = None
            summary_usage_metadata = None
            async for llm_response in self._llm.generate_content_async(
                llm_request, stream=False
            ):
                if llm_response.content:
                    summary_content = llm_response.content
                    summary_usage_metadata = llm_response.usage_metadata
                    break

            if summary_content is None:
                _safe_set(span, "compaction.status", "empty_response")
                return None

            summary_text = "".join(
                p.text for p in summary_content.parts if getattr(p, "text", None)
            ).strip()
            _safe_set(span, "output.value", summary_text)
            _safe_set(span, "output.mime_type", "text/plain")

            if summary_usage_metadata:
                prompt_tokens = getattr(summary_usage_metadata, "prompt_token_count", None)
                completion_tokens = getattr(summary_usage_metadata, "candidates_token_count", None)
                total_tokens = getattr(summary_usage_metadata, "total_token_count", None)
                if prompt_tokens is not None:
                    _safe_set(span, "llm.token_count.prompt", prompt_tokens)
                    _safe_set(span, "gen_ai.usage.input_tokens", prompt_tokens)
                if completion_tokens is not None:
                    _safe_set(span, "llm.token_count.completion", completion_tokens)
                    _safe_set(span, "gen_ai.usage.output_tokens", completion_tokens)
                if total_tokens is not None:
                    _safe_set(span, "llm.token_count.total", total_tokens)
                    _safe_set(span, "gen_ai.usage.total_tokens", total_tokens)

            if _StatusCode and span is not None:
                span.set_status(_StatusCode.OK)

            summary_content.role = "model"

            # The compaction range must cover exactly the span of events that were summarized.
            # Any trailing events excluded by _trim_to_completed_turns will have timestamp > end_timestamp
            # and will thus be safely preserved verbatim for the coordinator by ADK's _process_compaction_events.
            start_timestamp = getattr(unscoped_events[0], "timestamp", 0.0)
            end_timestamp = getattr(unscoped_events[-1], "timestamp", 0.0)

            compaction = EventCompaction(
                start_timestamp=start_timestamp,
                end_timestamp=end_timestamp,
                compacted_content=summary_content,
            )

            actions = EventActions(compaction=compaction)

            if active_inv_id:
                self._last_compacted_turn_inv_id = active_inv_id

            return Event(
                author="user",
                actions=actions,
                invocation_id=Event.new_id(),
                usage_metadata=summary_usage_metadata,
            )
        except Exception as exc:
            if span is not None:
                _safe_set(span, "error.type", type(exc).__name__)
                _safe_set(span, "error.message", str(exc))
                if _StatusCode:
                    span.set_status(_StatusCode.ERROR, str(exc))
            logger.warning(
                "Event compaction failed open; continuing turn without summary: %s",
                exc,
            )
            return None

build_agent_llm

build_agent_llm(agent_cfg, global_cfg=None, zotero_data_dir=None)

Build a dedicated LiteLlm instance for an agent role.

Resolves configuration according to precedence

agent_cfg > global_cfg > defaults

Parameters:

Name	Type	Description	Default
agent_cfg	dict[str, Any]	Role-specific override dict (e.g. from override_agent_models).	required
global_cfg	dict[str, Any] | None	Optional global configuration dictionary (loaded if None).	None
zotero_data_dir	Path | str | None	Optional Zotero data directory path.	None

Returns:

Type	Description
BaseLlm	Configured ADK BaseLlm (LiteLlm) instance.

Source code in zocrates/llm.py

def build_agent_llm(
    agent_cfg: dict[str, Any],
    global_cfg: dict[str, Any] | None = None,
    zotero_data_dir: Path | str | None = None,
) -> BaseLlm:
    """Build a dedicated LiteLlm instance for an agent role.

    Resolves configuration according to precedence:
      agent_cfg > global_cfg > defaults

    Args:
        agent_cfg: Role-specific override dict (e.g. from override_agent_models).
        global_cfg: Optional global configuration dictionary (loaded if None).
        zotero_data_dir: Optional Zotero data directory path.

    Returns:
        Configured ADK BaseLlm (LiteLlm) instance.
    """
    if global_cfg is None:
        global_cfg = load_zocrates_config(zotero_data_dir)

    raw_provider = agent_cfg.get("provider")
    raw_model = (
        agent_cfg.get("model")
        or global_cfg.get("llm_model")
        or PROVIDER_DEFAULT_MODELS.get(raw_provider or global_cfg.get("llm_provider", "openrouter_llm"), DEFAULT_LLM_MODEL)
    )

    if raw_provider:
        provider = raw_provider
    elif raw_model.startswith("ollama/") or raw_model.startswith("ollama_chat/"):
        provider = "ollama"
    elif raw_model.startswith("openai/"):
        provider = "vllm"
    elif raw_model.startswith("openrouter/"):
        provider = "openrouter_llm"
    else:
        provider = global_cfg.get("llm_provider") or "openrouter_llm"

    model_name = _resolve_model_name(raw_model, provider)

    kwargs: dict[str, Any] = {}

    # 1. OpenRouter (Remote LLM)
    if provider in ("openrouter", "openrouter_llm"):
        key = (
            agent_cfg.get("api_key")
            or get_secure_api_key("openrouter_api_key", ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY"))
        )
        if key:
            kwargs["api_key"] = key
        kwargs["extra_headers"] = {
            # "HTTP-Referer": "https://github.com/alexplaka/Zocrates",
            "X-OpenRouter-Title": "Zocrates",
        }

    # 2. Local Ollama
    elif provider == "ollama":
        raw_url = (
            agent_cfg.get("base_url")
            or global_cfg.get("llm_base_url")
            or "http://localhost:11434"
        )
        kwargs["api_base"] = _format_base_url(raw_url, "ollama")
        extra_opts = dict(global_cfg.get("llm_extra_options") or {})
        extra_opts.update(agent_cfg.get("extra_options") or {})
        effective_num_ctx = extra_opts.get("num_ctx") or get_ollama_num_ctx(cfg=global_cfg)
        try:
            kwargs["num_ctx"] = int(effective_num_ctx)
        except (TypeError, ValueError):
            kwargs["num_ctx"] = DEFAULT_OLLAMA_NUM_CTX
        other_opts = {k: v for k, v in extra_opts.items() if k != "num_ctx"}
        if other_opts:
            kwargs["extra_body"] = {"options": other_opts}

    # 3. Local vLLM / OpenAI-compatible
    elif provider in ("vllm", "openai_compatible"):
        default_url = "http://localhost:8000/v1"
        raw_url = (
            agent_cfg.get("base_url")
            or global_cfg.get("llm_base_url")
            or default_url
        )
        kwargs["api_base"] = _format_base_url(raw_url, provider)
        kwargs["api_key"] = (
            agent_cfg.get("api_key")
            or get_secure_api_key("llm_api_key", ("LLM_API_KEY",))
            or global_cfg.get("llm_api_key")
            or "dummy"
        )

    # Temperature (opt-in; None omitted)
    if "temperature" in agent_cfg and agent_cfg["temperature"] is not None:
        kwargs["temperature"] = float(agent_cfg["temperature"])
    elif (temp := global_cfg.get("llm_temperature")) is not None:
        kwargs["temperature"] = float(temp)

    # Max tokens
    if "max_tokens" in agent_cfg and agent_cfg["max_tokens"] is not None:
        kwargs["max_tokens"] = int(agent_cfg["max_tokens"])
    else:
        if (max_tok := global_cfg.get("llm_max_tokens")) is not None:
            kwargs["max_tokens"] = int(max_tok)
        else:
            kwargs["max_tokens"] = DEFAULT_LLM_MAX_TOKENS

        if provider in ("vllm", "openai_compatible") and kwargs.get("max_tokens") == 16384:
            kwargs["max_tokens"] = 4096

    if provider in ("ollama", "vllm", "openai_compatible"):
        kwargs["drop_params"] = True

    # Retries
    if "num_retries" in agent_cfg and agent_cfg["num_retries"] is not None:
        kwargs["num_retries"] = int(agent_cfg["num_retries"])
    elif (num_retries := global_cfg.get("llm_num_retries")) is not None:
        kwargs["num_retries"] = int(num_retries)
    else:
        kwargs["num_retries"] = DEFAULT_LLM_NUM_RETRIES

    return LiteLlm(model=model_name, **kwargs)

build_compaction_llm

build_compaction_llm(cfg, zotero_data_dir=None, default_model=None)

Build a fast, dedicated LiteLlm instance for context compaction with suppressed reasoning effort.

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

def build_compaction_llm(
    cfg: dict[str, Any],
    zotero_data_dir: Path | str | None = None,
    default_model: Any = None,
) -> BaseLlm:
    """Build a fast, dedicated LiteLlm instance for context compaction with suppressed reasoning effort."""
    model_override = cfg.get("compaction_summarizer_model")
    if not model_override:
        if isinstance(default_model, dict):
            coord_val = default_model.get("coordinator") or default_model.get("coordinator_agent")
            if isinstance(coord_val, dict):
                model_override = coord_val.get("model")
            elif hasattr(coord_val, "model"):
                model_override = coord_val.model
            elif isinstance(coord_val, str):
                model_override = coord_val
        elif hasattr(default_model, "model"):
            model_override = default_model.model
        elif isinstance(default_model, str) and default_model:
            model_override = default_model

    if not model_override:
        model_override = DEFAULT_COMPACTION_SUMMARIZER_MODEL

    llm = build_agent_llm(
        {"model": model_override, "max_tokens": 1200, "temperature": 0.2},
        cfg,
        zotero_data_dir=zotero_data_dir,
    )
    # Suppress reasoning/thinking effort for compaction
    _suppress_reasoning_effort(llm)
    return llm

build_events_compaction_config

build_events_compaction_config(cfg=None, zotero_data_dir=None, default_model=None)

Deactivated: native ADK pre-invocation triggers are replaced by post-turn adaptive compaction.

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

def build_events_compaction_config(
    cfg: dict[str, Any] | None = None,
    zotero_data_dir: Path | str | None = None,
    default_model: Any = None,
) -> EventsCompactionConfig | None:
    """Deactivated: native ADK pre-invocation triggers are replaced by post-turn adaptive compaction."""
    return None

create_coordinator_agent

create_coordinator_agent(model=None, thinking_level='medium', generate_content_config=None, search_library_tool=None, search_online_tool=None)

Create the root Coordinator agent.

The coordinator answers conversational turns directly and delegates literature retrieval queries to the autonomous search pipeline(s). model can be a single model or a dict specifying per-agent models.

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

def create_coordinator_agent(
    model: Any = None,
    thinking_level: str | None = "medium",
    generate_content_config: types.GenerateContentConfig | None = None,
    search_library_tool: Any = None,
    search_online_tool: Any = None,
) -> Agent:
    """Create the root Coordinator agent.

    The coordinator answers conversational turns directly and delegates
    literature retrieval queries to the autonomous search pipeline(s).
    `model` can be a single model or a dict specifying per-agent models.
    """
    if model is None:
        base_model = get_llm_model()
    elif isinstance(model, dict):
        base_model = (
            model.get("default")
            or model.get("coordinator")
            or get_llm_model()
        )
    else:
        base_model = model

    if not isinstance(base_model, (str, BaseLlm)) and hasattr(base_model, "_mock_name"):
        base_model = "test-model"

    library_search_subagent = create_library_search_agent(base_model, thinking_level="low")
    library_synth_subagent = create_library_synthesizer_agent(base_model, thinking_level="high")
    online_search_subagent = create_online_search_agent(base_model, thinking_level="low")
    online_synth_subagent = create_online_synthesizer_agent(base_model, thinking_level="high")

    if search_library_tool is None:
        search_library_tool = FunctionTool(func=search_library)
    if search_online_tool is None:
        search_online_tool = FunctionTool(func=search_online_literature)

    tools = [search_library_tool, search_online_tool]

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

    coordinator = Agent(
        name="coordinator_agent",
        model=base_model,
        static_instruction=COORDINATOR_INSTRUCTION,
        instruction=get_coordinator_instruction,
        tools=tools,
        generate_content_config=generate_content_config,
        mode="chat",
    )
    coordinator._search_tool = search_library_tool
    coordinator._online_search_tool = search_online_tool
    coordinator._subagents = {
        "library_search": library_search_subagent,
        "library_synthesizer": library_synth_subagent,
        "online_search": online_search_subagent,
        "online_synthesizer": online_synth_subagent,
    }

    if isinstance(model, dict):
        set_agent_models(coordinator, model)

    return coordinator

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_compaction_summarizer

get_compaction_summarizer()

Return the active standalone event summarizer instance.

Source code in zocrates/agents/app.py

def get_compaction_summarizer() -> FailOpenLlmEventSummarizer:
    """Return the active standalone event summarizer instance."""
    return standalone_summarizer

get_llm_model

get_llm_model(zotero_data_dir=None)

Instantiate the active ADK BaseLlm instance based on Zocrates configuration.

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path | str | None	Optional path to Zotero data directory.	None

Returns:

Type	Description
BaseLlm	Configured ADK BaseLlm (LiteLlm) instance.

Source code in zocrates/llm.py

def get_llm_model(zotero_data_dir: Path | str | None = None) -> BaseLlm:
    """Instantiate the active ADK BaseLlm instance based on Zocrates configuration.

    Args:
        zotero_data_dir: Optional path to Zotero data directory.

    Returns:
        Configured ADK BaseLlm (LiteLlm) instance.
    """
    return build_agent_llm({}, zotero_data_dir=zotero_data_dir)

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

patch_adk_active_task_scope_rewind

patch_adk_active_task_scope_rewind()

Patch ADK _find_active_task_scope to filter out rewound events before checking active scopes.

When an in-flight subagent pipeline (which stamps events with an isolation_scope, e.g. 'search_library_pipeline:xyz' or 'search_online_pipeline:xyz') is rewound via rewind_before_invocation_id, its unfinished isolation scope remains in session.events. ADK's native _find_active_task_scope walks session.events directly without taking rewinds into account, causing subsequent user turns to mistakenly be stamped with the dead subagent's isolation scope.

This patch ensures _find_active_task_scope evaluates active task scopes against the live events filtered by _apply_rewinds(session.events).

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

def patch_adk_active_task_scope_rewind() -> None:
    """Patch ADK _find_active_task_scope to filter out rewound events before checking active scopes.

    When an in-flight subagent pipeline (which stamps events with an isolation_scope,
    e.g. 'search_library_pipeline:xyz' or 'search_online_pipeline:xyz') is rewound via
    rewind_before_invocation_id, its unfinished isolation scope remains in session.events.
    ADK's native _find_active_task_scope walks session.events directly without taking rewinds
    into account, causing subsequent user turns to mistakenly be stamped with the dead
    subagent's isolation scope.

    This patch ensures _find_active_task_scope evaluates active task scopes against
    the live events filtered by _apply_rewinds(session.events).
    """
    global _adk_active_task_scope_rewind_patched
    if _adk_active_task_scope_rewind_patched:
        return

    try:
        import google.adk.runners as adk_runners
        from google.adk.flows.llm_flows.contents import _apply_rewinds

        orig_find_active_task_scope = adk_runners._find_active_task_scope

        class _RewoundSessionProxy:
            """Lightweight proxy exposing rewound events while preserving session attributes."""

            def __init__(self, session: Any, live_events: list[Any]) -> None:
                """Initialize session proxy wrapping live non-rewound events.

                Args:
                    session: The underlying ADK session instance.
                    live_events: Filtered list of active session events after applying rewinds.
                """
                self._session = session
                self.events = live_events

            def __getattr__(self, name: str) -> Any:
                """Delegate attribute lookups to the underlying session instance.

                Args:
                    name: Attribute name to resolve on the underlying session.

                Returns:
                    The attribute value from the underlying session.
                """
                return getattr(self._session, name)

        def patched_find_active_task_scope(session: Any) -> tuple[str, str] | None:
            """Find the active task scope by filtering out rewound events first.

            Args:
                session: The ADK session to inspect for an active task scope.

            Returns:
                A tuple of (scope_id, task_name) if an active scope exists, otherwise None.
            """
            if not session or getattr(session, "events", None) is None:
                return None
            live_events = _apply_rewinds(session.events)
            proxy = _RewoundSessionProxy(session, live_events)
            return orig_find_active_task_scope(proxy)

        adk_runners._find_active_task_scope = patched_find_active_task_scope
        _adk_active_task_scope_rewind_patched = True
        logger.debug(
            "Patched google.adk.runners._find_active_task_scope to respect _apply_rewinds"
        )
    except Exception as err:
        logger.warning("Could not patch ADK _find_active_task_scope: %s", err)

patch_adk_request_input_context_filter

patch_adk_request_input_context_filter()

Patch ADK contents filter to exclude adk_request_input function calls/responses from agent LLM context.

In dynamic workflows with Human-In-The-Loop (HITL), RequestInput emits an Event with a function call named 'adk_request_input' and the user resume response emits a function response named 'adk_request_input'. While ADK natively filters out internal framework events like 'adk_framework' and 'adk_request_credential' from being included in an agent's LLM context, it omits 'adk_request_input'. This causes ADK's _fencing module to wrap the checkpoint call and response in cross-agent fencing (<<>>) and inject them into search_subagent's prompt during extension passes.

This patch ensures 'adk_request_input' events are filtered out from LLM context generation, preserving clean continuation prompts for subagents.

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

def patch_adk_request_input_context_filter() -> None:
    """Patch ADK contents filter to exclude adk_request_input function calls/responses from agent LLM context.

    In dynamic workflows with Human-In-The-Loop (HITL), RequestInput emits an Event
    with a function call named 'adk_request_input' and the user resume response emits
    a function response named 'adk_request_input'. While ADK natively filters out
    internal framework events like 'adk_framework' and 'adk_request_credential' from
    being included in an agent's LLM context, it omits 'adk_request_input'. This causes
    ADK's _fencing module to wrap the checkpoint call and response in cross-agent
    fencing (<<<BEGIN_QUOTED_AGENT_CONTENT>>>) and inject them into search_subagent's
    prompt during extension passes.

    This patch ensures 'adk_request_input' events are filtered out from LLM context
    generation, preserving clean continuation prompts for subagents.
    """
    global _adk_request_input_context_filter_patched
    if _adk_request_input_context_filter_patched:
        return

    try:
        import google.adk.flows.llm_flows.contents as adk_contents

        orig_should_include = adk_contents._should_include_event_in_context

        def _is_request_input_event(event: Any) -> bool:
            """Check if an event contains an adk_request_input tool call or response.

            Args:
                event: The event to inspect for request-input parts.

            Returns:
                True if the event contains an adk_request_input call/response, False otherwise.
            """
            content = getattr(event, "content", None)
            parts = getattr(content, "parts", None) if content else None
            if not parts:
                return False
            for part in parts:
                fc = getattr(part, "function_call", None)
                if fc and getattr(fc, "name", None) == "adk_request_input":
                    return True
                fr = getattr(part, "function_response", None)
                if fr and getattr(fr, "name", None) == "adk_request_input":
                    return True
            return False

        def patched_should_include_event_in_context(
            current_branch: str | None,
            event: Any,
            isolation_scope: str | None = None,
            *,
            include_thoughts: bool = False,
            **kwargs: Any,
        ) -> bool:
            """Filter out adk_request_input events before delegating to the original ADK context filter.

            Args:
                current_branch: The current execution branch in the session flow.
                event: The session event being evaluated for inclusion in LLM context.
                isolation_scope: Active isolation scope identifier, if any.
                include_thoughts: Whether to retain agent reasoning/thought parts.
                **kwargs: Additional parameters forwarded to the original ADK filter.

            Returns:
                False if event is an adk_request_input event, otherwise the original filter decision.
            """
            if _is_request_input_event(event):
                return False
            return orig_should_include(
                current_branch,
                event,
                isolation_scope=isolation_scope,
                include_thoughts=include_thoughts,
                **kwargs,
            )

        adk_contents._should_include_event_in_context = patched_should_include_event_in_context
        _adk_request_input_context_filter_patched = True
        logger.debug(
            "Patched google.adk.flows.llm_flows.contents._should_include_event_in_context to filter adk_request_input"
        )
    except Exception as err:
        logger.warning("Could not patch ADK contents _should_include_event_in_context: %s", err)

patch_compaction_subagent_isolation

patch_compaction_subagent_isolation()

Ensure token compaction only executes for the root coordinator agent, never for subagents, and eliminate duplicate post-invocation compaction runs.

In multi-agent systems, subagents (e.g. library_search_agent, online_search_agent) execute inside isolated pipelines. If token-threshold compaction runs inside subagents, it mistakenly compacts the parent session's history before every internal tool call, causing massive latency overhead and redundant compaction cycles.

Additionally, Google ADK 2.8.0 contains an upstream context propagation bug where BaseAgent._create_invocation_context performs a shallow model_copy() of InvocationContext, causing token_compaction_checked = True to only be set on the child context and never reaching the root context in runners.py. This causes runners.py to pass skip_token_compaction=False to _run_post_invocation_compaction, triggering a second, completely redundant compaction cycle. This patch links parent contexts and guards _run_compaction_for_token_threshold so compaction never executes more than once per turn.

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

def patch_compaction_subagent_isolation() -> None:
    """Ensure token compaction only executes for the root coordinator agent, never for subagents,
    and eliminate duplicate post-invocation compaction runs.

    In multi-agent systems, subagents (e.g. library_search_agent, online_search_agent)
    execute inside isolated pipelines. If token-threshold compaction runs inside subagents,
    it mistakenly compacts the parent session's history before every internal tool call,
    causing massive latency overhead and redundant compaction cycles.

    Additionally, Google ADK 2.8.0 contains an upstream context propagation bug where
    `BaseAgent._create_invocation_context` performs a shallow `model_copy()` of InvocationContext,
    causing `token_compaction_checked = True` to only be set on the child context and never reaching
    the root context in `runners.py`. This causes runners.py to pass `skip_token_compaction=False` to
    `_run_post_invocation_compaction`, triggering a second, completely redundant compaction cycle.
    This patch links parent contexts and guards `_run_compaction_for_token_threshold` so compaction
    never executes more than once per turn.
    """
    global _compaction_subagent_isolation_patched
    if _compaction_subagent_isolation_patched:
        return

    try:
        from google.adk.agents.base_agent import BaseAgent
        import google.adk.apps.compaction as adk_app_compaction
        import google.adk.flows.llm_flows.compaction as adk_compaction

        # 1. Link child InvocationContext to parent context so flags propagate back to root runner
        orig_create_invocation_context = BaseAgent._create_invocation_context

        def patched_create_invocation_context(agent_self: Any, parent_context: Any) -> Any:
            """Create child invocation context and attach a reference to its parent context.

            Args:
                agent_self: The agent instance creating the invocation context.
                parent_context: The parent invocation context to link against.

            Returns:
                The created child invocation context with `_parent_context` attached.
            """
            child = orig_create_invocation_context(agent_self, parent_context)
            child._parent_context = parent_context
            return child

        BaseAgent._create_invocation_context = patched_create_invocation_context

        # 2. Patch CompactionRequestProcessor to isolate token compaction to root coordinator
        # and propagate token_compaction_checked flag to parent contexts
        orig_run_async = adk_compaction.CompactionRequestProcessor.run_async

        async def patched_run_async(self: Any, invocation_context: Any, llm_request: Any) -> Any:
            """Execute compaction request processor only for root coordinator agent instances.

            Args:
                self: The CompactionRequestProcessor instance.
                invocation_context: Current invocation context.
                llm_request: The pending LLM request.

            Yields:
                Events yielded by the original compaction request processor.
            """
            # Skip if running inside an isolation scope (subagent pipeline)
            if getattr(invocation_context, "isolation_scope", None) is not None:
                return
                yield

            # Skip if agent is a subagent (only coordinator_agent compacts session history)
            agent = getattr(invocation_context, "agent", None)
            if agent and getattr(agent, "name", None) not in ("coordinator_agent", "coordinator"):
                return
                yield

            async for event in orig_run_async(self, invocation_context, llm_request):
                yield event

            # Propagate token_compaction_checked back to root invocation context in runner loop
            if getattr(invocation_context, "token_compaction_checked", False):
                curr = getattr(invocation_context, "_parent_context", None)
                while curr is not None:
                    curr.token_compaction_checked = True
                    curr = getattr(curr, "_parent_context", None)

        adk_compaction.CompactionRequestProcessor.run_async = patched_run_async
        adk_compaction.request_processor.run_async = patched_run_async.__get__(
            adk_compaction.request_processor, adk_compaction.CompactionRequestProcessor
        )

        # 3. Guard post-invocation token-threshold compaction against duplicate execution
        orig_token_compaction = adk_app_compaction._run_compaction_for_token_threshold

        async def patched_run_compaction_for_token_threshold(
            app: Any, session: Any, session_service: Any
        ) -> bool:
            """Guard post-invocation token-threshold compaction against duplicate execution per turn.

            Args:
                app: The ADK application instance.
                session: The active session being evaluated for compaction.
                session_service: Session service handling session persistence.

            Returns:
                True if compaction was executed, False if skipped or not triggered.
            """
            if session and getattr(session, "events", None) and _has_compaction_for_active_turn(session.events):
                logger.debug(
                    "Skipping duplicate post-invocation token-threshold compaction: "
                    "a compaction event already exists for the active turn."
                )
                return False
            return await orig_token_compaction(app, session, session_service)

        adk_app_compaction._run_compaction_for_token_threshold = patched_run_compaction_for_token_threshold

        # 4. Patch Runner._run_post_invocation_compaction:
        # - Suppress compaction when invocation is interrupted mid-turn for HITL (e.g. search_budget_checkpoint)
        # - Always skip token-threshold compaction in post-invocation since it is strictly evaluated pre-invocation,
        #   preventing duplicate runs, empty Phoenix spans, and unnecessary end-of-turn latency.
        from google.adk.runners import Runner

        orig_post_compaction = Runner._run_post_invocation_compaction

        async def patched_post_invocation_compaction(
            runner_self: Any,
            *,
            session: Any,
            skip_token_compaction: bool,
        ) -> None:
            """Run adaptive post-turn compaction and suppress redundant post-invocation compaction.

            Args:
                runner_self: The Runner instance executing post-invocation steps.
                session: The active session.
                skip_token_compaction: Whether to skip token-threshold compaction.
            """
            if session and getattr(session, "events", None):
                last_ev = session.events[-1]
                if last_ev.content and any(getattr(p, "function_call", None) for p in last_ev.content.parts):
                    logger.debug("Skipping post-invocation compaction: invocation was interrupted mid-turn for HITL.")
                    return
                if getattr(last_ev, "actions", None) and (
                    getattr(last_ev.actions, "requested_tool_confirmations", None)
                    or getattr(last_ev.actions, "requested_auth_configs", None)
                ):
                    logger.debug("Skipping post-invocation compaction: invocation awaiting tool confirmation.")
                    return

            try:
                from zocrates.agents.common.compaction import run_adaptive_post_invocation_compaction

                await run_adaptive_post_invocation_compaction(runner_self, session)
            except Exception as exc:
                logger.warning("Adaptive post-invocation compaction failed open: %s", exc)

            await orig_post_compaction(
                runner_self,
                session=session,
                skip_token_compaction=True,
            )

        Runner._run_post_invocation_compaction = patched_post_invocation_compaction

        _compaction_subagent_isolation_patched = True
        logger.debug("Patched ADK CompactionRequestProcessor, Runner, and post-invocation token compaction guards")
    except Exception as err:
        logger.warning("Could not patch ADK CompactionRequestProcessor: %s", err)

set_agent_models

set_agent_models(new_model=None, zotero_data_dir=None)

Propagate updated LLM model(s) across coordinator, subagents, and compaction summarizer.

Parameters:

Name	Type	Description	Default
new_model	Any	BaseLlm instance, model name string, or dict mapping role keys to models. If None, re-instantiates default model via get_llm_model().	None
zotero_data_dir	Any	Optional Zotero data directory path for config loading.	None

Returns:

Type	Description
Any	The updated model or model configuration.

Source code in zocrates/agents/app.py

def set_agent_models(new_model: Any = None, zotero_data_dir: Any = None) -> Any:
    """Propagate updated LLM model(s) across coordinator, subagents, and compaction summarizer.

    Args:
        new_model: BaseLlm instance, model name string, or dict mapping role keys to models.
                   If None, re-instantiates default model via get_llm_model().
        zotero_data_dir: Optional Zotero data directory path for config loading.

    Returns:
        The updated model or model configuration.
    """
    cfg = load_zocrates_config(zotero_data_dir)

    if isinstance(new_model, dict):
        set_coordinator_agent_models(coordinator_agent, new_model)
        app.events_compaction_config = None
        standalone_summarizer._llm = build_compaction_llm(
            cfg, zotero_data_dir=zotero_data_dir, default_model=new_model
        )
        return new_model

    global_model = new_model or get_llm_model(zotero_data_dir=zotero_data_dir)
    overrides = cfg.get("override_agent_models") or {}

    agent_config: dict[str, dict[str, Any]] = {}
    is_mock_llm = hasattr(new_model, "script") or getattr(new_model, "model", None) == "mock-llm"

    for role in (
        "coordinator",
        "library_search",
        "library_synthesizer",
        "online_search",
        "online_synthesizer",
    ):
        override = overrides.get(role) or {}
        model_override = override.get("model")
        has_explicit_model = bool(model_override and str(model_override).strip())

        if not is_mock_llm and has_explicit_model:
            role_model = build_agent_llm(override, cfg, zotero_data_dir=zotero_data_dir)
        elif not is_mock_llm and any(
            k in override
            for k in (
                "provider",
                "temperature",
                "max_tokens",
                "extra_options",
                "base_url",
            )
        ):
            role_model = build_agent_llm(override, cfg, zotero_data_dir=zotero_data_dir)
        else:
            role_model = global_model

        default_thinking = ROLE_DEFAULT_THINKING.get(role, "medium")
        thinking_level = override.get("thinking_level", default_thinking)
        gen_config = get_agent_generate_content_config(
            thinking_level, zotero_data_dir=zotero_data_dir, model=role_model
        )
        agent_config[role] = {
            "model": role_model,
            "generate_content_config": gen_config,
        }

    set_coordinator_agent_models(coordinator_agent, agent_config)
    app.events_compaction_config = None
    standalone_summarizer._llm = build_compaction_llm(
        cfg, zotero_data_dir=zotero_data_dir, default_model=global_model
    )
    return global_model

set_coordinator_agent_models

set_coordinator_agent_models(coordinator, new_model)

Propagate updated LLM model(s) and content configs across coordinator and subagents.

Resolves 1:1 canonical role keys (coordinator, library_search, library_synthesizer, online_search, online_synthesizer).

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

def set_agent_models(coordinator: Agent, new_model: Any) -> None:
    """Propagate updated LLM model(s) and content configs across coordinator and subagents.

    Resolves 1:1 canonical role keys (coordinator, library_search, library_synthesizer,
    online_search, online_synthesizer).
    """
    if not isinstance(new_model, dict):
        coordinator.model = new_model
        if hasattr(coordinator, "_subagents") and isinstance(
            coordinator._subagents, dict
        ):
            for subagent in coordinator._subagents.values():
                if hasattr(subagent, "model"):
                    subagent.model = new_model
        logger.debug(
            "Propagated uniform model update across coordinator agent tree: %s",
            new_model,
        )
        return

    default_entry = new_model.get("default")

    # 1. Resolve coordinator model/config
    coord_entry = new_model.get("coordinator")
    if coord_entry is None and default_entry is not None:
        coord_entry = default_entry
    if coord_entry is not None:
        _apply_agent_model_or_config(coordinator, coord_entry)

    # 2. Resolve subagents models/configs
    if hasattr(coordinator, "_subagents") and isinstance(
        coordinator._subagents, dict
    ):
        subagents = coordinator._subagents
        roles = (
            "library_search",
            "library_synthesizer",
            "online_search",
            "online_synthesizer",
        )

        for role_name in roles:
            agent_obj = subagents.get(role_name)
            if agent_obj is None:
                continue
            target_entry = new_model.get(role_name)
            if target_entry is None and default_entry is not None:
                target_entry = default_entry

            if target_entry is not None:
                _apply_agent_model_or_config(agent_obj, target_entry)

    logger.debug(
        "Propagated per-agent model updates across coordinator agent tree: %s",
        new_model,
    )