compaction

zocrates.agents.common.compaction

Google ADK 2.8+ session context compaction utilities for Zocrates.

Provides fail-open event summarization, pipeline isolation scope filtering, and thought-stripped markdown formatting for dialogue history.

DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS module-attribute

DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS = 2

DEFAULT_COMPACTION_SUMMARIZER_MODEL module-attribute

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

DEFAULT_ENABLE_ADAPTIVE_COMPACTION module-attribute

DEFAULT_ENABLE_ADAPTIVE_COMPACTION = True

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__ = ['ZOCRATES_COMPACTION_PROMPT_TEMPLATE', 'FailOpenLlmEventSummarizer', 'build_compaction_llm', 'build_events_compaction_config', 'compact_session_adaptive', 'run_adaptive_post_invocation_compaction']

logger module-attribute

logger = logging.getLogger(__name__)

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

_extract_compacted_text

_extract_compacted_text(compacted_content)

Extract clean summary text from compacted_content (Content, str, or dict).

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

def _extract_compacted_text(compacted_content: Any) -> str:
    """Extract clean summary text from compacted_content (Content, str, or dict)."""
    if not compacted_content:
        return ""
    if isinstance(compacted_content, str):
        return compacted_content.strip()
    parts = getattr(compacted_content, "parts", None)
    if parts is None and isinstance(compacted_content, dict):
        parts = compacted_content.get("parts")
    if parts:
        texts = []
        for p in parts:
            if isinstance(p, dict):
                t = p.get("text")
            else:
                t = getattr(p, "text", None)
            if t:
                texts.append(str(t).strip())
        return "\n".join(texts).strip()
    return str(compacted_content).strip()

_get_active_turn_invocation_id

_get_active_turn_invocation_id(events)

Find the invocation_id of the active turn (the latest non-compaction invocation).

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

def _get_active_turn_invocation_id(events: list[Event]) -> str | None:
    """Find the invocation_id of the active turn (the latest non-compaction invocation)."""
    for e in reversed(events):
        if e.invocation_id and not (getattr(e, "actions", None) and getattr(e.actions, "compaction", None)):
            return e.invocation_id
    return None

_get_completed_turn_groups

_get_completed_turn_groups(events)

Group unscoped, non-compaction events into chronological completed turns.

Returns a list of (invocation_id, turn_events) pairs for turns that have an assistant completion. Incomplete turns (e.g. unanswered user queries or open tool calls) are excluded.

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

def _get_completed_turn_groups(events: list[Event]) -> list[tuple[str, list[Event]]]:
    """Group unscoped, non-compaction events into chronological completed turns.

    Returns a list of (invocation_id, turn_events) pairs for turns that have an assistant completion.
    Incomplete turns (e.g. unanswered user queries or open tool calls) are excluded.
    """
    if not events:
        return []

    invocations: dict[str, list[Event]] = {}
    inv_order: list[str] = []
    synthetic_counter = 0
    current_inv = None

    for e in events:
        inv_id = e.invocation_id
        if not inv_id:
            if e.author == "user":
                synthetic_counter += 1
                current_inv = f"_synthetic_turn_{synthetic_counter}"
            inv_id = current_inv or f"_synthetic_turn_{synthetic_counter}"
        else:
            current_inv = inv_id

        if inv_id not in invocations:
            invocations[inv_id] = []
            inv_order.append(inv_id)
        invocations[inv_id].append(e)

    completed: list[tuple[str, list[Event]]] = []
    for inv_id in inv_order:
        inv_events = invocations[inv_id]
        has_assistant_completion = any(
            e.author != "user"
            and (
                (
                    e.content
                    and any(
                        p.text
                        for p in e.content.parts
                        if not getattr(p, "thought", False)
                    )
                )
                or any(
                    getattr(p, "function_response", None)
                    for p in (e.content.parts if e.content else [])
                )
            )
            for e in inv_events
        )
        if has_assistant_completion:
            completed.append((inv_id, inv_events))

    return completed

