Agents¶

def get_checkpointer(request: Request) -> BaseCheckpointSaver:
    """FastAPI dependency returning the app-wide checkpointer from ``app.state``.

    Set during ``lifespan`` startup (``AsyncPostgresSaver``). In tests the
    ``client`` fixture injects an ``InMemorySaver`` onto
    ``app.state.checkpointer``.
    """
    return request.app.state.checkpointer

@asynccontextmanager
async def open_checkpointer(conn_string: str) -> AsyncIterator[AsyncPostgresSaver]:
    """Open an ``AsyncPostgresSaver`` held for the whole app lifetime.

    ``AsyncPostgresSaver.from_conn_string`` is itself an async context manager
    that CLOSES its psycopg connection on ``__aexit__``, so we wrap it and keep
    it entered across the lifespan (the lifespan enters this CM manually and
    exits it on shutdown). ``setup()`` creates/migrates the LangGraph tables
    (idempotent).

    Args:
        conn_string: Raw psycopg URI (``postgresql://…`` — NOT the
            ``postgresql+psycopg://`` SQLAlchemy form), e.g.
            ``PostgresHandle.psycopg_url``.

    Yields a ready-to-use saver.
    """
    async with AsyncPostgresSaver.from_conn_string(conn_string) as saver:
        await saver.setup()  # creates/migrates LangGraph tables (idempotent)
        logger.info("Conversation checkpointer ready (PostgreSQL)")
        yield saver

