fix: simplify animation montage notices

astrbot/core/pipeline/process_stage/method/agent_sub_stages/image_input.py

@@ -8,8 +8,6 @@ from astrbot.core.platform.astr_message_event import AstrMessageEvent
 from astrbot.core.provider.entities import ProviderRequest
 from astrbot.core.utils.media_utils import (
     MediaResolver,
-    PreparedModelImage,
-    format_animation_montage_notice,
     is_recoverable_image_error,
     prepare_model_image,
 )
@@ -51,12 +49,12 @@ async def prepare_request_images(
     if quote_image_ref:
         refs.append(quote_image_ref)
     failed = False
-    montage_frames: list[int] = []
+    has_montage = False
     for ref in dict.fromkeys(refs):
         if ref not in prepared:
             path = None
             if enabled:
-                image: PreparedModelImage | None = await prepare_model_image(
+                image = await prepare_model_image(
                     ref,
                     max_size=max_size,
                     output_dir=output_dir,
@@ -64,10 +62,9 @@ async def prepare_request_images(
                     montage_max_size=montage_max_size,
                 )
                 if image:
-                    path = image.path
+                    path, is_montage = image
                     event.track_temporary_local_file(path)
-                    if image.montage_frames is not None:
-                        montage_frames.append(image.montage_frames)
+                    has_montage |= is_montage
             else:
                 try:
                     async with MediaResolver(
@@ -125,10 +122,13 @@ async def prepare_request_images(
         ):
             req.prompt = "[Image unavailable]"
 
-    notice = format_animation_montage_notice(montage_frames)
-    if notice:
+    if has_montage:
         # Transient per-request hint; the montage file itself is not persisted.
         req.extra_user_content_parts = [
             *req.extra_user_content_parts,
-            TextPart(text=notice).mark_as_temp(),
+            TextPart(
+                text="[Animated image] Animated inputs (e.g. GIF) have been converted "
+                "to frame montages. Read each montage left to right, top to bottom "
+                "as frames of the same animation; unused cells are blank."
+            ).mark_as_temp(),
         ]

astrbot/core/utils/media_utils.py

@@ -195,27 +195,6 @@ class ResolvedMediaFile:
         _cleanup_paths(self.cleanup_paths)
 
 
-@dataclass(slots=True)
-class PreparedModelImage:
-    """A prepared model-ready image file produced for one request.
-
-    Attributes:
-        path: Local working file the caller owns and attaches to the request.
-        montage_frames: Number of animation frames tiled into a grid montage
-            when the input was animated; ``None`` for still images. Callers
-            describe this to the model so a contact sheet is not misread as a
-            single picture.
-    """
-
-    path: str
-    montage_frames: int | None = None
-
-    @property
-    def is_montage(self) -> bool:
-        """Whether this file is a tiled animation frame montage."""
-        return self.montage_frames is not None
-
-
 def is_file_uri(value: object) -> bool:
     """Return whether a value is a ``file:`` URI.
 
@@ -1294,33 +1273,11 @@ def _even_frame_indices(total_frames: int, max_frames: int) -> list[int]:
     return sorted({round(i * (total_frames - 1) / (count - 1)) for i in range(count)})
 
 
-def _animation_frame_indices(image: PILImage.Image) -> list[int]:
-    """Frame indices tiled into the montage for an opened animated image.
-
-    APNG's independent default image is a cover, not an animation frame, so
-    sampling starts after it.
-
-    Args:
-        image: Opened animated source image.
-
-    Returns:
-        Ascending, unique frame indices to tile, capped at the grid size.
-    """
-    total_frames = getattr(image, "n_frames", 1)
-    first_frame = 1 if image.info.get("default_image", False) else 0
-    return [
-        first_frame + index
-        for index in _even_frame_indices(
-            total_frames - first_frame, ANIMATED_MONTAGE_FRAME_COUNT
-        )
-    ]
-
-
 def _extract_animation_montage_sync(
     source_bytes: bytes,
     max_size: int,
     quality: int,
-) -> tuple[bytes, int]:
+) -> tuple[bytes, bool]:
     """Tile evenly spaced frames of an animated image into one grid montage.
 
     Frame sampling includes the first and last frames; grid cells beyond the
@@ -1332,8 +1289,8 @@ def _extract_animation_montage_sync(
         max_size: Longest edge of the montage in pixels.
 
     Returns:
-        Tuple of the montage bytes and the number of frames tiled into it, so
-        callers can describe the contact sheet without decoding it again.
+        Tuple of the montage bytes and whether extraction actually ran
+        (``False`` when a valid cached montage was served).
     """
     cache_key = _image_convert_cache_key(
         source_bytes,
@@ -1342,12 +1299,18 @@ def _extract_animation_montage_sync(
     output_path = _image_convert_cache_dir() / (cache_key + ".img")
     cached = _read_valid_cached_image_bytes(output_path)
     if cached is not None:
-        # Only Sampling metadata is needed here, never the cached pixels.
-        with PILImage.open(io.BytesIO(source_bytes)) as cached_source:
-            return cached, len(_animation_frame_indices(cached_source))
+        return cached, False
     with PILImage.open(io.BytesIO(source_bytes)) as image:
-        frame_indices = _animation_frame_indices(image)
-        image.seek(frame_indices[0])
+        total_frames = getattr(image, "n_frames", 1)
+        # APNG's independent default image is a cover, not an animation frame.
+        first_frame = 1 if image.info.get("default_image", False) else 0
+        frame_indices = [
+            first_frame + index
+            for index in _even_frame_indices(
+                total_frames - first_frame, ANIMATED_MONTAGE_FRAME_COUNT
+            )
+        ]
+        image.seek(first_frame)
         with ImageOps.exif_transpose(image) as oriented:
             display_size = oriented.size
         # Floor the per-cell scale so the montage never exceeds max_size.
@@ -1394,38 +1357,7 @@ def _extract_animation_montage_sync(
         finally:
             canvas.close()
     _publish_image_cache_atomic(output_path, encoded)
-    return encoded, len(frame_indices)
-
-
-def format_animation_montage_notice(frame_counts: list[int]) -> str:
-    """Describe tiled animation frames to the model in one user-content notice.
-
-    A montage replaces the animation with a contact sheet, so without a notice
-    the model reads the grid as a single picture. Only montages are described;
-    still images never produce a notice.
-
-    Args:
-        frame_counts: Number of tiled frames, one entry per montage image.
-
-    Returns:
-        A notice block, or an empty string when no montage was built.
-    """
-    if not frame_counts:
-        return ""
-    grid = f"{ANIMATED_MONTAGE_GRID}x{ANIMATED_MONTAGE_GRID}"
-    subject = (
-        "1 attached image is an animated image"
-        if len(frame_counts) == 1
-        else f"{len(frame_counts)} attached images are animated images"
-    )
-    counts = ", ".join(str(count) for count in frame_counts)
-    return (
-        f"[Animated image] {subject} (e.g. GIF) rendered as a {grid} contact "
-        f"sheet of evenly spaced frames (frames per sheet: {counts}). Cells are "
-        "laid out in reading order, left to right then top to bottom, and any "
-        "remaining cells are blank. Treat each cell as one frame of the same "
-        "animation, not as a separate picture."
-    )
+    return encoded, True
 
 
 async def prepare_model_image(
@@ -1435,7 +1367,7 @@ async def prepare_model_image(
     output_dir: Path,
     quality: int = IMAGE_COMPRESS_DEFAULT_QUALITY,
     montage_max_size: int | None = None,
-) -> PreparedModelImage | None:
+) -> tuple[str, bool] | None:
     """Prepare a single local model-ready image for the caller to own until consumption.
 
     Args:
@@ -1450,18 +1382,18 @@ async def prepare_model_image(
             ``max_size``.
 
     Returns:
-        An existing JPEG or PNG working file, or None for a recoverable input
-        or write failure. ``montage_frames`` reports how many animation frames
-        were tiled when the input was animated, and is ``None`` otherwise.
+        Tuple of an existing JPEG or PNG path and whether it is an animation
+        frame montage (including cache hits), or None for a recoverable input
+        or write failure.
         The caller owns this file; shared cache entries are never returned.
     """
     try:
         async with MediaResolver(image_ref, media_type="image").as_path() as source:
             image_bytes = await asyncio.to_thread(source.read_bytes)
         frame_count = await asyncio.to_thread(_inspect_image, image_bytes)
-        montage_frames: int | None = None
-        if frame_count > 1:
-            converted_bytes, montage_frames = await asyncio.to_thread(
+        is_montage = frame_count > 1
+        if is_montage:
+            converted_bytes, _ = await asyncio.to_thread(
                 _extract_animation_montage_sync,
                 image_bytes,
                 montage_max_size if montage_max_size is not None else max_size,
@@ -1485,7 +1417,7 @@ async def prepare_model_image(
         except BaseException:
             output_path.unlink(missing_ok=True)
             raise
-        return PreparedModelImage(path=str(output_path), montage_frames=montage_frames)
+        return str(output_path), is_montage
     except Exception as exc:
         if not is_recoverable_image_error(exc):
             raise

tests/test_model_image_preparation.py

@@ -21,7 +21,7 @@ def isolated_cache(tmp_path, monkeypatch):
 async def _prepare_image_path(*args, **kwargs):
     """Path-only view of prepare_model_image for pixel-focused assertions."""
     prepared = await media.prepare_model_image(*args, **kwargs)
-    return None if prepared is None else prepared.path
+    return None if prepared is None else prepared[0]
 
 
 @pytest.mark.asyncio
@@ -386,9 +386,9 @@ async def test_real_download_status_failure_is_skipped_without_secret_logs(
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("count,expected", [(2, 2), (12, 9)])
-async def test_animation_reports_tiled_frame_count(tmp_path, count, expected):
-    """Animated inputs report how many frames were tiled into the montage."""
+@pytest.mark.parametrize("count", [2, 12])
+async def test_animation_reports_montage_including_cache_hits(tmp_path, count):
+    """Animated inputs report a montage even when its bytes are cached."""
     frames = [Image.new("RGB", (12, 8), (index * 20, 0, 0)) for index in range(count)]
     source = tmp_path / "animation.gif"
     frames[0].save(
@@ -398,19 +398,19 @@ async def test_animation_reports_tiled_frame_count(tmp_path, count, expected):
     prepared = await media.prepare_model_image(
         str(source), max_size=1280, output_dir=tmp_path
     )
-    assert prepared and prepared.is_montage and prepared.montage_frames == expected
-    assert Path(prepared.path).is_file()
+    assert prepared and prepared[1] is True
+    assert Path(prepared[0]).is_file()
 
     cached = await media.prepare_model_image(
         str(source), max_size=1280, output_dir=tmp_path
     )
-    assert cached and cached.montage_frames == expected
-    assert Path(cached.path).read_bytes() == Path(prepared.path).read_bytes()
+    assert cached and cached[1] is True
+    assert Path(cached[0]).read_bytes() == Path(prepared[0]).read_bytes()
 
 
 @pytest.mark.asyncio
-async def test_apng_cover_is_not_counted_as_tiled_frame(tmp_path):
-    """An APNG cover shifts sampling but is not reported as a tiled frame."""
+async def test_apng_with_cover_reports_montage(tmp_path):
+    """An APNG with a cover still reports an animation montage."""
     frames = [Image.new("RGB", (12, 8), (index * 20, 0, 0)) for index in range(3)]
     # An APNG cover frame precedes the animation frames it is not part of.
     frames.insert(0, Image.new("RGB", (12, 8), "blue"))
@@ -427,26 +427,16 @@ async def test_apng_cover_is_not_counted_as_tiled_frame(tmp_path):
     prepared = await media.prepare_model_image(
         str(source), max_size=1280, output_dir=tmp_path
     )
-    assert prepared and prepared.montage_frames == 3
+    assert prepared and prepared[1] is True
 
 
 @pytest.mark.asyncio
 async def test_still_image_reports_no_montage(tmp_path):
-    """Still images never report tiled frames, so no montage notice is built."""
+    """Still images do not report a montage, so no notice is built."""
     source = tmp_path / "still.png"
     Image.new("RGB", (32, 16), "red").save(source)
     prepared = await media.prepare_model_image(
         str(source), max_size=1280, output_dir=tmp_path
     )
     assert prepared is not None
-    assert not prepared.is_montage and prepared.montage_frames is None
-
-
-def test_animation_montage_notice_describes_frames_only_when_present():
-    """The notice describes montages and stays empty for still-only requests."""
-    assert media.format_animation_montage_notice([]) == ""
-    single = media.format_animation_montage_notice([4])
-    assert "1 attached image" in single and "4" in single
-    assert "3x3" in single and "blank" in single
-    multi = media.format_animation_montage_notice([9, 2])
-    assert "2 attached images" in multi and "9, 2" in multi
+    assert prepared[1] is False

tests/test_process_stage_images.py

@@ -327,14 +327,6 @@ async def test_cua_montage_keeps_configured_cap(harness, tmp_path):
         assert max(image.size) <= 90
 
 
-def _montage_notices(req):
-    return [
-        part
-        for part in req.extra_user_content_parts
-        if isinstance(part, TextPart) and part.text.startswith("[Animated image]")
-    ]
-
-
 @pytest.mark.asyncio
 async def test_animation_montage_notice_reaches_model(harness, tmp_path):
     """Animated inputs tell the model that the image is a frame montage."""
@@ -342,15 +334,25 @@ async def test_animation_montage_notice_reaches_model(harness, tmp_path):
     frames = [PILImage.new("RGB", (60, 30), c) for c in ("red", "green", "blue")]
     frames[0].save(animated, "GIF", save_all=True, append_images=frames[1:])
 
-    await process_event(harness, make_event([Image(file=str(animated))]))
-    notices = _montage_notices(harness.captured[-1].req)
-    assert len(notices) == 1
-    assert notices[0]._no_save and "3x3" in notices[0].text
-
     still = tmp_path / "still.png"
     PILImage.new("RGB", (60, 30), "red").save(still)
-    await process_event(harness, make_event([Image(file=str(still))]))
-    assert _montage_notices(harness.captured[-1].req) == []
+    # The second animated request reuses the montage cache.
+    for source, enabled, expected in (
+        (animated, True, True),
+        (animated, True, True),
+        (still, True, False),
+        (animated, False, False),
+    ):
+        harness.config["provider_settings"]["image_compress_enabled"] = enabled
+        await process_event(harness, make_event([Image(file=str(source))]))
+        notices = [
+            part
+            for part in harness.captured[-1].req.extra_user_content_parts
+            if isinstance(part, TextPart) and part.text.startswith("[Animated image]")
+        ]
+        assert len(notices) == int(expected)
+        if expected:
+            assert notices[0]._no_save and "frame montages" in notices[0].text
 
 
 @pytest.mark.asyncio