wehub-resource-sync
926 lines
34 KiB
Python
926 lines
34 KiB
Python
"""Layered conversation history compaction for LLM context management.
|
|
|
|
Compaction fires when the estimated prompt token count approaches the
|
|
model's context window. Three layers are applied in order, from
|
|
least-lossy to most-lossy:
|
|
|
|
1. Surgical clearing — tool result bodies and binary content blocks
|
|
outside the recent window are replaced with markers.
|
|
2. LLM summarization — a @step LLM call summarises all messages
|
|
outside the recent window into a single summary pair.
|
|
3. Truncation — oldest messages are dropped when layers 1+2
|
|
are still insufficient (emergency fallback).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import tiktoken
|
|
|
|
from omnigent.entities import (
|
|
CompactionData,
|
|
ConversationItem,
|
|
MessageData,
|
|
)
|
|
from omnigent.llms.summarize import (
|
|
build_summarization_input,
|
|
build_summarization_prompt,
|
|
extract_summary_text,
|
|
)
|
|
from omnigent.spec.types import CompactionConfig
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
# Marker written into cleared tool result bodies.
|
|
_TOOL_RESULT_CLEARED = "[Previous tool result cleared — re-call tool if needed]"
|
|
|
|
# Marker written into cleared binary content block payloads.
|
|
_BINARY_CONTENT_CLEARED = (
|
|
"[binary content removed for context management — use file_id to retrieve]"
|
|
)
|
|
|
|
# Default compaction settings when AgentSpec.compaction is None.
|
|
_DEFAULT_TRIGGER_THRESHOLD: float = 0.8
|
|
_DEFAULT_RECENT_WINDOW: int = 5
|
|
|
|
|
|
@dataclass
|
|
class SummaryMetadata:
|
|
"""
|
|
Metadata from a Layer 2 summarization, passed from
|
|
:func:`compact` to the workflow's end-of-execution
|
|
persistence step.
|
|
|
|
:param text: The LLM-generated summary text.
|
|
:param last_item_id: The ID of the last conversation item
|
|
covered by this summary, e.g. ``"msg_abc123"``.
|
|
:param model: The model used for summarization, e.g.
|
|
``"openai/gpt-4o"``.
|
|
:param token_count: Approximate token count of the summary
|
|
text, e.g. ``342``.
|
|
"""
|
|
|
|
text: str
|
|
last_item_id: str
|
|
model: str
|
|
token_count: int
|
|
|
|
|
|
@dataclass
|
|
class CompactionResult:
|
|
"""
|
|
Result of running :func:`compact` on a messages list.
|
|
|
|
:param messages: The compacted messages list, ready to pass
|
|
to the LLM.
|
|
:param summary_metadata: Present only when Layer 2
|
|
(summarization) was triggered. Contains the summary text
|
|
and the ``last_item_id`` of the last item covered.
|
|
``None`` when only Layer 1 or Layer 3 applied, or when
|
|
summarization failed and Layer 3 was used as fallback.
|
|
:param total_tokens: Tiktoken estimate of the post-compaction
|
|
token count for ``messages``, e.g. ``8421``. Populated by
|
|
:func:`compact` using the count it already computed during
|
|
budget checks — no additional tokenization pass needed.
|
|
``None`` when the count is unavailable (e.g. early-return
|
|
paths that skip counting).
|
|
"""
|
|
|
|
messages: list[dict[str, Any]]
|
|
summary_metadata: SummaryMetadata | None
|
|
total_tokens: int | None = None
|
|
|
|
|
|
@dataclass
|
|
class _CompactionState:
|
|
"""
|
|
Per-execution compaction state maintained in the agent loop.
|
|
|
|
:param context_window: Cached context window size discovered
|
|
from the first ContextWindowExceededError,
|
|
e.g. ``128000``. ``None`` until the first overflow occurs.
|
|
:param last_summary: Metadata from the most recent Layer 2
|
|
summarization during this execution, or ``None`` if no
|
|
summarization has occurred yet.
|
|
:param config: The compaction config from the agent spec, or
|
|
``None`` to use defaults.
|
|
:param model: The LLM model string used for tiktoken estimation,
|
|
e.g. ``"openai/gpt-4o"``.
|
|
:param connection: Per-provider connection overrides (api_key,
|
|
base_url, etc.) for the summarization LLM call. ``None``
|
|
means use environment variable defaults.
|
|
:param conversation_id: Conversation id, e.g.
|
|
``"conv_0123456789abcdef"``. Used to look up the runner
|
|
client from the router so Layer 2 summarization runs through
|
|
the runner's credentials rather than the Omnigent server's.
|
|
"""
|
|
|
|
context_window: int | None
|
|
last_summary: SummaryMetadata | None
|
|
config: CompactionConfig | None
|
|
model: str
|
|
connection: dict[str, str] | None = None
|
|
conversation_id: str | None = None
|
|
post_compaction_tokens: int | None = None
|
|
history_len_at_compaction: int | None = None
|
|
|
|
|
|
def count_tokens(messages: list[dict[str, Any]], model: str) -> int:
|
|
"""
|
|
Estimate the token count for a messages list using tiktoken.
|
|
|
|
Used as a sanity check against provider-reported token counts
|
|
(within ~30%) and for proactive threshold checks. Not used as
|
|
the authoritative count — tiktoken is ~85-95% accurate for
|
|
non-OpenAI models, and the 20% headroom from
|
|
``trigger_threshold`` absorbs the difference.
|
|
|
|
:param messages: The messages list to count tokens for.
|
|
:param model: The LLM model string, e.g. ``"openai/gpt-4o"``.
|
|
Used to select the appropriate tiktoken encoding; falls
|
|
back to ``cl100k_base`` for unknown models.
|
|
:returns: Approximate token count for the serialised messages.
|
|
"""
|
|
# Strip provider prefix (e.g. "openai/gpt-4o" -> "gpt-4o")
|
|
# so tiktoken can look up the model encoding.
|
|
bare_model = model.split("/", 1)[-1] if "/" in model else model
|
|
try:
|
|
enc = tiktoken.encoding_for_model(bare_model)
|
|
except KeyError:
|
|
# Unknown model — fall back to the most common encoding.
|
|
enc = tiktoken.get_encoding("cl100k_base")
|
|
text = json.dumps(messages, ensure_ascii=False)
|
|
return len(enc.encode(text))
|
|
|
|
|
|
def _find_recent_boundary(
|
|
history: list[ConversationItem],
|
|
recent_window: int,
|
|
) -> int:
|
|
"""
|
|
Find the index in *history* where the recent window begins.
|
|
|
|
The recent window covers the last *recent_window* LLM response
|
|
groups. One group = one assistant message or one function_call
|
|
item (both mark an LLM response boundary). Items at or after
|
|
the returned index are protected from compaction.
|
|
|
|
:param history: The full conversation history list.
|
|
:param recent_window: Number of LLM response groups to protect,
|
|
e.g. ``5``.
|
|
:returns: The index of the first item inside the recent window.
|
|
Returns ``0`` if the history has fewer groups than the window
|
|
size (protect everything).
|
|
"""
|
|
if recent_window <= 0:
|
|
return len(history)
|
|
groups_seen = 0
|
|
for i in range(len(history) - 1, -1, -1):
|
|
item = history[i]
|
|
is_assistant_msg = (
|
|
item.type == "message"
|
|
and isinstance(item.data, MessageData)
|
|
and item.data.role == "assistant"
|
|
)
|
|
is_function_call = item.type == "function_call"
|
|
if is_assistant_msg or is_function_call:
|
|
groups_seen += 1
|
|
if groups_seen >= recent_window:
|
|
return i
|
|
return 0
|
|
|
|
|
|
def _clear_tool_results(
|
|
messages: list[dict[str, Any]],
|
|
protect_from: int,
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Replace tool result bodies outside the recent window with a
|
|
clearing marker.
|
|
|
|
The function_call / function_call_output pair structure is
|
|
preserved — no orphaned tool calls are created. Only the
|
|
``output`` field of ``function_call_output`` items is
|
|
replaced.
|
|
|
|
:param messages: The messages list to process (modified in place).
|
|
:param protect_from: Index of the first message in the recent
|
|
window. Messages at indices < *protect_from* are eligible
|
|
for clearing.
|
|
:returns: The same list (modified in place) for convenience.
|
|
"""
|
|
for i, msg in enumerate(messages):
|
|
if i >= protect_from:
|
|
break
|
|
if msg.get("type") == "function_call_output":
|
|
msg["output"] = _TOOL_RESULT_CLEARED
|
|
return messages
|
|
|
|
|
|
def _clear_binary_content(
|
|
messages: list[dict[str, Any]],
|
|
protect_from: int,
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Replace binary payload data in image/file content blocks
|
|
outside the recent window with a clearing marker.
|
|
|
|
The ``file_id`` is preserved so the agent can re-fetch the
|
|
content if needed. Text content blocks within the same message
|
|
are untouched.
|
|
|
|
:param messages: The messages list to process (modified in place).
|
|
:param protect_from: Index of the first message in the recent
|
|
window. Messages at indices < *protect_from* are eligible
|
|
for clearing.
|
|
:returns: The same list (modified in place) for convenience.
|
|
"""
|
|
for i, msg in enumerate(messages):
|
|
if i >= protect_from:
|
|
break
|
|
content = msg.get("content")
|
|
if not isinstance(content, list):
|
|
continue
|
|
for block in content:
|
|
if (
|
|
isinstance(block, dict)
|
|
and block.get("type") in ("image", "file")
|
|
and "data" in block
|
|
):
|
|
block["data"] = _BINARY_CONTENT_CLEARED
|
|
return messages
|
|
|
|
|
|
def _strip_output_annotations(
|
|
messages: list[dict[str, Any]],
|
|
protect_from: int,
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Remove ``annotations`` from ``output_text`` blocks outside
|
|
the recent window.
|
|
|
|
Annotations (e.g. ``file_citation``) are output metadata for
|
|
the client, not content the summarization LLM should see.
|
|
Stripping them before Layer 2 keeps the summarization input
|
|
clean and reduces token waste.
|
|
|
|
:param messages: The messages list to process (modified in place).
|
|
:param protect_from: Index of the first message in the recent
|
|
window. Messages at indices < *protect_from* are eligible
|
|
for stripping.
|
|
:returns: The same list (modified in place) for convenience.
|
|
"""
|
|
for i, msg in enumerate(messages):
|
|
if i >= protect_from:
|
|
break
|
|
content = msg.get("content")
|
|
if not isinstance(content, list):
|
|
continue
|
|
for block in content:
|
|
if (
|
|
isinstance(block, dict)
|
|
and block.get("type") == "output_text"
|
|
and "annotations" in block
|
|
):
|
|
del block["annotations"]
|
|
return messages
|
|
|
|
|
|
def _truncate_oldest(
|
|
messages: list[dict[str, Any]],
|
|
budget: int,
|
|
model: str,
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Emergency Layer 3: drop oldest messages until the token count
|
|
fits within *budget*.
|
|
|
|
Preserves tool call pair integrity — never drops a
|
|
``function_call`` without also dropping its matching
|
|
``function_call_output``, and vice versa. Drops from the front
|
|
of the list.
|
|
|
|
:param messages: The messages list to truncate.
|
|
:param budget: Maximum token count for the returned list,
|
|
e.g. ``102400``.
|
|
:param model: LLM model string for token counting.
|
|
:returns: A new messages list with oldest items dropped.
|
|
"""
|
|
result = list(messages)
|
|
while result and count_tokens(result, model) > budget:
|
|
drop_count = _pair_aware_drop_count(result)
|
|
if drop_count == 0:
|
|
break
|
|
result = result[drop_count:]
|
|
return result
|
|
|
|
|
|
def _pair_aware_drop_count(messages: list[dict[str, Any]]) -> int:
|
|
"""
|
|
Return how many items to drop from the front to avoid
|
|
orphaning a tool call pair.
|
|
|
|
If the first item is a ``function_call`` and the second is its
|
|
matching ``function_call_output``, both are dropped together.
|
|
Otherwise, a single item is dropped.
|
|
|
|
:param messages: The messages list (must be non-empty).
|
|
:returns: Number of items to drop (1 or 2), or 0 if the list
|
|
is empty.
|
|
"""
|
|
if not messages:
|
|
return 0
|
|
if (
|
|
len(messages) >= 2
|
|
and messages[0].get("type") == "function_call"
|
|
and messages[1].get("type") == "function_call_output"
|
|
and messages[0].get("call_id") == messages[1].get("call_id")
|
|
):
|
|
return 2
|
|
return 1
|
|
|
|
|
|
async def summarize_history(
|
|
messages_to_summarize: list[dict[str, Any]],
|
|
llm_client: Any, # llms.Client — typed as Any to avoid circular import
|
|
model: str,
|
|
connection: dict[str, str] | None = None,
|
|
runner_client: Any | None = None, # httpx.AsyncClient | None
|
|
conversation_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Layer 2: call the LLM to summarise conversation messages.
|
|
|
|
When *runner_client* is provided the summarization LLM call is
|
|
delegated to the runner's ``POST /v1/summarize`` endpoint so the
|
|
runner's credentials are used instead of the Omnigent server's. Falls
|
|
back to *llm_client* when no runner client is configured.
|
|
|
|
:param messages_to_summarize: The messages outside the recent
|
|
window to summarise, as Responses API input dicts. By the
|
|
time this is called, Layer 1 has already cleared binary
|
|
content blocks and tool result bodies from these messages.
|
|
:param llm_client: The LLM client to use, e.g. an instance of
|
|
``llms.Client``. Ignored when *runner_client* is set.
|
|
:param model: The LLM model string, e.g. ``"openai/gpt-4o"``.
|
|
:param connection: Per-provider connection overrides (api_key,
|
|
base_url, etc.) from the agent spec. ``None`` means use
|
|
environment variable defaults.
|
|
:param runner_client: Optional ``httpx.AsyncClient`` pointed at the
|
|
runner. When set, the summarization LLM call is delegated to
|
|
the runner's ``POST /v1/summarize`` endpoint. ``None`` falls
|
|
back to *llm_client*.
|
|
:param conversation_id: Session/conversation identifier, e.g.
|
|
``"conv_abc123"``. Forwarded to the runner so it can look up
|
|
the spec's auth credentials for the LLM call.
|
|
:returns: A dict with ``"text"`` (the summary) and
|
|
``"token_count"`` (approximate token count).
|
|
"""
|
|
if runner_client is not None:
|
|
return await _summarize_via_runner_uncached(
|
|
runner_client,
|
|
messages_to_summarize,
|
|
model,
|
|
connection,
|
|
conversation_id=conversation_id,
|
|
)
|
|
return await _summarize_history_uncached(
|
|
messages_to_summarize,
|
|
llm_client,
|
|
model,
|
|
connection,
|
|
)
|
|
|
|
|
|
async def _summarize_history_uncached(
|
|
messages_to_summarize: list[dict[str, Any]],
|
|
llm_client: Any,
|
|
model: str,
|
|
connection: dict[str, str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Run the Layer 2 summarization LLM call.
|
|
|
|
:param messages_to_summarize: Messages to summarize.
|
|
:param llm_client: LLM client instance, e.g. ``llms.Client()``.
|
|
:param model: LLM model string, e.g. ``"openai/gpt-4o"``.
|
|
:param connection: Per-provider connection overrides. ``None``
|
|
uses environment variable defaults.
|
|
:returns: Dict with ``"text"`` and ``"token_count"`` keys.
|
|
"""
|
|
system_prompt = build_summarization_prompt(messages_to_summarize)
|
|
resp = await llm_client.responses.create(
|
|
model=model,
|
|
input=build_summarization_input(messages_to_summarize),
|
|
instructions=system_prompt,
|
|
tools=[],
|
|
connection_params=connection,
|
|
)
|
|
summary_text = extract_summary_text(resp)
|
|
token_count = count_tokens([{"role": "assistant", "content": summary_text}], model)
|
|
return {"text": summary_text, "token_count": token_count}
|
|
|
|
|
|
async def _summarize_via_runner_uncached(
|
|
runner_client: Any, # httpx.AsyncClient
|
|
messages_to_summarize: list[dict[str, Any]],
|
|
model: str,
|
|
connection: dict[str, str] | None = None,
|
|
conversation_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
POST to the runner's ``/v1/summarize`` endpoint and return the result.
|
|
|
|
The runner creates its own LLM client using its local credentials,
|
|
so the Omnigent server needs no LLM auth for this call.
|
|
|
|
:param runner_client: ``httpx.AsyncClient`` pointed at the runner.
|
|
:param messages_to_summarize: Messages to summarize.
|
|
:param model: LLM model string, e.g. ``"openai/gpt-4o"``.
|
|
:param connection: Per-provider connection overrides forwarded to
|
|
the runner verbatim. ``None`` omits the field.
|
|
:param conversation_id: Session/conversation identifier, e.g.
|
|
``"conv_abc123"``. Sent in the payload so the runner can
|
|
look up the spec's auth credentials for the LLM call.
|
|
:returns: Dict with ``"text"`` (summary) and ``"token_count"``
|
|
(approximate tiktoken estimate) keys.
|
|
:raises httpx.HTTPStatusError: On non-2xx responses from the runner.
|
|
"""
|
|
payload: dict[str, Any] = {"messages": messages_to_summarize, "model": model}
|
|
if connection:
|
|
payload["connection"] = connection
|
|
if conversation_id is not None:
|
|
payload["session_id"] = conversation_id
|
|
resp = await runner_client.post("/v1/summarize", json=payload, timeout=120.0)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
def compaction_to_history_items(
|
|
compaction_item: ConversationItem,
|
|
) -> list[ConversationItem]:
|
|
"""
|
|
Convert a compaction item into a synthetic user + assistant
|
|
message pair for inclusion at the front of conversation history.
|
|
|
|
The pair preserves natural turn-taking structure: a synthetic
|
|
user message requests a summary, and a synthetic assistant
|
|
message provides it. This avoids attribution confusion —
|
|
the LLM knows it produced a summary (not a real prior response).
|
|
|
|
The synthetic items are NOT persisted to the conversation store;
|
|
they exist only in the in-memory history list for prompt
|
|
construction.
|
|
|
|
:param compaction_item: The compaction item from the store,
|
|
with ``type="compaction"`` and
|
|
``data`` of type :class:`~omnigent.entities.CompactionData`.
|
|
:returns: Two :class:`~omnigent.entities.ConversationItem`
|
|
instances: a ``role=user`` message requesting the summary
|
|
and a ``role=assistant`` message containing it.
|
|
"""
|
|
assert isinstance(compaction_item.data, CompactionData)
|
|
data = compaction_item.data
|
|
|
|
# Prefer compacted_messages when available — they carry the
|
|
# full compacted state (e.g. OpenAI's opaque compaction tokens
|
|
# or Claude's post-compaction transcript) that the harness can
|
|
# replay directly. Fall back to the synthetic summary pair for
|
|
# older compaction items that don't have compacted messages.
|
|
if data.compacted_messages:
|
|
items: list[ConversationItem] = []
|
|
for i, msg in enumerate(data.compacted_messages):
|
|
items.append(
|
|
ConversationItem(
|
|
id=f"{compaction_item.id}_compacted_{i}",
|
|
type=msg.get("type", "message"),
|
|
status="completed",
|
|
response_id=compaction_item.response_id,
|
|
created_at=compaction_item.created_at,
|
|
data=MessageData(
|
|
role=msg.get("role", "user"),
|
|
content=msg.get("content", []),
|
|
),
|
|
)
|
|
)
|
|
return items
|
|
|
|
synthetic_user_content = (
|
|
"[This is an automatically generated summary of the prior conversation "
|
|
"context. The original messages are available but not included in this "
|
|
"prompt for brevity.]\n\n"
|
|
"Please provide a summary of our conversation so far."
|
|
)
|
|
user_item = ConversationItem(
|
|
id=f"{compaction_item.id}_user",
|
|
type="message",
|
|
status="completed",
|
|
response_id=compaction_item.response_id,
|
|
created_at=compaction_item.created_at,
|
|
data=MessageData(
|
|
role="user",
|
|
content=[{"type": "input_text", "text": synthetic_user_content}],
|
|
),
|
|
)
|
|
assistant_item = ConversationItem(
|
|
id=f"{compaction_item.id}_assistant",
|
|
type="message",
|
|
status="completed",
|
|
response_id=compaction_item.response_id,
|
|
created_at=compaction_item.created_at,
|
|
data=MessageData(
|
|
role="assistant",
|
|
content=[{"type": "output_text", "text": data.summary}],
|
|
agent=data.model,
|
|
),
|
|
)
|
|
return [user_item, assistant_item]
|
|
|
|
|
|
async def compact(
|
|
messages: list[dict[str, Any]],
|
|
history: list[ConversationItem],
|
|
*,
|
|
config: CompactionConfig | None,
|
|
context_window: int,
|
|
system_token_budget: int,
|
|
model: str,
|
|
task_id: str,
|
|
llm_client: Any, # llms.Client — typed as Any to avoid circular import
|
|
connection: dict[str, str] | None = None,
|
|
runner_client: Any | None = None, # httpx.AsyncClient | None
|
|
force: bool = False,
|
|
fail_on_summary_error: bool = False,
|
|
conversation_id: str | None = None,
|
|
) -> CompactionResult:
|
|
"""
|
|
Apply layered compaction to a messages list to fit within the
|
|
context window budget.
|
|
|
|
Layers are applied in order from least-lossy to most-lossy:
|
|
|
|
1. **Layer 1** — Clear tool result bodies and binary content
|
|
blocks outside the recent window (fast, no LLM call).
|
|
2. **Layer 2** — LLM summarization of messages outside the
|
|
recent window (slow).
|
|
3. **Layer 3** — Truncate oldest messages (emergency fallback).
|
|
|
|
The in-memory *history* list is never modified — only the
|
|
*messages* copy passed to the LLM is compacted.
|
|
|
|
:param messages: The messages list to compact. This is a copy
|
|
— the original history is not modified.
|
|
:param history: The original conversation history items, used
|
|
to find ``last_item_id`` for the summary.
|
|
:param config: Compaction configuration from the agent spec.
|
|
``None`` uses defaults.
|
|
:param context_window: The model's context window size in tokens,
|
|
e.g. ``128000``.
|
|
:param system_token_budget: Tokens already consumed by the system
|
|
prompt and tool schemas, subtracted from the window budget.
|
|
:param model: The LLM model string, e.g. ``"openai/gpt-4o"``.
|
|
:param task_id: The task identifier for SSE event emission.
|
|
:param llm_client: The LLM client instance for Layer 2
|
|
summarization. Ignored when *runner_client* is set.
|
|
:param connection: Per-provider connection overrides (api_key,
|
|
base_url, etc.) passed through to the summarization call.
|
|
:param runner_client: Optional ``httpx.AsyncClient`` pointed at
|
|
the runner. When set, Layer 2 summarization is delegated to
|
|
the runner's ``POST /v1/summarize`` endpoint so the runner's
|
|
credentials are used instead of the Omnigent server's. ``None``
|
|
falls back to *llm_client*.
|
|
:param force: When ``True``, run through Layer 2 even if Layer 1
|
|
already fits within the budget. Used by explicit user-initiated
|
|
compaction (``/compact``) so a summary item is persisted even
|
|
before the automatic threshold is crossed.
|
|
:param fail_on_summary_error: When ``True``, propagate Layer 2
|
|
summarization failures instead of silently falling back to
|
|
Layer 3. Explicit ``/compact`` needs this so it never reports
|
|
success without a durable summary item.
|
|
:param conversation_id: When set, ``response.compaction.in_progress``
|
|
and ``response.compaction.completed`` are published to the session
|
|
stream so the REPL and web UI show the compaction indicator.
|
|
``None`` for explicit ``/compact`` — sessions.py handles those
|
|
events directly, e.g. ``"conv_abc123"``.
|
|
:returns: A :class:`CompactionResult` with the compacted messages
|
|
and optional summary metadata.
|
|
"""
|
|
trigger_threshold = config.trigger_threshold if config else _DEFAULT_TRIGGER_THRESHOLD
|
|
recent_window = config.recent_window if config else _DEFAULT_RECENT_WINDOW
|
|
# Budget = fraction of context window minus system/tool tokens.
|
|
budget = int(context_window * trigger_threshold) - system_token_budget
|
|
|
|
# Deep-copy messages so Layer 1 modifications don't affect the
|
|
# caller's list.
|
|
working = _deep_copy_messages(messages)
|
|
|
|
history_boundary = _find_recent_boundary(history, recent_window)
|
|
msg_boundary = _history_idx_to_msg_idx(history, history_boundary)
|
|
|
|
# --- Layer 1 ---
|
|
_clear_tool_results(working, msg_boundary)
|
|
_clear_binary_content(working, msg_boundary)
|
|
_strip_output_annotations(working, msg_boundary)
|
|
|
|
l1_tokens = count_tokens(working, model)
|
|
if not force and l1_tokens <= budget:
|
|
return CompactionResult(messages=working, summary_metadata=None, total_tokens=l1_tokens)
|
|
|
|
# --- Layer 2 ---
|
|
_logger.debug(
|
|
"Compaction Layer 2 summarization starting for task %s: "
|
|
"%d tokens after Layer 1 clearing, budget %d%s",
|
|
task_id,
|
|
l1_tokens,
|
|
budget,
|
|
" (forced)" if force else "",
|
|
)
|
|
if conversation_id:
|
|
# Publish to session stream so REPL/web UI shows "Compacting…" indicator.
|
|
# Imported locally to avoid circular imports at module level.
|
|
from omnigent.runtime import session_stream as _session_stream
|
|
|
|
_session_stream.publish(
|
|
conversation_id,
|
|
{"type": "response.compaction.in_progress", "task_id": task_id},
|
|
)
|
|
summary_metadata = await _run_layer2(
|
|
working,
|
|
history,
|
|
history_boundary,
|
|
msg_boundary,
|
|
budget,
|
|
model,
|
|
task_id,
|
|
llm_client,
|
|
connection=connection,
|
|
fail_on_error=fail_on_summary_error,
|
|
runner_client=runner_client,
|
|
conversation_id=conversation_id,
|
|
)
|
|
if summary_metadata is not None:
|
|
summary_messages = _summary_to_messages(summary_metadata)
|
|
recent_messages = working[msg_boundary:]
|
|
compacted = summary_messages + recent_messages
|
|
compacted_tokens = count_tokens(compacted, model)
|
|
if compacted_tokens <= budget:
|
|
_logger.debug(
|
|
"Compaction Layer 2 complete for task %s: %d tokens (budget %d)",
|
|
task_id,
|
|
compacted_tokens,
|
|
budget,
|
|
)
|
|
if conversation_id:
|
|
from omnigent.runtime import session_stream as _session_stream
|
|
|
|
_session_stream.publish(
|
|
conversation_id,
|
|
{
|
|
"type": "response.compaction.completed",
|
|
"task_id": task_id,
|
|
"total_tokens": compacted_tokens,
|
|
},
|
|
)
|
|
return CompactionResult(
|
|
messages=compacted,
|
|
summary_metadata=summary_metadata,
|
|
total_tokens=compacted_tokens,
|
|
)
|
|
# Summary + recent still exceeds budget — fall through to Layer 3.
|
|
working = compacted
|
|
|
|
# --- Layer 3 ---
|
|
_logger.warning(
|
|
"Layer 3 truncation triggered for task %s — context still exceeds budget after layers 1+2",
|
|
task_id,
|
|
)
|
|
truncated = _truncate_oldest(working, budget, model)
|
|
l3_tokens = count_tokens(truncated, model)
|
|
# Emit completed on the Layer 3 path so "Compacting…" spinners
|
|
# resolve even when Layer 2 failed or its output exceeded budget.
|
|
if conversation_id:
|
|
from omnigent.runtime import session_stream as _session_stream
|
|
|
|
_session_stream.publish(
|
|
conversation_id,
|
|
{
|
|
"type": "response.compaction.completed",
|
|
"task_id": task_id,
|
|
"total_tokens": l3_tokens,
|
|
},
|
|
)
|
|
return CompactionResult(
|
|
messages=truncated,
|
|
summary_metadata=summary_metadata,
|
|
total_tokens=l3_tokens,
|
|
)
|
|
|
|
|
|
def _deep_copy_messages(
|
|
messages: list[dict[str, Any]],
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Return a deep copy of the messages list so Layer 1 clearing
|
|
does not mutate the caller's list.
|
|
|
|
:param messages: The messages list to copy.
|
|
:returns: A deep copy.
|
|
"""
|
|
result: list[dict[str, Any]] = json.loads(json.dumps(messages))
|
|
return result
|
|
|
|
|
|
def _history_idx_to_msg_idx(
|
|
history: list[ConversationItem],
|
|
history_idx: int,
|
|
) -> int:
|
|
"""
|
|
Map a history index to the corresponding messages list index.
|
|
|
|
``history_to_input_items`` skips reasoning items, so the
|
|
messages list may be shorter than the history list. This
|
|
function counts non-reasoning items up to *history_idx*.
|
|
|
|
:param history: The full conversation history.
|
|
:param history_idx: The index in *history* to map.
|
|
:returns: The corresponding index in the messages list.
|
|
"""
|
|
msg_idx = 0
|
|
for i, item in enumerate(history):
|
|
if i >= history_idx:
|
|
break
|
|
if item.type != "reasoning":
|
|
msg_idx += 1
|
|
return msg_idx
|
|
|
|
|
|
def _is_summary_auth_error(exc: BaseException) -> bool:
|
|
"""Return ``True`` when *exc* is (or wraps) an HTTP 401/403.
|
|
|
|
Layer 2 summarization calls an LLM *outside* the harness, so a missing or
|
|
invalid summarizer credential surfaces here as an auth error. Detecting it
|
|
lets the caller surface a distinct, actionable message instead of burying a
|
|
persistent misconfiguration behind a routine Layer-3 truncation fallback
|
|
(issue #1121).
|
|
|
|
:param exc: The exception raised by the summarization call.
|
|
:returns: ``True`` for a 401/403 (by ``response.status_code`` or message).
|
|
"""
|
|
status = getattr(getattr(exc, "response", None), "status_code", None)
|
|
if status in (401, 403):
|
|
return True
|
|
text = str(exc)
|
|
return any(token in text for token in ("401", "403", "Unauthorized", "Forbidden"))
|
|
|
|
|
|
async def _run_layer2(
|
|
messages: list[dict[str, Any]],
|
|
history: list[ConversationItem],
|
|
history_boundary: int,
|
|
msg_boundary: int,
|
|
budget: int,
|
|
model: str,
|
|
task_id: str,
|
|
llm_client: Any,
|
|
connection: dict[str, str] | None = None,
|
|
fail_on_error: bool = False,
|
|
runner_client: Any | None = None, # httpx.AsyncClient | None
|
|
conversation_id: str | None = None,
|
|
) -> SummaryMetadata | None:
|
|
"""
|
|
Attempt Layer 2 LLM summarisation.
|
|
|
|
Emits a ``response.compaction.in_progress`` SSE event before
|
|
the LLM call. Returns ``None`` and falls through to Layer 3 if
|
|
the LLM call fails.
|
|
|
|
:param messages: The working messages list (after Layer 1).
|
|
:param history: The original conversation history items.
|
|
:param history_boundary: The boundary index in *history*.
|
|
:param msg_boundary: The boundary index in *messages*.
|
|
:param budget: Token budget for the compacted result.
|
|
:param model: LLM model string.
|
|
:param task_id: Task identifier for SSE event emission.
|
|
:param llm_client: LLM client instance. Ignored when
|
|
*runner_client* is set.
|
|
:param connection: Per-provider connection overrides (api_key,
|
|
base_url, etc.) passed through to the summarization call.
|
|
:param fail_on_error: When ``True``, re-raise summarization errors
|
|
after logging them.
|
|
:param runner_client: Optional ``httpx.AsyncClient`` pointed at the
|
|
runner. When set, delegates the summarization LLM call to the
|
|
runner's ``POST /v1/summarize`` endpoint. ``None`` uses
|
|
*llm_client* directly.
|
|
:param conversation_id: Session/conversation identifier, e.g.
|
|
``"conv_abc123"``. Forwarded to the runner so it can look up
|
|
the spec's auth credentials for the LLM call.
|
|
:returns: :class:`SummaryMetadata` on success, ``None`` on failure.
|
|
"""
|
|
|
|
to_summarize = messages[:msg_boundary]
|
|
# If too large for the model, apply Layer 1 clearing to the
|
|
# summarization input too.
|
|
if count_tokens(to_summarize, model) > budget:
|
|
to_summarize = _deep_copy_messages(to_summarize)
|
|
_clear_tool_results(to_summarize, len(to_summarize))
|
|
_clear_binary_content(to_summarize, len(to_summarize))
|
|
|
|
try:
|
|
summarize_kwargs: dict[str, Any] = {"connection": connection}
|
|
if runner_client is not None:
|
|
summarize_kwargs["runner_client"] = runner_client
|
|
if conversation_id is not None:
|
|
summarize_kwargs["conversation_id"] = conversation_id
|
|
result = await summarize_history(
|
|
to_summarize,
|
|
llm_client,
|
|
model,
|
|
**summarize_kwargs,
|
|
)
|
|
except Exception as exc:
|
|
if _is_summary_auth_error(exc):
|
|
# Distinct, actionable signal: an auth/config problem (not a
|
|
# transient blip) is silently degrading compaction to lossy
|
|
# truncation. Don't bury a 401 as a routine fallback.
|
|
_logger.error(
|
|
"Compaction Layer 2 summarisation is UNAUTHORIZED for task %s "
|
|
"(%s) — the summarizer's credentials are missing or invalid, so "
|
|
"compaction is degrading to lossy Layer-3 truncation. Fix the "
|
|
"summarizer auth/config to restore summary-quality compaction.",
|
|
task_id,
|
|
exc,
|
|
)
|
|
else:
|
|
_logger.warning(
|
|
"Layer 2 summarisation failed for task %s — falling back to Layer 3",
|
|
task_id,
|
|
exc_info=True,
|
|
)
|
|
if fail_on_error:
|
|
raise
|
|
return None
|
|
|
|
last_item_id = _find_last_summarized_item_id(history, history_boundary)
|
|
if last_item_id is None:
|
|
return None
|
|
|
|
return SummaryMetadata(
|
|
text=result["text"],
|
|
last_item_id=last_item_id,
|
|
model=model,
|
|
token_count=result["token_count"],
|
|
)
|
|
|
|
|
|
def _find_last_summarized_item_id(
|
|
history: list[ConversationItem],
|
|
history_boundary: int,
|
|
) -> str | None:
|
|
"""
|
|
Find the ID of the last history item included in the summary.
|
|
|
|
This is the item at ``history[history_boundary - 1]``, skipping
|
|
any synthetic items (those without a real store ID). Synthetic
|
|
items are identified by the ``_user`` or ``_assistant`` suffix
|
|
added by :func:`compaction_to_history_items`.
|
|
|
|
:param history: The conversation history items.
|
|
:param history_boundary: The boundary index (exclusive).
|
|
:returns: The last real item ID, or ``None`` if no real items
|
|
exist before the boundary.
|
|
"""
|
|
for i in range(history_boundary - 1, -1, -1):
|
|
item = history[i]
|
|
# Skip synthetic items: IDs with _user / _assistant suffix
|
|
# (from compaction_to_history_items) and synthetic_N IDs
|
|
# (from the runner's in-memory ConversationItem construction).
|
|
if not item.id.endswith(("_user", "_assistant")) and not item.id.startswith("synthetic_"):
|
|
return item.id
|
|
return None
|
|
|
|
|
|
def _summary_to_messages(
|
|
summary: SummaryMetadata,
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Convert a :class:`SummaryMetadata` into the synthetic
|
|
user + assistant message pair for inclusion in the prompt.
|
|
|
|
:param summary: The summary metadata from Layer 2.
|
|
:returns: Two message dicts: user request and assistant summary.
|
|
"""
|
|
user_text = (
|
|
"[This is an automatically generated summary of the prior conversation "
|
|
"context. The original messages are available but not included in this "
|
|
"prompt for brevity.]\n\n"
|
|
"Please provide a summary of our conversation so far."
|
|
)
|
|
return [
|
|
{"role": "user", "content": user_text},
|
|
{"role": "assistant", "content": summary.text},
|
|
]
|