def plan_turn(
    llm,
    *,
    question: str,
    retrieve: Callable[[], List[KbExcerpt]],
    custom_prompt: Optional[str] = None,
    starred_messages: Optional[List[str]] = None,
    web_search_enabled: bool = False,
    effort_section: Optional[str] = None,
) -> TurnPlan:
    """Decide the turn's mode and build the runner bundle.

    ``retrieve`` is only called in systematic mode: agentic mode defers
    retrieval to the model's tool call, and plain mode has no KB.

    ``effort_section`` is the graded reasoning instruction for this turn
    (1.1.2, ``src.agents.reasoning_effort``) or ``None`` when the level needs
    no instruction. Every mode takes it in the same slot -- after the tool
    regime, before the user's own instructions -- so the level means the same
    thing on a plain, a systematic and an agentic turn.

    ``web_search_enabled`` is the conversation's toggle (#310, copied from the
    global default at creation; arena passes the global setting directly). The
    web_search tool joins the turn iff the toggle is on AND the model is
    wire-capable — the SAME gate as agentic KB (#301): supports_tools AND
    supports_tools_wire is True. It rides plain turns (plain+web becomes a
    tool turn) and agentic-KB turns (both tools); the systematic path stays
    zero-tool, toggle or not — a non-wire model cannot execute tools, and the
    debug force-systematic flag deliberately keeps its zero-tool contract.
    The tool is ALWAYS exposed when gated in, online or not: offline turns get
    the tool's honest deterministic failure text instead of a missing tool.
    """
    # Deferred (#160): importing the tools module pulls the LangChain ``@tool``
    # machinery, needed on turns only — never at boot. Deterministic tools are
    # carried by KB turns only (the KB tool is added in agentic mode); plain
    # chat is zero-tool (#129) unless the web toggle turns it into a tool turn.
    from src.agents.tools import (
        TurnToolContext,
        calculator,
        search_knowledge_base,
        web_search,
    )
    from src.core import config

    base_tools = [calculator]

    # #298 agentic gate: the tri-state flag arbitrates, the per-model verified
    # wire capability decides in the default (flag None) state.
    flag = config.KB_AGENTIC_MODE
    supports_tools = bool(getattr(llm, "supports_tools", False))
    wire = getattr(llm, "supports_tools_wire", None)
    agentic = should_use_kb(llm) and (
        flag is True or (flag is None and supports_tools and wire is True)
    )

    # #310 web gate: strictly the verified wire capability — the KB tri-state
    # flag never bypasses it (a forced-agentic debug run on an unverified wire
    # still gets no web tool).
    wire_capable = supports_tools and wire is True
    web = web_search_enabled and wire_capable

    def _log_web(mode_allows_tools: bool) -> None:
        if not web_search_enabled:
            return
        if web and mode_allows_tools:
            logger.info(
                "Turn tools: web_search enabled "
                "(decided_by=web toggle on + verified wire capability)"
            )
        elif not wire_capable:
            logger.info(
                "Turn tools: web_search unavailable "
                "(model tool-call wire capability not verified)"
            )
        else:
            logger.info("Turn tools: web_search unavailable (systematic KB path is zero-tool)")

    if agentic:
        if flag is True:
            decided_by = "flag force-agentic (ERUDI_KB_AGENTIC=1)"
        else:
            decided_by = (
                "per-model verified wire capability "
                "(supports_tools=True, supports_tools_wire=True)"
            )
        budget = get_prompting_strategy(_param_size(llm))["kb_token_budget"]
        _log_web(mode_allows_tools=True)
        logger.info(
            f"Turn mode: agentic KB (kb_id={getattr(llm, 'kb_id', None)}, "
            f"decided_by={decided_by}, web_search={'on' if web else 'off'})"
        )
        return TurnPlan(
            system_prompt=build_kb_agentic_system_prompt(
                llm,
                custom_prompt=custom_prompt,
                starred_messages=starred_messages,
                web_search=web,
                effort_section=effort_section,
            ),
            tools=[*base_tools, search_knowledge_base, *([web_search] if web else [])],
            kb_context_block=None,
            kb_language_line="",
            context=TurnToolContext(
                kb_id=llm.kb_id,
                kb_token_budget=budget,
                web_token_budget=budget,
            ),
        )

    # Systematic: retrieve() encapsulates is_attached + tier + failure policy.
    excerpts = retrieve()
    if excerpts:
        # Name the gate that kept this KB turn off the agentic path (#298).
        if flag is False:
            decided_by = "flag force-systematic (ERUDI_KB_AGENTIC=0)"
        elif not supports_tools:
            decided_by = "model does not declare tool support"
        elif wire is None:
            decided_by = "tool-call wire capability unverified (NULL)"
        else:
            decided_by = "tool-call wire capability verified unreliable (False)"
        _log_web(mode_allows_tools=False)
        logger.info(
            f"Turn mode: systematic KB (kb_id={getattr(llm, 'kb_id', None)}, "
            f"excerpts={len(excerpts)}, decided_by={decided_by})"
        )
        return TurnPlan(
            system_prompt=build_kb_system_prompt(
                llm,
                custom_prompt=custom_prompt,
                starred_messages=starred_messages,
                effort_section=effort_section,
            ),
            # Zero tools on the systematic path (#288): the context is injected
            # directly, so the model only has to answer. Carrying the calculator
            # here bought nothing for document Q&A and made leak-prone models
            # (e.g. Llama 3.1 8B) wrap the answer in a raw calculator tool-call
            # JSON. Same #129 rationale that made plain chat zero-tool.
            tools=[],
            kb_context_block=build_kb_context_block(excerpts=excerpts, question=question),
            kb_language_line=answer_language_line(question),
            context=None,
        )

    if not getattr(llm, "is_attached_to_kb", False):
        plain_reason = "no KB attached"
    elif not should_use_kb(llm):
        plain_reason = "size tier disables KB context"
    else:
        plain_reason = "KB retrieval returned no excerpts"
    _log_web(mode_allows_tools=True)
    logger.info(
        f"Turn mode: plain (reason={plain_reason}, " f"web_search={'on' if web else 'off'})"
    )
    # Zero tools in plain mode (#129): the calculator was a demo tool whose
    # scaffolding cost hurt every model (and wrecked small ones) for no product
    # value outside KB turns. The web toggle is the ONE exception (#310):
    # a wire-capable model with the toggle on carries web_search alone.
    if web:
        budget = get_prompting_strategy(_param_size(llm))["kb_token_budget"]
        return TurnPlan(
            system_prompt=build_agent_system_prompt(
                llm,
                custom_prompt=custom_prompt,
                starred_messages=starred_messages,
                web_search=True,
                effort_section=effort_section,
            ),
            tools=[web_search],
            kb_context_block=None,
            kb_language_line="",
            context=TurnToolContext(web_token_budget=budget),
        )
    return TurnPlan(
        system_prompt=build_agent_system_prompt(
            llm,
            custom_prompt=custom_prompt,
            starred_messages=starred_messages,
            effort_section=effort_section,
        ),
        tools=[],
        kb_context_block=None,
        kb_language_line="",
        context=None,
    )

