fix(gateway): keep in-flight final delivery single-writer

[michaelversluis]

Summary

Prevent duplicate gateway replies when a streamed final delivery is still in flight under a platform RetryAfter longer than the gateway's own wait ceiling.

This preserves and builds on the implementation from #79592 (original commit/authorship retained), then closes the remaining long-RetryAfter gap found in a live Telegram incident.

Root cause

The stream consumer and the normal gateway final-send path can both become writers for the same final response:

1. the stream consumer starts its final send/edit;
2. Telegram returns RetryAfter and the adapter keeps that attempt alive;
3. the gateway waits for a fixed total duration;
4. if the platform wait outlives that duration, the gateway cancels/abandons the consumer and starts its own final send;
5. the already-issued first request can still land, producing two separate messages.

#79592 correctly introduced final_delivery_in_progress and waits past the original five seconds, but capped the total at 90 seconds. A production case on v0.20.2 returned 202–212 second waits, reproducing the duplicate beyond that cap:

06:11:29 one inbound message
06:11:34 streamed final path: RetryAfter 212s
06:11:44 one Codex call completes; normal final-send path starts
06:11:44 normal final path: RetryAfter 202s
06:15:06 both separately queued messages land

Changes

• Preserve fix(gateway): await in-flight streamed final send before duplicate-resend decision #79592's explicit final_delivery_in_progress state and all three gateway call-site conversions.
• Make an active final-delivery attempt the single writer until its platform adapter settles.
• Remove the arbitrary gateway total ceiling; adapters retain ownership of retry counts and request timeouts.
• Keep the short cancellation path unchanged when no final-delivery attempt is active.
• Add deterministic coverage for:
  • an in-flight final that resolves after the former total ceiling;
  • an in-flight final that fails, followed by exactly one normal fallback;
  • a stalled non-final consumer that is still cancelled normally.

TDD evidence

The beyond-ceiling regression was run first against the #79592 base and failed with two visible messages:

FAILED test_retry_after_longer_than_wait_ceiling_still_delivers_once
AssertionError: final response visible in 2 separate messages
all sends=[... 'm-1', ... 'm-2']

After the single-writer change:

HERMES_TEST_FILE_RETRIES=0 scripts/run_tests.sh \
  tests/gateway/test_final_send_flood_race.py \
  tests/gateway/test_stream_consumer.py \
  tests/gateway/test_run_cleanup_progress.py \
  tests/gateway/test_run_progress_interrupt.py \
  tests/gateway/test_run_progress_topics.py -q

102 passed, 1 skipped

Additional checks:

ruff check gateway/run.py gateway/stream_consumer.py tests/gateway/test_final_send_flood_race.py
All checks passed!

git diff --check
clean

A local full tests/gateway/ run was attempted but the minimal dev environment lacks optional platform extras (aiohttp, Matrix, etc.). uv sync --all-extras is blocked on macOS by the existing python-olm==3.2.16 native build (cmake/gmake plus a const-qualified C++ compile error). The affected production and regression surfaces above are green; upstream CI should exercise the complete platform matrix.

Scope and attribution

• Original in-flight state tracking and three call-site conversion: JARVIS / fix(gateway): await in-flight streamed final send before duplicate-resend decision #79592, preserved as its own authored commit.
• Long-RetryAfter production reproduction, RED test, single-writer completion, failure fallback, and non-final cancellation coverage: this PR.

Related: #74248, #76494, #79592.

---

Nederlandse samenvatting: bij Telegram-wachttijden boven 90 seconden kon de bestaande fix alsnog twee eindberichten versturen. Deze versie houdt één actieve eindlevering als enige schrijver, maar laat na een echte mislukking nog steeds precies één normale fallback toe.

[Enough1122]

AI code review - automated review for reference; please use your judgment.

Reviewed the diff. Correct fix for a duplicate-send race with a genuinely subtle async detail handled properly: the old flat 5s wait_for cancelled the stream task on timeout and waited through its unwinding before raising, so by the time the caller checked final_delivery_in_progress the in-flight Telegram flood-control send had already been torn down (guaranteed False) and the gateway sent its own duplicate "final" response alongside the abandoned-but-delivered one. Switching to asyncio.wait (reports pending without touching the task), treating an active final-delivery attempt as the single writer awaited to completion under the adapter's own retry policy, and never reading "in flight" as "delivered" is exactly right. All three legacy wait sites are migrated, and the docstring explains the wait-vs-wait_for choice so nobody simplifies it back.

Nit: the no-timeout branch means an adapter with an unbounded retry loop would hold the turn forever; the comment leans on "adapters own bounded retries" - worth one assert/comment at the adapter layer stating retries are bounded by policy.