agent/image_eviction_policy.py
Rama main · réplica de lectura del repositorio original.
"""Send-path image eviction policy shared by both stateless outbound passes.
Two passes retire old tool-result images from the per-request copy of the conversation:
``agent.context_compressor.evict_stale_outbound_tool_images`` on the OpenAI-shaped list and
``agent.anthropic_message_convert._evict_old_screenshots`` on the Anthropic wire list. Both run
from scratch on a fresh clone every request, so they must agree on one policy or the second
pass re-evicts on a different frontier than the first (#113517). This module is stdlib-only so
the wire converter, a leaf, can import it without dragging in the compaction stack.
Why the trigger is a provider limit, not a keep-newest count: retiring an image edits a message
the provider has already cached, and Anthropic matches its prompt cache on an exact byte prefix.
A keep-newest-N window retires one more message on every new image, so every turn is a
full-prefix miss. Holding images until the request would cross a real API limit and then
retiring a batch costs one slower turn per batch and nothing below the limit.
20 is the documented threshold at which Anthropic applies a stricter per-image dimension cap
(2000 px) to EVERY image in the request, counting images nested in tool_result content. Crossing
it costs nothing by itself: it only matters when some image actually exceeds 2000 px on a side.
Hermes already shrinks its own tool images far below that before they enter history
(``vision_analyze`` and browser screenshots embed at 1568 px, ``computer_use`` at 1456 px by
default), so the 21st screenshot of a run trips no provider constraint and a batch retire there
buys a full prefix rewrite for nothing — and, on the preserved-thinking models, invalidates every
later thinking block. Anthropic's own computer-use guidance now says to keep screenshots at or
under 2000 px and NOT to prune them client-side for exactly that reason.
So the block ceiling is CONDITIONAL on the request's actual dimensions: the strict 20 applies
only when some image exceeds ``MANY_IMAGE_DIMENSION_LIMIT``, when the stricter per-side cap is
live and the retire genuinely protects the request; otherwise the ceiling is the hard count
``OUTBOUND_IMAGE_HARD_LIMIT``. The hard ceilings above 20 are real but distant (100 images per
request on 200K-context models, 600 otherwise) and the 32 MB request-size limit usually binds
first, which the byte budget guards with headroom for text. Classification reads image HEADERS
only — never a full decode of every base64 payload per request — and treats unreadable or unknown
dimensions as unsafe, so a payload this module cannot measure keeps today's strict trigger.
"""
from __future__ import annotations
import base64
import struct
from typing import Any, Iterable, Optional, Sequence
OUTBOUND_IMAGE_LIMIT = 20
# Hard per-request image count on 200K-context Claude models; the ceiling once the many-image
# stricter cap is provably not live (see module docstring). Deliberately not 600 — the byte
# budget below binds first for a 200K-context transcript, and the byte ceiling needs no context
# knowledge to be safe.
OUTBOUND_IMAGE_HARD_LIMIT = 100
OUTBOUND_IMAGE_BUDGET_BYTES = 24_000_000
IMAGE_EVICTION_BATCH = 8
# Satisfiability floor — see outbound_image_retire_count.
OUTBOUND_IMAGE_FLOOR = 3
# Anthropic's stricter many-image per-side cap (px): above it, every image in the request is held
# to this limit, so an image over it keeps the strict trigger live.
MANY_IMAGE_DIMENSION_LIMIT = 2000
# Bytes of an embedded image to decode when reading its header. PNG IHDR ends at byte 24; a JPEG
# SOF marker lives within the first few KB. ~32 KB of base64 is generous headroom, and only the
# prefix is ever decoded (base64 expands 3 bytes per 4 chars).
_DIMENSION_PROBE_CHARS = 32 * 1024
_JPEG_SOF_MARKERS = frozenset({0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF})
def image_dimensions_from_bytes(raw: bytes) -> Optional[tuple[int, int]]:
"""(width, height) for PNG / JPEG bytes, or None when unreadable. PNG: IHDR. JPEG: walk
segments (skipping 0xFF fill bytes) to the first SOF marker; stop at SOS. Lives here so both
the send-path passes and the tool layer share one header-only reader."""
if raw.startswith(b"\x89PNG\r\n\x1a\n") and len(raw) >= 24:
width, height = struct.unpack(">II", raw[16:24]) # cannot fail: 8 bytes are guaranteed present
return int(width), int(height)
if raw.startswith(b"\xff\xd8") and len(raw) > 4:
i = 2
while i + 9 < len(raw):
if raw[i] != 0xFF:
i += 1
continue
marker, i = raw[i + 1], i + 2
while marker == 0xFF and i < len(raw):
marker, i = raw[i], i + 1
if marker in {0xD8, 0xD9}:
continue
if marker == 0xDA or i + 2 > len(raw):
break
segment_len = int.from_bytes(raw[i:i + 2], "big")
if segment_len < 2 or i + segment_len > len(raw):
break
if marker in _JPEG_SOF_MARKERS and segment_len >= 7:
return int.from_bytes(raw[i + 5:i + 7], "big"), int.from_bytes(raw[i + 3:i + 5], "big")
i += segment_len
return None
def image_max_side_from_data(data: Optional[str]) -> Optional[int]:
"""Longest side (px) of a data-URL / bare-base64 image, or None when it cannot be read.
Decodes only the leading base64 prefix (``_DIMENSION_PROBE_CHARS``), so classifying a request
with dozens of images costs a few header reads, not a full decode of every payload. Accepts
both ``data:image/...;base64,....`` and a bare base64 string; anything else (a remote URL, an
unrecognized container) yields None.
"""
if not data:
return None
payload = data
if payload.startswith("data:"):
_, separator, payload = payload.partition(",")
if not separator or ";base64" not in data[:data.index(",")]:
return None
payload = payload.strip()
prefix = payload[:_DIMENSION_PROBE_CHARS]
prefix = prefix[: len(prefix) - (len(prefix) % 4)] # decode needs a multiple-of-four length
if not prefix:
return None
try:
raw = base64.b64decode(prefix, validate=False)
except (ValueError, TypeError):
return None
dims = image_dimensions_from_bytes(raw)
return max(dims) if dims is not None else None
def image_within_many_image_limit(data: Optional[str]) -> bool:
"""True only when the image's longest side is known and within ``MANY_IMAGE_DIMENSION_LIMIT``.
Unknown dimensions are NOT safe: the stricter per-request cap would still be live, so the
caller keeps the strict trigger. See :func:`images_all_within_many_image_limit`.
"""
longest = image_max_side_from_data(data)
return longest is not None and longest <= MANY_IMAGE_DIMENSION_LIMIT
def images_all_within_many_image_limit(datas: Iterable[Optional[str]]) -> bool:
"""True when EVERY image in the request is a readable image within the many-image limit.
The Anthropic rule is per request: one oversized image holds every image in the request to the
stricter cap, so a single unreadable or over-limit image must keep the strict block ceiling.
Short-circuits on the first offender, which is what keeps this cheap on image-heavy turns.
"""
for data in datas:
if not image_within_many_image_limit(data):
return False
return True
def outbound_image_block_limit(many_image_safe: bool) -> int:
"""Block ceiling for a request: the strict 20 unless every image is within the many-image cap."""
return OUTBOUND_IMAGE_HARD_LIMIT if many_image_safe else OUTBOUND_IMAGE_LIMIT
def outbound_image_retire_count(
carrier_blocks_newest_first: Sequence[int],
reserved_blocks: int,
*,
carrier_bytes_newest_first: Optional[Sequence[int]] = None,
reserved_bytes: int = 0,
many_image_safe: bool = False,
limit: Optional[int] = None,
budget: int = OUTBOUND_IMAGE_BUDGET_BYTES,
batch: int = IMAGE_EVICTION_BATCH,
floor: int = OUTBOUND_IMAGE_FLOOR,
) -> int:
"""How many of the OLDEST image-bearing tool results to retire.
``carrier_blocks_newest_first`` is the image-block count per image-bearing tool result,
newest first; ``reserved_blocks`` counts images the pass must never rewrite (user
uploads). The byte dimension is active only when ``carrier_bytes_newest_first`` is
given. ``limit`` defaults to :func:`outbound_image_block_limit` of ``many_image_safe``,
so both passes pick the same ceiling from the same per-request classification.
The retire count must be a STEP FUNCTION of the overshoot, because the pass is recomputed
on every request: an exact ``count - limit`` target moves the frontier on every new image,
and a fixed one-batch retire stops enforcing the limit after the first batch. So the count
advances in quanta until the request fits.
The quantum is ``batch`` capped at ``window - floor``, where ``window`` is how many newest
carriers fit under the ceiling. A quantum wider than that would step past the newest frames
on every advance; cutting each step back to exactly ``total - floor`` instead makes the
retire count track ``total`` again — the per-image frontier this policy exists to avoid,
visible whenever a tool result carries several images or uploads fill most of the ceiling.
Holding for ``window - floor`` turns per advance is the most the floor allows.
The floor is a SATISFIABILITY floor: it shelters the newest frames only when reserved
uploads alone breach the block ceiling (no retirement can fix that), and never under byte
pressure — the request-size limit is hard and the provider answers 413. Even then the
count stays a step function: it retires whole quanta down toward the floor (keeping
``floor`` to ``floor + batch - 1`` carriers, fewer when the byte budget binds), so the
cached prefix is rewritten once per quantum rather than on every new image.
"""
if limit is None:
limit = outbound_image_block_limit(many_image_safe)
total = len(carrier_blocks_newest_first)
sizes = carrier_bytes_newest_first
return max(
(
_retire_count_for_request(
carrier_blocks_newest_first[total - m:],
reserved_blocks,
None if sizes is None else sizes[total - m:],
reserved_bytes,
limit,
budget,
batch,
floor,
)
for m in range(1, total + 1)
),
default=0,
)
def outbound_retired_image_ids(agent: Any) -> set[str]:
"""The session's latched set of tool-call ids whose images the send path has retired.
Keyed on ``agent.session_id`` so a new session (or a compaction rotation, which rewrites the
prefix anyway) starts clean and a reused provider call id cannot retire a fresh screenshot.
"""
session_id = getattr(agent, "session_id", None)
held = getattr(agent, "_outbound_retired_image_ids", None)
if held is None or held[0] != session_id:
held = (session_id, set())
agent._outbound_retired_image_ids = held
return held[1]
def latched_retire_count(
retire: int, carrier_ids_newest_first: Sequence[Optional[str]], retired_ids: Optional[set[str]]
) -> int:
"""Widen ``retire`` so every carrier up to the newest previously retired one stays retired,
then record the carriers being retired into ``retired_ids``.
The per-request decision cannot see history: a request that drops an upload (once it leaves
the replay window) relaxes the ceiling, and re-deciding from scratch would restore
screenshots an earlier request already retired, rewriting the cached prefix back.
"""
if retired_ids is None:
return retire
total = len(carrier_ids_newest_first)
latched = [n for n, cid in enumerate(carrier_ids_newest_first) if cid and cid in retired_ids]
retire = max(retire, total - latched[0]) if latched else retire
retired_ids.update(cid for cid in carrier_ids_newest_first[total - retire:] if cid)
return retire
def _retire_count_for_request(
carrier_blocks_newest_first: Sequence[int],
reserved_blocks: int,
sizes: Optional[Sequence[int]],
reserved_bytes: int,
limit: int,
budget: int,
batch: int,
floor: int,
) -> int:
"""Retire count for one request in isolation; see :func:`outbound_image_retire_count`."""
total = len(carrier_blocks_newest_first)
blocks_kept = [0] * (total + 1)
bytes_kept = [0] * (total + 1)
for i in range(total):
blocks_kept[i + 1] = blocks_kept[i] + carrier_blocks_newest_first[i]
bytes_kept[i + 1] = bytes_kept[i] + (sizes[i] if sizes is not None else 0)
def _bytes_fit(kept: int) -> bool:
return sizes is None or reserved_bytes + bytes_kept[kept] <= budget
def _fits(kept: int) -> bool:
return reserved_blocks + blocks_kept[kept] <= limit and _bytes_fit(kept)
if _fits(total):
return 0
floor = min(max(floor, 0), total)
if not _fits(0) and _bytes_fit(floor):
# Reserved uploads alone breach the block ceiling: no retirement can make the request
# fit it, so shelter the floor but keep advancing in whole quanta. Retiring down to
# exactly ``total - floor`` here moves the frontier on every new image (#133999). The
# quantum is capped by how many carriers the byte budget lets us keep, so byte
# pressure narrows the step instead of collapsing it back to the floor.
byte_window = max(k for k in range(floor, total + 1) if _bytes_fit(k))
quantum = max(1, min(batch, byte_window - floor + 1))
excess = total - floor
return excess - excess % quantum
window = max(k for k in range(total + 1) if _fits(k)) if _fits(0) else 0
quantum = max(1, min(batch, window - floor))
retire = 0
while retire < total:
retire = min(retire + quantum, total)
if _fits(total - retire):
break
return retire