_is_compaction_event

_is_compaction_event(e)

Check if event is an ADK compaction event.

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

def _is_compaction_event(e: Any) -> bool:
    """Check if event is an ADK compaction event."""
    actions = getattr(e, "actions", None)
    if actions is None:
        return False
    return getattr(actions, "compaction", None) is not None

_prepare_compaction_slice

_prepare_compaction_slice(unscoped_events, live_events, cfg, summarizer)

Prepares the conversation history prompt, timestamps, and metadata for adaptive compaction.

Returns (prompt, start_timestamp, end_timestamp, slice_turns_count, k) or None. Fails open to None on any error without raising.

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

def _prepare_compaction_slice(
    unscoped_events: list[Event],
    live_events: list[Event],
    cfg: dict[str, Any],
    summarizer: Any,
) -> tuple[str, float, float, int, int] | None:
    """Prepares the conversation history prompt, timestamps, and metadata for adaptive compaction.

    Returns (prompt, start_timestamp, end_timestamp, slice_turns_count, k) or None.
    Fails open to None on any error without raising.
    """
    try:
        completed_turns = _get_completed_turn_groups(unscoped_events)
        n = len(completed_turns)

        k = max(
            0,
            int(
                cfg.get(
                    "adaptive_compaction_retention_turns",
                    DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS,
                )
            ),
        )

        if n <= k:
            logger.debug(
                "Adaptive compaction skipped: completed turns (%d) <= retention turns (%d).",
                n,
                k,
            )
            return None

        candidate_turns = completed_turns[: n - k]

        # Look for latest non-subsumed compaction event in live_events
        prev_compaction_event = None
        for e in reversed(live_events):
            actions = getattr(e, "actions", None)
            comp = getattr(actions, "compaction", None) if actions else None
            if comp is not None:
                prev_compaction_event = e
                break

        prev_comp = (
            getattr(getattr(prev_compaction_event, "actions", None), "compaction", None)
            if prev_compaction_event
            else None
        )
        prev_end_ts = getattr(prev_comp, "end_timestamp", None) if prev_comp else None

        if prev_comp is not None and prev_end_ts is not None:
            prev_end_ts_f = float(prev_end_ts)
            new_candidate_turns = []
            for inv_id, turn_evts in candidate_turns:
                turn_max_ts = max(
                    (getattr(e, "timestamp", 0.0) for e in turn_evts), default=0.0
                )
                if float(turn_max_ts) > prev_end_ts_f:
                    new_candidate_turns.append((inv_id, turn_evts))

            new_slice_events = [
                e
                for _, turn_evts in new_candidate_turns
                for e in turn_evts
                if float(getattr(e, "timestamp", 0.0)) > prev_end_ts_f
            ]

            if not new_slice_events:
                logger.debug(
                    "Adaptive compaction skipped: all candidate turns already subsumed by previous compaction."
                )
                return None

            previous_summary = _extract_compacted_text(prev_comp.compacted_content)
            formatted_new_events = summarizer._format_events_for_prompt(new_slice_events)
            conversation_history = (
                "### Existing Conversation Summary\n"
                f"{previous_summary}\n\n"
                "### New Completed Turns to Integrate\n"
                f"{formatted_new_events}"
            )
            prev_start_ts = getattr(prev_comp, "start_timestamp", None)
            if prev_start_ts is None:
                logger.debug(
                    "Previous compaction missing start_timestamp; skipping rolling compaction to prevent fragmentation."
                )
                return None
            start_timestamp = float(prev_start_ts)
            end_timestamp = float(getattr(new_slice_events[-1], "timestamp", 0.0))
            slice_turns_count = len(new_candidate_turns)
        else:
            slice_events = [e for _, turn_evts in candidate_turns for e in turn_evts]
            if not slice_events:
                return None
            conversation_history = summarizer._format_events_for_prompt(slice_events)
            start_timestamp = float(getattr(slice_events[0], "timestamp", 0.0))
            end_timestamp = float(getattr(slice_events[-1], "timestamp", 0.0))
            slice_turns_count = len(candidate_turns)

        template = getattr(
            summarizer, "_prompt_template", ZOCRATES_COMPACTION_PROMPT_TEMPLATE
        )
        prompt = template.format(conversation_history=conversation_history)
        return prompt, start_timestamp, end_timestamp, slice_turns_count, k
    except Exception as exc:
        logger.warning(
            "Adaptive context compaction slice preparation failed open: %s", exc
        )
        return None