def should_use_kb(llm) -> bool:
    """True when the model has a KB attached AND its size tier allows context."""
    if not getattr(llm, "is_attached_to_kb", False):
        return False
    return get_prompting_strategy(_param_size(llm)).get("use_kb_context", False)

def detect_language(text: str) -> Optional[str]:
    """ISO 639-1 code of ``text``'s language, or None when unconfident."""
    text = (text or "").strip()
    if not text:
        return None
    identifier = _get_identifier()
    if identifier is None:
        return None
    try:
        language, probability = identifier.classify(text)
    except Exception:
        # Per-turn and best-effort: the generic line is the fallback, and the
        # text itself is already in the request's INFO records.
        logger.debug("Language detection failed for this turn", exc_info=True)
        return None
    return language if probability >= MIN_CONFIDENCE else None

def past_tool_result_ids(messages) -> frozenset:
    """Message ids of the tool results ``strip_stale_tool_results`` marks.

    Message ids, not tool-call ids: a parser with deterministic call ids
    (mlx_vlm's kimi_k2 parser emits ``functions.<name>:0``) reuses the same
    call id every turn, while message ids are unique once
    ``_ensure_message_ids`` ran. The compaction counter designates past
    results through this set, so it counts any list LangChain hands it --
    suffixes, the reversed pool of ``trim_messages``, partial copies -- as
    sent, without knowing where the list came from.
    """
    messages = list(messages)
    keep = last_human_index(messages)
    if keep is None:
        return frozenset()
    return frozenset(
        message.id
        for message in messages[:keep]
        if message.type == "tool"
        and message.id is not None
        and getattr(message, "name", None) in STALE_TOOL_RESULT_MARKERS
    )

def strip_stale_tool_results(messages, *, all_past: bool = False) -> list:
    """``messages`` as sent: past turns' KB/web results replaced by their markers.

    Pure (copies, never mutates). A result is past when it lies before the
    last user message -- the positional rule ``_StripStaleToolResults``
    applies to every request. ``all_past=True`` replaces every such result:
    the summarizer's pool holds only past turns, and the amber-warning
    projection judges the NEXT request, where the turn just answered is past.
    """
    messages = list(messages)
    if all_past:
        keep = len(messages)
    else:
        keep = last_human_index(messages)
        if keep is None:
            return messages
    for i, message in enumerate(messages[:keep]):
        messages[i] = stale_result_copy(message)
    return messages

