fix(anthropic): keep the type-discipline budget flat

Build the candidate indices as a tuple, take the already-marked messages as a
Container with a frozenset default, and say why the one remaining set is mutable.
This commit is contained in:
Tan Nguyen 2026-09-23 13:39:11 +07:00
parent b6a22917d9
commit d45077ea82

View file

@ -12,7 +12,7 @@ Supported for both `v1/chat/completions` (via the prompt-management hook) and
import copy
import os
import re
from collections.abc import Iterable, Mapping, Sequence
from collections.abc import Container, Iterable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, cast
from urllib.parse import urlparse
@ -439,7 +439,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
"""
used_blocks = AnthropicCacheControlHook.count_request_cache_breakpoints(messages)
taken: set[int] = set()
taken: set[int] = set() # mutable-ok: the messages this request has already marked
limit_reached = False
for point in points:
if used_blocks >= max_blocks:
@ -481,7 +481,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
@staticmethod
def _resolve_target_indices(
point: CacheControlMessageInjectionPoint, messages: list[AllMessageValues], taken: set[int] | None = None
point: CacheControlMessageInjectionPoint, messages: list[AllMessageValues], taken: Container[int] = frozenset()
) -> list[int]:
"""Resolve which message indices an injection point targets.
@ -490,7 +490,6 @@ class AnthropicCacheControlHook(CustomPromptManagement):
point turns two configured breakpoints into one, and the four exist so that a
prefix which stops matching at one can still match at an earlier one.
"""
already_marked: Final = taken or set()
_targetted_index: Final[int | str | None] = point.get("index", None)
targetted_index: int | None = None
if isinstance(_targetted_index, str):
@ -505,11 +504,11 @@ class AnthropicCacheControlHook(CustomPromptManagement):
# that role's turns, so {role: assistant, index: -1} is the last assistant
# turn rather than the last message.
targetted_role: Final = point.get("role", None)
candidates: Final = [
candidates: Final = tuple(
index
for index, message in enumerate(messages)
if targetted_role is None or message.get("role") == targetted_role
]
)
# Case 1: Target by role alone
if targetted_index is None:
@ -538,7 +537,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
position = targetted_index
while position >= 0:
index = candidates[position]
if index not in already_marked and _message_accepts_cache_control(messages[index]):
if index not in taken and _message_accepts_cache_control(messages[index]):
return [index]
position -= 1
return []