_safe_apply_rewinds

_safe_apply_rewinds(events)

Safely apply rewinds to events list, failing open to original list if unneeded.

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

def _safe_apply_rewinds(events: list[Any]) -> list[Any]:
    """Safely apply rewinds to events list, failing open to original list if unneeded."""
    if not events:
        return []
    has_rewinds = any(
        isinstance(
            getattr(getattr(e, "actions", None), "rewind_before_invocation_id", None),
            str,
        )
        for e in events
    )
    if has_rewinds:
        try:
            return _apply_rewinds(events)
        except Exception as exc:
            logger.debug("Failed applying rewinds: %s", exc)
            return list(events)
    return list(events)

_safe_set

_safe_set(span, key, value)

Safely set an attribute on an active OpenTelemetry span, no-oping on error or None.

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

def _safe_set(span: Any, key: str, value: Any) -> None:
    """Safely set an attribute on an active OpenTelemetry span, no-oping on error or None."""
    if span is None:
        return
    try:
        if hasattr(span, "set_attribute"):
            if isinstance(value, (dict, list)):
                span.set_attribute(key, json.dumps(value, ensure_ascii=False))
            else:
                span.set_attribute(key, value)
        elif hasattr(span, "_attributes") and isinstance(span._attributes, dict):
            span._attributes[key] = value
    except Exception as exc:
        logger.debug("Failed setting span attribute %s: %s", key, exc)

_start_compaction_span

_start_compaction_span(name='compact_session_history')

Return a context manager yielding an OpenTelemetry span for Phoenix observability, or None on fallback.

Parameters:

Name	Type	Description	Default
name	str	Name of the OpenTelemetry span.	'compact_session_history'

Yields:

Type	Description
Any	Active span if tracing is available, otherwise None.

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

@contextlib.contextmanager
def _start_compaction_span(name: str = "compact_session_history") -> Iterator[Any]:
    """Return a context manager yielding an OpenTelemetry span for Phoenix observability, or None on fallback.

    Args:
        name: Name of the OpenTelemetry span.

    Yields:
        Active span if tracing is available, otherwise None.
    """
    if _otel_trace is None:
        yield None
        return
    span_cm = None
    try:
        tracer = _otel_trace.get_tracer("zocrates.compaction")
        span_cm = tracer.start_as_current_span(name)
    except Exception as exc:
        logger.debug("Failed starting compaction span: %s", exc)
        yield None
        return

    span = None
    if hasattr(span_cm, "__enter__") and hasattr(span_cm, "__exit__"):
        try:
            with span_cm as s:
                span = s
                yield span
        except Exception as exc:
            if span is not None and _StatusCode is not None:
                try:
                    span.set_status(_StatusCode.ERROR, str(exc))
                except Exception:
                    pass
            raise
    else:
        yield span_cm

_suppress_reasoning_effort

_suppress_reasoning_effort(llm)

Safely suppress reasoning/thinking effort on LLM instances without raising Pydantic field errors.

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

def _suppress_reasoning_effort(llm: Any) -> None:
    """Safely suppress reasoning/thinking effort on LLM instances without raising Pydantic field errors."""
    try:
        current_extra = getattr(llm, "extra_body", None)
        if isinstance(current_extra, dict):
            current_extra["reasoning"] = {"effort": "none"}
        else:
            current_extra = {"reasoning": {"effort": "none"}}
        try:
            llm.extra_body = current_extra
        except (ValueError, AttributeError):
            object.__setattr__(llm, "extra_body", current_extra)
    except Exception:
        pass

    try:
        if hasattr(llm, "_additional_args"):
            existing_extra = getattr(llm, "_additional_args", {}).get("extra_body") or {}
            if isinstance(existing_extra, dict):
                llm._additional_args["extra_body"] = {**existing_extra, "reasoning": {"effort": "none"}}
            else:
                llm._additional_args["extra_body"] = {"reasoning": {"effort": "none"}}
    except Exception:
        pass