def build_chat_model(
    llm,
    *,
    temperature: float,
    top_p: float,
    max_tokens: int,
    repetition_penalty: float = DEFAULT_REPETITION_PENALTY,
    repetition_context_size: int = DEFAULT_REPETITION_CONTEXT_SIZE,
    disable_thinking: bool = False,
    auto_output_budget: bool = True,
    sampling: Optional[SamplingDefaults] = None,
    effort_plan: Optional[EffortPlan] = None,
    kb_additions: str = "",
    preflight_retry: bool = True,
    record_usage: bool = False,
) -> ChatOpenAI:
    """Resolve the engine child for ``llm`` and wrap it as a ``ChatOpenAI``.

    SYNC and potentially slow: ``get_model_and_tokenizer`` spawns/probes the
    child server under the engine lock, so call this via ``run_in_threadpool``
    from async code (the agent runner does).

    The ``model`` field MUST go through the engine's ``_payload_model_value`` —
    mlx_vlm.server resolves it via ``get_cached_model(request.model)`` so MLX
    sends the real preloaded model path, while llama.cpp sends the alias;
    hardcoding either would break the other.

    Params are set on the constructor (NOT via ``.bind`` — LangChain v1 rejects
    pre-bound models passed to ``create_agent``).

    ``max_tokens`` is normally only the FALLBACK output budget: the client
    recomputes a real one per model call from the window it is running in
    (``src.agents.output_budget``). ``auto_output_budget=False`` turns that off
    for the callers whose budget is deliberate -- the one-shot title path and
    the capped compaction summary.

    ``kb_additions`` is the KB text this turn's requests carry (block, joins,
    language line): a request that carried it stamps its size, so the ratio
    it measured excludes a block the next request no longer holds.
    ``preflight_retry=False``
    turns off the single retry with a smaller cap after a context-check
    rejection (the summary client: a smaller cap would truncate the summary).

    Every client asks the server for usage (``stream_usage=True``): the last
    chunk then carries the real prompt size, which the client stamps with its
    own estimate of the request (``Erudi_Chat_OpenAI._astream``).
    ``record_usage=True`` (the conversation and Arena clients only) also
    pushes each call's usage to the engine handle the client was built
    against (``note_call_usage``), where the MLX engine measures what a turn
    really used; the summary and title clients push nothing.

    ``effort_plan`` carries the turn's reasoning effort (1.1.2). Its
    ``wire_effort`` rides the NATIVE ``reasoning_effort`` request field, not
    ``extra_body``: it is a first-class field of all three layers (the
    ``ChatOpenAI`` constructor, llama-server, mlx_vlm). ``None`` (no plan, or a
    level no native lever carries) leaves it unset, and langchain drops unset
    fields from the payload -- so the request body stays byte-identical to
    today's.
    """
    # Deferred (#160): langchain_openai only loads on the first turn, not at
    # boot -- the subclass that inherits from it is built on the same first call.
    chat_openai_class = erudi_chat_openai_class()

    engine = config.LLM_Engine
    handle, _tokenizer = engine.get_model_and_tokenizer(llm.id, llm.link)
    model_field = engine._payload_model_value(handle)

    # The ALLOCATED window of the child just resolved (llama-server: read from
    # /props after its fit; MLX: the --max-kv-size bound stamped at spawn, or
    # the memory bound the child was lowered to after load when memory binds).
    # Handed to the client so the first-chunk watchdog ceiling scales with the
    # window a full-window prompt can now legitimately fill -- without it, the
    # dynamic context window would recreate the #573 kill (a healthy long
    # prefill cut at the fixed 900 s ceiling). Fresh per turn: this factory
    # runs on every turn, after any model swap. None = window unknown.
    window_probe = getattr(engine, "effective_context_tokens", None)
    effective_window = window_probe() if callable(window_probe) else None

    # The MEMORY value for the SAME client: the ONE canonical working window
    # (min of the allocated window and the memory ceiling, over the known
    # candidates -- src.engines.working_window). Only the output budget reads
    # it; the first-chunk watchdog and preflight retry keep the raw allocated
    # window above. ``working_context_tokens`` calls ``MemoryBudget.from_engine``
    # (the loaded child's static facts are read from disk once and cached on
    # its handle), which is safe here because this factory runs in a
    # threadpool per turn.
    working_window = working_context_tokens(engine)
    # What a digit costs on this child's tokenizer, for the output budget (a
    # static fact read once per child from its tokenizer.json, MLX only).
    digit_tokens = budget_digit_tokens(engine)

    # Extra sampling params absent from the OpenAI wire schema. mlx_vlm.server reads
    # the HF names natively; llama.cpp engines translate them to their wire names
    # (repeat_penalty / repeat_last_n) via ``_translate_payload_kwargs``. Sent via
    # ChatOpenAI.extra_body so they land in the local server's chat-completions
    # body. (getattr keeps non-server engines / test stubs working via identity.)
    # The MLX translation also stamps a fresh random ``seed`` per request: this
    # factory runs once per turn, so every generation gets its own, which is what
    # makes temperature / top_p / top_k actually vary the output on Apple Silicon
    # (mlx_vlm.server replays a fixed default seed when the request has none).
    # #388: a resolved per-model profile supplies the repetition controls and,
    # ONLY when it defines them, top_k / min_p / presence_penalty. Without a
    # profile (or for a model without hints, whose profile is the fallback)
    # the body is byte-identical to the #129-validated one above.
    if sampling is not None:
        raw_kwargs = sampling.wire_kwargs()
    else:
        raw_kwargs = {
            "repetition_penalty": repetition_penalty,
            "repetition_context_size": repetition_context_size,
        }
    # Suppress reasoning at the chat-template level (#266): one-shot utility
    # calls (e.g. conversation titles) run on a ~12-token budget that a thinking
    # model would burn entirely inside <think>. mlx_vlm.server reads the
    # per-request ``enable_thinking`` field natively (it overrides the server
    # default); llama.cpp engines translate it to ``chat_template_kwargs`` in
    # their ``_translate_payload_kwargs``. Chat paths never pass
    # ``disable_thinking``, so their request body stays byte-identical to today.
    if disable_thinking:
        raw_kwargs["enable_thinking"] = False
    translate = getattr(engine, "_translate_payload_kwargs", lambda kw: kw)
    extra_body = translate(raw_kwargs)

    # A stream that does not end normally flags the child it ran against
    # (``Erudi_Chat_OpenAI.abandon_hook``; the MLX engine then sends a barrier
    # before its next prefix-cache reset). Bound NOW to this handle, never
    # re-read from the engine later: a stream finalized after a model swap
    # must flag its own child, not whichever child is loaded by then.
    note_abandoned = getattr(engine, "note_stream_abandoned", None)
    abandon_hook = functools.partial(note_abandoned, handle) if callable(note_abandoned) else None
    # Same binding for the usage a call reports: it belongs to the turn's
    # measurement window on THIS child.
    note_usage = getattr(engine, "note_call_usage", None)
    usage_hook = (
        functools.partial(note_usage, handle) if record_usage and callable(note_usage) else None
    )
    note_start = getattr(engine, "note_call_start", None)
    call_start_hook = (
        functools.partial(note_start, handle) if record_usage and callable(note_start) else None
    )

    # Log the extra_body AS SENT (post-translation, so llama.cpp's wire names
    # show up), one key=value per entry on the same line: the optional profile
    # keys (top_k / min_p / presence_penalty, #388) are otherwise invisible in
    # the INFO log and a QA pass cannot confirm they reached the server.
    extra_body_desc = ", ".join(f"{key}={value}" for key, value in extra_body.items())
    wire_effort = effort_plan.wire_effort if effort_plan is not None else None
    logger.info(
        f"ChatOpenAI built: model={model_field}, base_url={handle['base_url']}/v1, "
        f"temperature={temperature}, top_p={top_p}, max_tokens={max_tokens}, "
        f"reasoning_effort={wire_effort or 'unset'}, "
        f"extra_body=[{extra_body_desc}]"
    )
    return chat_openai_class(
        base_url=f"{handle['base_url']}/v1",
        # Every inference child (llama-server, mlx_vlm.server) is spawned with a
        # per-spawn `--api-key` so nothing else on the loopback interface can
        # drive it, and the handle is the only place that key lives. A handle
        # without one keeps the literal: an empty api_key would make the OpenAI
        # client fall back to reading OPENAI_API_KEY from the environment.
        # Deliberately absent from the log line above.
        api_key=handle.get("api_key") or "not-needed",
        model=model_field,
        temperature=temperature,
        top_p=top_p,
        max_tokens=max_tokens,
        # Native field of all three layers (1.1.2): llama-server reads "none"
        # as enable_thinking=false and forwards any other value to the chat
        # template; mlx_vlm normalizes it the same way. None = unset, dropped
        # from the payload.
        reasoning_effort=wire_effort,
        extra_body=extra_body,  # restore small-model coherence (repetition controls)
        timeout=None,  # cold model load can stall several seconds before first token
        max_retries=0,  # don't silently double-submit a slow local generation
        # langchain's ONE uniform per-chunk budget (120 s) cannot tell a long
        # prefill from a hang and killed healthy long-history turns (#573).
        # Off here; ``Erudi_Chat_OpenAI._astream`` enforces the two budgets that
        # replace it.
        stream_chunk_timeout=None,
        effective_context_tokens=effective_window,
        working_context_tokens=working_window,
        auto_output_budget=auto_output_budget,
        kb_additions=kb_additions or "",
        digit_tokens=digit_tokens,
        preflight_retry=preflight_retry,
        abandon_hook=abandon_hook,
        usage_hook=usage_hook,
        call_start_hook=call_start_hook,
        streaming=True,
        # ``stream_options.include_usage``: both local servers then end the
        # stream with a usage chunk (mlx_vlm 0.6.17 and llama-server alike);
        # a server that sends none leaves every consumer on its estimate.
        stream_usage=True,
    )

