Function _checkpoint_history

uncommon_route/composition.py:472–547

(
    messages: list[dict[str, Any]],
    artifact_store: ArtifactStore,
    semantic_compressor: SemanticCompressor,
    policy: CompositionPolicy,
    *,
    session_id: str,
    request: Any,
    step_type: str,
    is_agentic: bool,
)

async def _checkpoint_history(
    messages: list[dict[str, Any]],
    artifact_store: ArtifactStore,
    semantic_compressor: SemanticCompressor,
    policy: CompositionPolicy,
    *,
    session_id: str,
    request: Any,
    step_type: str,
    is_agentic: bool,
) -> tuple[list[dict[str, Any]], str, str, SemanticCallResult | None] | None:
    if step_type == "tool-selection" and policy.checkpoint_skip_tool_selection:
        return None

    keep_last_messages = (
        policy.checkpoint_agentic_keep_last_messages
        if is_agentic
        else policy.checkpoint_keep_last_messages
    )
    threshold_tokens = (
        policy.checkpoint_agentic_threshold_tokens
        if is_agentic
        else policy.checkpoint_threshold_tokens
    )

    if len(messages) < max(policy.checkpoint_min_messages, keep_last_messages + 2):
        return None
    if _estimate_messages_tokens(messages) < threshold_tokens:
        return None

    leading_system, remainder = _split_leading_system(messages)
    if len(remainder) <= keep_last_messages:
        return None
    if (
        is_agentic
        and policy.checkpoint_skip_recent_tool_window > 0
        and _has_recent_tool_activity(remainder[-policy.checkpoint_skip_recent_tool_window:])
    ):
        return None

    head = remainder[:-keep_last_messages]
    tail = remainder[-keep_last_messages:]
    transcript = _messages_to_transcript(head)
    if not transcript.strip():
        return None

    history_record = artifact_store.store_text(
        transcript,
        kind="history-block",
        role="system",
        session_id=session_id,
        content_type="text/plain",
    )
    summary = history_record.summary
    call: SemanticCallResult | None = None
    if not summary:
        call = await semantic_compressor.summarize_history(
            transcript,