_trim_to_completed_turns

_trim_to_completed_turns(events)

Ensure events to compact only contain fully completed turns.

In conversational agent workflows, a turn may have 2 events (chat), 3 events (tool call + tool response), or more. If raw event slicing leaves an incomplete turn at the tail (such as an unanswered user question or open tool call), drop that turn completely from the compaction slice so the summarizer prompt always terminates cleanly on an Assistant completion.

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

def _trim_to_completed_turns(events: list[Event]) -> list[Event]:
    """Ensure events to compact only contain fully completed turns.

    In conversational agent workflows, a turn may have 2 events (chat), 3 events (tool call
    + tool response), or more. If raw event slicing leaves an incomplete turn at the tail
    (such as an unanswered user question or open tool call), drop that turn completely
    from the compaction slice so the summarizer prompt always terminates cleanly on an
    Assistant completion.
    """
    if not events:
        return []

    invocations: dict[str, list[Event]] = {}
    inv_order: list[str] = []
    for e in events:
        inv_id = e.invocation_id or "unknown"
        if inv_id not in invocations:
            invocations[inv_id] = []
            inv_order.append(inv_id)
        invocations[inv_id].append(e)

    valid_inv_ids = []
    for inv_id in inv_order:
        inv_events = invocations[inv_id]
        has_assistant_completion = any(
            e.author != "user" and (
                (e.content and any(p.text for p in e.content.parts if not getattr(p, "thought", False)))
                or any(getattr(p, "function_response", None) for p in (e.content.parts if e.content else []))
            )
            for e in inv_events
        )
        if has_assistant_completion:
            valid_inv_ids.append(inv_id)

    result: list[Event] = []
    for inv_id in valid_inv_ids:
        result.extend(invocations[inv_id])
    return result

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

compact_session_adaptive async

compact_session_adaptive(session, cfg=None, summarizer=None)

Compact session history using adaptive retention slicing and rolling consolidation.

• Slices history into candidate turns covering Turns 1 .. N-K, retaining the latest K turns verbatim.
• If a previous compaction event exists: Consolidates the previous summary with new completed turns up to Turn N-K, preserving start_timestamp = previous_compaction.start_timestamp and setting end_timestamp = new_slice_events[-1].timestamp.
• If no previous compaction event exists: Summarizes all events from Turns 1 .. N-K spanning [slice_events[0].ts, slice_events[-1].ts].
• Wraps LLM invocation in OpenTelemetry span compact_session_history (kind=LLM) with full attributes.
• Fails open to None on any error or timeout without raising.

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