def answer_language_line(question: str) -> str:
    """The dynamic answer-language request, localized to the question's
    language (generic fallback when detection is unconfident). Appended
    by the runner's middleware AFTER the question — the one spot this
    model demonstrably honors (eval T5: in-question language requests
    are followed and even persist to the next turn)."""
    return _LANGUAGE_LINES.get(detect_language(question), _GENERIC_LANGUAGE_LINE)

def build_agent_system_prompt(
    llm,
    *,
    starred_messages: Optional[List[str]] = None,
    custom_prompt: Optional[str] = None,
    web_search: bool = False,
    effort_section: Optional[str] = None,
) -> str:
    """Build the size-adaptive system prompt for ``llm`` as a real ``SystemMessage``.

    The old hand-rolled flow merged the system text into the first user message
    (some local models lack a system role); the OpenAI-compatible servers handle
    a proper system message per the model's chat template, so we pass it as-is.

    ``web_search=True`` (#310) appends the scoped web section — the plain path
    without it stays byte-identical (regression-pinned). ``effort_section``
    (1.1.2) takes the same slot on the zero-tool concat path as it does in
    ``_compose_prompt``: after the regime, before the user's instructions.
    """
    strategy = _tier_strategy(llm)

    if web_search:
        size_category = strategy["system_prompt_size_category"]
        return _compose_prompt(
            llm,
            [_web_section(size_category, with_kb=False)],
            size_category=size_category,
            custom_prompt=custom_prompt,
            starred_messages=starred_messages,
            effort_section=effort_section,
        )

    sys_prompt = build_system_prompt(
        model_name=llm.name,
        size_category=strategy["system_prompt_size_category"],
        starred_messages=starred_messages or None,
    )

    if effort_section:
        sys_prompt += f"\n\n{effort_section}"

    if custom_prompt and custom_prompt.strip():
        sys_prompt += f"\nAdditional instructions: {custom_prompt.strip()}"

    return sys_prompt

