refactor(gateway): add adapter reply delivery policy

[jethac]

Summary

• Adds a generic ReplyDeliveryPolicy hook for platform adapters.
• Lets adapters observe inbound messages before gateway dispatch.
• Routes the gateway's auto voice-reply decision through the adapter policy while preserving default behavior.

Why this matters

• Keeps platform-specific delivery decisions inside each adapter instead of growing more platform branches in the shared gateway runner.
• Gives adapters a narrow, testable ownership point for reply behavior that depends on inbound platform context.
• Reduces future merge-conflict risk by letting LINE, Telegram, Discord, and other adapters evolve delivery policy without repeatedly editing the same central voice/text decision path.
• Preserves the current gateway default for adapters that do not opt in, so this is an extension seam rather than a behavior change.

Stack

• Builds on test(line): regression coverage for typed media-cache routing #35785 (fix(line): correct media caching and message type mapping).
• Follow-up LINE behavior PR: feat(line): add opt-in smart reply modality #40933 (feat(line): add opt-in smart reply modality).
• GitHub does not allow these fork branches to be used as upstream PR base refs, so this PR targets main; review it after test(line): regression coverage for typed media-cache routing #35785 and before feat(line): add opt-in smart reply modality #40933.

Test Plan

• python -m py_compile gateway/run.py gateway/platforms/base.py plugins/platforms/line/adapter.py
• python -m pytest tests/gateway/test_reply_delivery_policy.py tests/gateway/test_line_plugin.py tests/gateway/test_telegram_audio_vs_voice.py -q -o 'addopts='
• git diff --check line-media-adapter-pr35785-rebased...refactor/gateway-reply-delivery-policy

[liuhao1024]

cache_image_from_url regression: image validation lost

The refactoring from cache_image_from_url → cache_media_from_url drops the _looks_like_image() validation that previously guarded the download-and-cache path.

Before (current main):

async def cache_image_from_url(url, ext=".jpg", retries=2):
    # ...
    return cache_image_from_bytes(response.content, ext)
    #   ^ validates via _looks_like_image(), raises ValueError for non-image data

After (this PR):

async def cache_image_from_url(url, ext=".jpg", retries=2):
    """Deprecated alias for cache_media_from_url"""
    return await cache_media_from_url(url, ext=ext, retries=retries)

async def cache_media_from_url(url, ext=".bin", retries=2):
    # ...
    return cache_media_from_bytes(response.content, ext)
    #   ^ is_image defaults to False — no image validation, cache dir becomes cache/media/

The Accept header also changes from image/*,*/*;q=0.8 to */*, so non-image responses are now silently cached instead of rejected.

Callers affected (at least):

• plugins/platforms/discord/adapter.py:4648 — await cache_image_from_url(att.url, ext=ext)
• plugins/platforms/teams/adapter.py:790 — await cache_image_from_url(content_url)

Suggested fix — add is_image parameter to cache_media_from_url so cache_image_from_url can preserve validation:

async def cache_media_from_url(url, ext=".bin", retries=2, *, is_image=False):
    # ...
    return cache_media_from_bytes(response.content, ext, is_image=is_image)

async def cache_image_from_url(url, ext=".jpg", retries=2):
    return await cache_media_from_url(url, ext=ext, retries=retries, is_image=True)

[jethac]

Rebased this PR onto the refreshed LINE media branch (#35785) and fixed the voice-reply regressions from the previous revision.

Additional cleanup in this update:

• Preserves legacy empty/error response gates before adapter reply policy can request TTS.
• Ignores invalid/non-ReplyDeliveryPolicy adapter returns and falls back to legacy behavior.
• Keeps existing streaming/fresh-final hooks while adding the adapter policy seam.

Local verification:

uv run --with pytest --with pytest-asyncio --with aiohttp python -m pytest \
  tests/gateway/test_voice_command.py \
  tests/gateway/test_reply_delivery_policy.py \
  tests/gateway/test_media_download_retry.py \
  tests/gateway/test_line_plugin.py \
  -q -o 'addopts='
# 283 passed, 21 skipped

[jethac]

@liuhao1024 thanks again for catching the cache_image_from_url regression. This is actioned in the current branch head.\n\nThe refactor now preserves the image-specific path by threading an is_image flag through cache_media_from_url(...) / cache_media_from_bytes(...), so cache_image_from_url(...) calls cache_media_from_url(..., is_image=True). That keeps:\n\n- _looks_like_image() validation via cache_image_from_bytes(...)\n- image cache routing\n- the stricter Accept: image/*,*/*;q=0.8 header\n\nI also added regression coverage for rejecting non-image response bodies and preserving the image Accept header. Verified locally with the targeted tests passing.

[teknium1]

Thanks for extracting a platform delivery seam; the stated LINE follow-up gives it a concrete consumer. Two changes are needed when salvaging onto current main.

Problems

• gateway/run.py:12331 uses self.adapters.get(event.source.platform). Current main commit 8a9bc38c documents that this selects the default profile's adapter in multiplex mode; use _adapter_for_source(event.source) so secondary-profile delivery policy is evaluated by the correct adapter.
• gateway/run.py:12339 calls the new adapter policy without a fallback on exception. The adjacent new observe_inbound_message path deliberately catches callback failures (gateway/run.py:12314-12321); policy failures should likewise preserve the legacy reply path rather than disrupt final delivery.

Suggested changes

• Resolve via _adapter_for_source(event.source) and add a multiplex-source regression.
• Catch policy callback exceptions, fall back to the legacy policy, and test that behavior.

Automated hermes-sweeper review.

[teknium1]

Current main's multiplex fix 8a9bc38c establishes that self.adapters is the default-profile map. Resolve this through _adapter_for_source(event.source) so a secondary profile evaluates its own adapter's policy.

[jethac]

Fixed in 1646488 — both lookups this PR introduced (_observe_inbound_message and _reply_delivery_policy) now resolve via self._adapter_for_source(event.source). Two multiplex regressions in tests/gateway/test_reply_delivery_policy.py prove a source stamped for a secondary profile consults that profile's adapter's policy, and the default-profile adapter is never called.

[teknium1]

Please catch policy-callback exceptions and fall back to the legacy policy. The adjacent inbound-observer callback is already isolated; an adapter policy failure should not interrupt final-reply delivery.

[jethac]

Fixed in e1a9252 — the reply_delivery_policy callback is now wrapped in try/except mirroring the inbound-observer isolation pattern: on exception it logs a warning (exc_info=True) and falls back to the legacy voice-mode gate, so a policy failure can never disrupt final-reply delivery. Covered by test_policy_callback_failure_falls_back_to_legacy_delivery (raising policy → legacy gate still honored in both the off and /voice all states).

[jethac]

Thanks for the review — both points fixed; specifics in the inline replies (adapter resolution via _adapter_for_source with multiplex regressions in 1646488; exception-isolated policy callback with legacy fallback in e1a9252). The branch is also rebased onto current main and restacked on #35785's re-scoped media fix, so this diff no longer carries the old generic cache API. CI green, mergeable.