async def compact_session_adaptive(
    session: Any,
    cfg: dict[str, Any] | None = None,
    summarizer: Any = None,
) -> Event | None:
    """Compact session history using adaptive retention slicing and rolling consolidation.

    - Slices history into candidate turns covering Turns 1 .. N-K, retaining the latest K turns verbatim.
    - If a previous compaction event exists:
      Consolidates the previous summary with new completed turns up to Turn N-K,
      preserving start_timestamp = previous_compaction.start_timestamp and setting
      end_timestamp = new_slice_events[-1].timestamp.
    - If no previous compaction event exists:
      Summarizes all events from Turns 1 .. N-K spanning [slice_events[0].ts, slice_events[-1].ts].
    - Wraps LLM invocation in OpenTelemetry span `compact_session_history` (kind=LLM) with full attributes.
    - Fails open to None on any error or timeout without raising.
    """
    if cfg is None:
        try:
            cfg = load_zocrates_config()
        except Exception as exc:
            logger.debug("Failed to load zocrates config: %s", exc)
            cfg = {}

    if summarizer is None:
        try:
            from zocrates.agents.app import get_compaction_summarizer

            summarizer = get_compaction_summarizer()
        except Exception as exc:
            logger.debug("Failed to get compaction summarizer: %s", exc)
            return None

    if summarizer is None:
        return None

    llm = getattr(summarizer, "_llm", getattr(summarizer, "llm", None))
    if llm is None:
        return None

    raw_events = (
        getattr(session, "events", [])
        if hasattr(session, "events")
        else (session if isinstance(session, list) else [])
    )
    live_events = _safe_apply_rewinds(raw_events)

    # Exclude pipeline isolation scopes and existing compaction events
    unscoped_events = [
        e
        for e in live_events
        if getattr(e, "isolation_scope", None) is None
        and not _is_compaction_event(e)
    ]

    prep_result = _prepare_compaction_slice(
        unscoped_events, live_events, cfg, summarizer
    )
    if prep_result is None:
        return None

    prompt, start_timestamp, end_timestamp, slice_turns_count, k = prep_result

    # Suppress reasoning effort
    _suppress_reasoning_effort(llm)

    model_name = getattr(llm, "model", None) or cfg.get(
        "compaction_summarizer_model", DEFAULT_COMPACTION_SUMMARIZER_MODEL
    )

    with _start_compaction_span("compact_session_history") as span:
        _safe_set(span, "openinference.span.kind", "LLM")
        _safe_set(span, "llm.model_name", str(model_name))
        _safe_set(span, "gen_ai.request.model", str(model_name))
        _safe_set(
            span,
            "llm.invocation_parameters",
            {"temperature": 0.2, "reasoning": {"effort": "none"}},
        )
        _safe_set(span, "input.value", prompt)
        _safe_set(span, "input.mime_type", "text/plain")
        _safe_set(span, "compaction.summarize.start_timestamp", start_timestamp)
        _safe_set(span, "compaction.summarize.end_timestamp", end_timestamp)
        _safe_set(span, "compaction.summarize.slice_turns", slice_turns_count)
        _safe_set(span, "compaction.summarize.retained_turns", k)

        t0 = time.perf_counter()
        try:
            llm_request = LlmRequest(
                model=getattr(llm, "model", None),
                contents=[Content(role="user", parts=[Part(text=prompt)])],
            )
            summary_content = None
            summary_usage_metadata = None
            async for llm_response in 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

            latency_ms = (time.perf_counter() - t0) * 1000.0
            _safe_set(span, "compaction.summarize.latency_ms", round(latency_ms, 2))

            if summary_content is None or not summary_content.parts:
                _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()

            if not summary_text:
                _safe_set(span, "compaction.status", "empty_text")
                return None

            _safe_set(span, "output.value", summary_text)
            _safe_set(span, "output.mime_type", "text/plain")

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

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

            clean_parts = [
                Part.from_text(text=p.text)
                for p in summary_content.parts
                if getattr(p, "text", None) and not getattr(p, "thought", False)
            ]
            if not clean_parts:
                clean_parts = [Part.from_text(text=summary_text)]

            compacted_content = Content(role="model", parts=clean_parts)
            compaction = EventCompaction(
                start_timestamp=start_timestamp,
                end_timestamp=end_timestamp,
                compacted_content=compacted_content,
            )
            actions = EventActions(compaction=compaction)

            return Event(
                author="user",
                actions=actions,
                invocation_id=Event.new_id(),
                usage_metadata=summary_usage_metadata,
            )
        except Exception as exc:
            latency_ms = (time.perf_counter() - t0) * 1000.0
            _safe_set(span, "compaction.summarize.latency_ms", round(latency_ms, 2))
            _safe_set(span, "error.type", type(exc).__name__)
            _safe_set(span, "error.message", str(exc))
            if _StatusCode is not None and span is not None:
                span.set_status(_StatusCode.ERROR, str(exc))
            logger.warning("Adaptive context compaction LLM call failed open: %s", exc)
            return None