def build_kb_agentic_system_prompt(
    llm,
    *,
    custom_prompt: Optional[str] = None,
    starred_messages: Optional[List[str]] = None,
    web_search: bool = False,
    effort_section: Optional[str] = None,
) -> str:
    """SYSTEM prompt for a TOOL-CALLING KB assistant (issues #84 / #129).

    Composes the size-tier persona with an appended KB-tool section whose
    trigger is SCOPED to document-related questions (the broad "for ANY
    question" imperative over-triggered on everyday questions) while keeping
    the #84 discipline: never claim the information is absent before a
    search has come back empty. Tiny/small tiers get a compact variant.
    When the model searches, the tool result carries a localized language
    line in tail position (``format_kb_tool_result``); otherwise the tier
    persona's "user's language" line applies.
    """
    strategy = _tier_strategy(llm)
    size_category = strategy["system_prompt_size_category"]
    kb_section = (
        _KB_AGENTIC_SECTION_COMPACT
        if size_category in _COMPACT_KB_TIERS
        else _KB_AGENTIC_SECTION_FULL
    )
    sections = [kb_section]
    if web_search:
        # KB section first, then the web section with the arbitration clause
        # (#310): document questions go to the documents, fresh facts to the
        # web — the two triggers must never compete.
        sections.append(_web_section(size_category, with_kb=True))
    return _compose_prompt(
        llm,
        sections,
        size_category=size_category,
        custom_prompt=custom_prompt,
        starred_messages=starred_messages,
        effort_section=effort_section,
    )

def build_kb_context_block(*, excerpts: List["KbExcerpt"], question: str) -> str:
    """Per-turn KB block: attributed excerpts + grounding reminder, with
    the scaffolding LOCALIZED to the question's language.

    The runner's middleware merges it into the model request's LAST user
    message (request-time only — never persisted): on small local models,
    instructions dissolve with turn depth when they live in the system
    prompt, while the tail of the last user message stays in the effective
    window. The answer-language line is NOT here: it goes AFTER the
    question (``answer_language_line``) — run-4 eval showed the model
    treats pre-question lines as document-block metadata and ignores
    them, while user-voiced post-question requests are followed.
    """
    scaffold = _SCAFFOLDS.get(detect_language(question), _SCAFFOLDS["en"])
    blocks = "\n\n".join(
        f"[Document: {excerpt.source_file}]\n{excerpt.text}" for excerpt in excerpts
    )
    return f"{scaffold['header']}\n\n{blocks}\n\n{scaffold['reminder']}"

def build_kb_system_prompt(
    llm,
    *,
    custom_prompt: Optional[str] = None,
    starred_messages: Optional[List[str]] = None,
    effort_section: Optional[str] = None,
) -> str:
    """SYSTEM prompt for a KB assistant on the systematic path (no tools).

    Composes the size-tier persona with the appended excerpts contract
    (#129) — the persona base stays byte-identical to the plain path, so
    attaching a KB no longer turns the assistant into a refusal-prone
    "document analyst" for everyday questions.
    """
    strategy = _tier_strategy(llm)
    return _compose_prompt(
        llm,
        [_KB_SYSTEMATIC_SECTION],
        size_category=strategy["system_prompt_size_category"],
        custom_prompt=custom_prompt,
        starred_messages=starred_messages,
        effort_section=effort_section,
    )

def compaction_cutoff(
    messages: list,
    working_window: Optional[int],
    counter: Any,
    *,
    pct: int,
    overhead: int = 0,
    allocated_window: Optional[int] = None,
) -> int:
    """Index of the first message a compaction keeps (0: nothing to compact).

    THE cutoff rule, pure: the summarization middleware uses it, and so does
    the amber-warning projection (which has no middleware instance).
    ``counter`` counts real tokens as sent; ``overhead`` is O, what the
    request carries beyond the state (system prompt, KB block, tool schemas),
    already scaled. ``pct`` is the conversation's compaction threshold: the
    keep budget and the futility guard read the trigger through it.

    1. With a known window, the token cutoff alone: the earliest suffix that
       fits ``max(1, keep_token_budget(W, pct) - O)`` tokens, however many
       messages that is (token pressure is the only trigger there, so a
       count bound is not needed). Without a window, the message cutoff
       alone: the last ``SUMMARY_KEEP_MESSAGES`` messages, the fallback
       paired with the ``SUMMARY_TRIGGER_MESSAGES`` trigger.
    2. Never past the LAST user message: the question the current turn
       answers is never summarized away mid-turn (a tool round after it can
       be large on its own).
    3. Never ON a user message: LangChain inserts the summary as a user
       message, and two user messages in a row make the strict chat
       templates (Gemma, Mistral) reject every later turn, so the kept tail
       always starts with an assistant message and the state alternates
       ``[summary, answer, question, ...]``. On an OLDER user message the
       cutoff moves FORWARD to the next assistant message (keeping less, so
       the budget holds); only on the current question -- or when no
       assistant message lies between the cutoff and it -- does it move BACK
       to the answer before it.
    4. Tool-pair safe: an AI message is never separated from its tool
       results. A token cut on a tool result of an EARLIER turn (before the
       last user message) moves FORWARD past the round's results, so the
       call and its results are summarized together and the budget holds --
       mirroring step 3's forward move; landing on an older user message,
       step 3 then applies. A cut on a tool result AFTER the current
       question (mid-turn) has nothing to move forward to and moves back to
       its call (LangChain's ``_find_safe_cutoff_point``), as does the
       message cutoff without a window and any move back of step 3.
    5. If only the previous summary would be summarized (cutoff <= 1 after a
       summary), nothing is: re-summarizing a summary alone adds nothing and
       would reset the prefix cache for nothing.
    6. Futility guard -- only with a known window, where the token trigger is
       the only trigger, and only when it fired (``count + O >=
       trigger_tokens(W, pct)``; the amber-warning projection also asks below
       it): a compaction whose gain ``count(all) - count(kept) -
       summary_cap(W)`` is below ``POST_COMPACTION_MARGIN_TOKENS`` (256, the
       keep margin's floor in a different role: the smallest gain worth a
       summary call, whatever the threshold) -- nothing at all included --
       is skipped unless the request would overflow without it: against the
       allocated window (``O + count + 512 > W_alloc``, the output budget's
       floor) or the working window (``O + count > W``). A small gain is not
       worth a summary call and a prefix-cache reset; any compaction that may
       save an overflowing turn is taken (the gain is reckoned with the
       summary's CAP, the real summary is often shorter).

    Termination: every compaction strictly lowers the number of messages
    before the last user message; steps 2-3 bound the cutoff at
    the answer before the current question, and there step 5 returns 0 on
    the next hop -- the kept tail can stay above the trigger, so termination
    rests on step 5, not on a margin under the trigger.

    With ordinary messages the kept tail stays inside the token budget. It
    exceeds it only by the answer before the current question when the cut
    falls ON that question (steps 2-3), by a last message larger than the
    budget on its own (kept whole, summarized on a later turn, never
    truncated), and by a tool round kept with its call when it lies after
    the current question (mid-turn) or right before it with no answer in
    between (step 3 moves back onto it) -- accepted: there the alternatives
    drop the question or split the round, which breaks the conversation.
    """
    from langchain.agents.middleware import SummarizationMiddleware
    from langchain_core.messages import HumanMessage, ToolMessage

    safe_point = SummarizationMiddleware._find_safe_cutoff_point
    last_human = last_human_index(messages)
    if _known_window(working_window):
        keep_budget = max(1, keep_token_budget(working_window, pct) - overhead)
        cutoff = _token_cutoff(messages, keep_budget, counter)
        if (
            last_human is not None
            and cutoff < last_human
            and isinstance(messages[cutoff], ToolMessage)
        ):
            # A tool round of an EARLIER turn: move FORWARD past its results,
            # so the call and its results are summarized together. Moving
            # back to the call would keep the whole round, over the budget.
            while isinstance(messages[cutoff], ToolMessage):
                cutoff += 1
        else:
            cutoff = safe_point(messages, cutoff)
    elif len(messages) <= SUMMARY_KEEP_MESSAGES:
        cutoff = 0
    else:
        cutoff = safe_point(messages, len(messages) - SUMMARY_KEEP_MESSAGES)
    if cutoff <= 0:
        return 0
    if last_human is not None:
        cutoff = min(cutoff, last_human)
    if (
        last_human is not None
        and cutoff < last_human
        and isinstance(messages[cutoff], HumanMessage)
    ):
        # An OLDER user message: move FORWARD to the next assistant message
        # (tool results after a call are fine, they follow it). Moving back
        # would keep the answer before it -- one that did not fit the budget.
        forward = cutoff
        while forward <= last_human and isinstance(messages[forward], (HumanMessage, ToolMessage)):
            forward += 1
        if forward <= last_human:
            cutoff = forward
    # Only on the current question (or with no assistant message between the
    # cutoff and it): every step moves strictly back, so the loop ends.
    while cutoff > 0:
        message = messages[cutoff]
        if isinstance(message, HumanMessage):
            cutoff -= 1
        elif isinstance(message, ToolMessage):
            paired = safe_point(messages, cutoff)
            # LangChain answers an orphaned tool result (no matching call)
            # by moving FORWARD, which could land back on a user message:
            # step back past it instead.
            cutoff = paired if paired < cutoff else cutoff - 1
        else:
            break
    if cutoff <= 1 and messages and _is_previous_summary(messages[0]):
        return 0
    if cutoff > 0 and _known_window(working_window):
        total = counter(messages)
        if total + overhead >= trigger_tokens(working_window, pct):
            gain = total - counter(messages[cutoff:]) - summary_cap(working_window)
            overflows = overhead + total > working_window or (
                _known_window(allocated_window)
                and overhead + total + OUTPUT_BUDGET_FLOOR_TOKENS > allocated_window
            )
            # The keep margin's 256-token floor, in another role: the
            # smallest gain worth a summary call. It does not scale with the
            # threshold.
            if gain < POST_COMPACTION_MARGIN_TOKENS and not overflows:
                return 0
    return cutoff

def compaction_reason(
    engine: Any, working_window: Optional[int], allocated_window: Optional[int]
) -> str:
    """``memory`` when the machine's memory bounds the window the compaction
    ran against, ``length`` otherwise.

    Memory bounds it when the engine says so (``memory_bound_tokens``: the CPU
    and CUDA engines size their window from the memory budget at spawn, MLX
    lowers its child's to the memory ceiling -- the criterion the amber
    warning and the overflow copy use), or when the memory ceiling folded the
    working window below the allocated one -- or stood alone, the allocated
    window unknown (the compact-as-early-as-possible clamp). The
    message-count fallback, with no window at all, is ``length``.
    """
    if _memory_bound_of(engine) is not None:
        return COMPACTION_REASON_MEMORY
    if _known_window(working_window) and (
        not _known_window(allocated_window) or working_window < allocated_window
    ):
        return COMPACTION_REASON_MEMORY
    return COMPACTION_REASON_LENGTH