load_zocrates_config

load_zocrates_config(zotero_data_dir=None)

Load the Zocrates config from the data directory.

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

Returns config merged with standard defaults.

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path | str | None	Root Zotero data directory.	None

Returns:

Type	Description
dict	Parsed JSON config dict merged with defaults.

Source code in zocrates/config.py

def load_zocrates_config(zotero_data_dir: Path | str | None = None) -> dict:
    """Load the Zocrates config from the data directory.

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

    Returns config merged with standard defaults.

    Args:
        zotero_data_dir: Root Zotero data directory.

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

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

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

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

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

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

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

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

                # Backfill override_agent_models if missing or incomplete
                needs_save = False

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

run_adaptive_post_invocation_compaction async

run_adaptive_post_invocation_compaction(runner, session, cfg=None)

Execute post-turn adaptive compaction check and compaction event append.

Returns True if a compaction event was generated and appended to the session, False otherwise. Fails open defensively without raising exceptions.

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

async def run_adaptive_post_invocation_compaction(
    runner: Any,
    session: Any,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """Execute post-turn adaptive compaction check and compaction event append.

    Returns True if a compaction event was generated and appended to the session, False otherwise.
    Fails open defensively without raising exceptions.
    """
    if cfg is None:
        try:
            cfg = load_zocrates_config()
        except Exception as exc:
            logger.debug("Failed to load zocrates config: %s", exc)
            cfg = {}

    if not cfg.get("enable_adaptive_compaction", DEFAULT_ENABLE_ADAPTIVE_COMPACTION):
        return False

    if not session or not getattr(session, "events", None):
        return False

    last_ev = session.events[-1]
    # Check HITL / paused state: open function_call
    if last_ev.content and any(
        getattr(p, "function_call", None) for p in (last_ev.content.parts or [])
    ):
        logger.debug(
            "Skipping adaptive compaction: invocation interrupted mid-turn for HITL / open function call."
        )
        return False

    # Requested tool confirmation or auth config
    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 adaptive compaction: invocation awaiting tool confirmation."
        )
        return False

    try:
        should_compress, reason = await asyncio.to_thread(
            should_trigger_adaptive_compaction, session, cfg
        )
    except Exception as exc:
        logger.warning(
            "Error evaluating should_trigger_adaptive_compaction: %s. Failing open.",
            exc,
        )
        return False

    logger.debug(
        "Adaptive compaction decision check: %s (%s)", should_compress, reason
    )
    if not should_compress:
        return False

    try:
        compaction_event = await compact_session_adaptive(session, cfg)
    except Exception as exc:
        logger.warning(
            "Error generating adaptive compaction event: %s. Failing open.", exc
        )
        return False

    if compaction_event is None:
        return False

    session_service = getattr(runner, "session_service", None)
    if not session_service:
        logger.warning("Runner has no session_service; cannot append compaction event.")
        return False

    try:
        await session_service.append_event(session=session, event=compaction_event)
        logger.info(
            "Successfully appended adaptive compaction event to session %s",
            getattr(session, "id", getattr(session, "session_id", "unknown")),
        )
        return True
    except Exception as exc:
        logger.warning(
            "Defensively caught exception appending compaction event (e.g. StaleSessionError): %s",
            exc,
        )
        return False

should_trigger_adaptive_compaction

should_trigger_adaptive_compaction(session, cfg, api_key=None)

Evaluate whether adaptive compaction should trigger post-turn for the given session.

Enforces: 1. Global enable toggle (enable_adaptive_compaction). 2. Minimum completed turns invariant: effective_min_turns = max(configured_min_turns, k + 1). 3. Coordinator prompt token envelope (token floor and ceiling): - If tokens >= token_ceiling (default 100k): forces COMPRESS immediately without probing. - If tokens < token_floor (default 30k): skips compaction without probing. 4. Fast decision model rubric probe on state (Returns COMPRESS if p >= threshold, else CONTINUE). Fails open to CONTINUE on any probe error or timeout.

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

def should_trigger_adaptive_compaction(
    session: Any,
    cfg: dict[str, Any],
    api_key: str | None = None,
) -> tuple[bool, str]:
    """Evaluate whether adaptive compaction should trigger post-turn for the given session.

    Enforces:
    1. Global enable toggle (`enable_adaptive_compaction`).
    2. Minimum completed turns invariant: `effective_min_turns = max(configured_min_turns, k + 1)`.
    3. Coordinator prompt token envelope (token floor and ceiling):
       - If tokens >= token_ceiling (default 100k): forces COMPRESS immediately without probing.
       - If tokens < token_floor (default 30k): skips compaction without probing.
    4. Fast decision model rubric probe on state (Returns COMPRESS if p >= threshold, else CONTINUE).
    Fails open to CONTINUE on any probe error or timeout.
    """
    if not cfg.get("enable_adaptive_compaction", DEFAULT_ENABLE_ADAPTIVE_COMPACTION):
        return False, "adaptive_compaction_disabled"

    k = max(
        0,
        int(
            cfg.get(
                "adaptive_compaction_retention_turns",
                DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS,
            )
        ),
    )
    raw_min_turns = int(
        cfg.get(
            "adaptive_compaction_min_turns", DEFAULT_ADAPTIVE_COMPACTION_MIN_TURNS
        )
    )
    effective_min_turns = max(raw_min_turns, k + 1)
    if raw_min_turns <= k:
        logger.warning(
            "adaptive_compaction_min_turns (%d) must be strictly greater than "
            "retention_turns (%d). Disambiguating by clamping effective_min_turns to %d.",
            raw_min_turns,
            k,
            effective_min_turns,
        )

    completed_turns = count_completed_turns_since_last_compaction(session)
    if completed_turns < effective_min_turns:
        return False, f"turn_count_below_min ({completed_turns} < {effective_min_turns})"

    raw_events = (
        getattr(session, "events", [])
        if hasattr(session, "events")
        else (session if isinstance(session, list) else [])
    )
    live_events = _safe_apply_rewinds(raw_events)
    tokens = get_coordinator_prompt_tokens(live_events)
    token_ceiling = int(
        cfg.get(
            "adaptive_compaction_token_ceiling",
            DEFAULT_ADAPTIVE_COMPACTION_TOKEN_CEILING,
        )
    )
    token_floor = int(
        cfg.get(
            "adaptive_compaction_token_floor",
            DEFAULT_ADAPTIVE_COMPACTION_TOKEN_FLOOR,
        )
    )

    if tokens is not None and tokens >= token_ceiling:
        return True, f"token_ceiling_exceeded ({tokens} >= {token_ceiling})"

    if tokens is not None and tokens < token_floor:
        return False, f"tokens_below_floor ({tokens} < {token_floor})"

    state = extract_decision_state(session, k=k, since_last_compaction=True)
    if not state.get("earlier_discussion") or not state.get("latest_turn"):
        return False, "insufficient_dialogue_state"

    threshold = float(
        cfg.get("adaptive_compaction_threshold", DEFAULT_ADAPTIVE_COMPACTION_THRESHOLD)
    )
    metadata = {
        "completed_turns": completed_turns,
        "coordinator_prompt_tokens": tokens,
        "threshold": threshold,
    }
    session_id = getattr(session, "id", None) or getattr(session, "session_id", None)
    if session_id:
        metadata["session_id"] = str(session_id)

    prob = evaluate_adaptive_compaction_rubric(
        state, cfg, api_key=api_key, metadata=metadata
    )
    if prob is None:
        return False, "probe_failed_or_timed_out"

    if prob >= threshold:
        return True, f"decision_probe_triggered (p={prob:.3f} >= {threshold:.2f})"
    else:
        return False, f"decision_probe_declined (p={prob:.3f} < {threshold:.2f})"