fix(rate-limiter): bound hosted-key queue wait to execution budget; fix heartbeat + telemetry

Tie the per-workspace hosted-key queue wait to the surrounding execution
budget instead of a flat 5-minute cap. acquireKey now accepts the execution
AbortSignal (threaded from ExecutionContext): when present, the wait is
bounded by the run's actual plan timeout / cancellation, with the enterprise
async ceiling as a backstop; when absent it falls back to MAX_QUEUE_WAIT_MS.
This lets long-running async (Trigger.dev) runs use their full budget while
no longer letting a single queued call burn a short sync run's entire budget.

Also addresses Greptile review:
- P1: share one lastHeartbeatAt across all wait phases and cap every sleep to
  HEARTBEAT_REFRESH_INTERVAL_MS so a long low-RPM retryAfterMs can no longer
  let the head's heartbeat lapse mid-wait and break FIFO ordering.
- P2: derive hostedKeyQueueWaited telemetry reason from the actual bottleneck
  (queue_position / dimension / actor_requests) instead of hardcoding it.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: TheodoreSpeaks

apps/sim/lib/core/rate-limiter/hosted-key/hosted-key-rate-limiter.test.ts

@@ -5,7 +5,7 @@ import type {
   TokenStatus,
 } from '@/lib/core/rate-limiter/storage'
 import { HostedKeyRateLimiter } from './hosted-key-rate-limiter'
-import type { HostedKeyQueue } from './queue'
+import { HEARTBEAT_REFRESH_INTERVAL_MS, type HostedKeyQueue } from './queue'
 import type { CustomRateLimit, PerRequestRateLimit } from './types'
 
 /** Force the queue wait to give up on the first iteration by reporting a retry time
@@ -370,6 +370,102 @@ describe('HostedKeyRateLimiter', () => {
     })
   })
 
+  describe('execution-budget-bounded waits', () => {
+    it('bails immediately when the execution signal is already aborted', async () => {
+      const blocked: ConsumeResult = {
+        allowed: false,
+        tokensRemaining: 0,
+        resetAt: new Date(Date.now() + 100),
+      }
+      mockAdapter.consumeTokens.mockResolvedValue(blocked)
+
+      const result = await rateLimiter.acquireKey(
+        testProvider,
+        envKeyPrefix,
+        perRequestRateLimit,
+        'workspace-1',
+        AbortSignal.abort()
+      )
+
+      expect(result.success).toBe(false)
+      expect(result.billingActorRateLimited).toBe(true)
+      // Aborted budget => give up on the first bucket check rather than looping.
+      expect(mockAdapter.consumeTokens).toHaveBeenCalledTimes(1)
+    })
+
+    it('keeps waiting past the no-signal fallback cap while the signal is live', async () => {
+      // A live (non-aborted) signal means the run still has budget, so the wait must not
+      // 429 at the 5-minute MAX_QUEUE_WAIT_MS fallback. The bucket frees up after ~7 min.
+      const blocked: ConsumeResult = {
+        allowed: false,
+        tokensRemaining: 0,
+        resetAt: new Date(Date.now() + 10_000),
+      }
+      const allowedResult: ConsumeResult = {
+        allowed: true,
+        tokensRemaining: 9,
+        resetAt: new Date(Date.now() + 60_000),
+      }
+      mockAdapter.consumeTokens.mockResolvedValue(blocked)
+
+      vi.useFakeTimers()
+      try {
+        const promise = rateLimiter.acquireKey(
+          testProvider,
+          envKeyPrefix,
+          perRequestRateLimit,
+          'workspace-1',
+          new AbortController().signal
+        )
+        // Burn well past the 5-minute fallback cap — without a signal this would have 429'd.
+        await vi.advanceTimersByTimeAsync(7 * 60 * 1000)
+        mockAdapter.consumeTokens.mockResolvedValue(allowedResult)
+        await vi.advanceTimersByTimeAsync(HEARTBEAT_REFRESH_INTERVAL_MS)
+        const result = await promise
+
+        expect(result.success).toBe(true)
+        expect(result.key).toBe('test-key-1')
+      } finally {
+        vi.useRealTimers()
+      }
+    })
+
+    it('refreshes the heartbeat during a long low-RPM bucket wait', async () => {
+      // Provider with a long refill (retryAfterMs >> heartbeat TTL). The sleep must be
+      // capped so the heartbeat is renewed and the head is not reaped mid-wait.
+      const blocked: ConsumeResult = {
+        allowed: false,
+        tokensRemaining: 0,
+        resetAt: new Date(Date.now() + 60_000),
+      }
+      const allowedResult: ConsumeResult = {
+        allowed: true,
+        tokensRemaining: 0,
+        resetAt: new Date(Date.now() + 60_000),
+      }
+      mockAdapter.consumeTokens.mockResolvedValue(blocked)
+
+      vi.useFakeTimers()
+      try {
+        const promise = rateLimiter.acquireKey(
+          testProvider,
+          envKeyPrefix,
+          perRequestRateLimit,
+          'workspace-1',
+          new AbortController().signal
+        )
+        await vi.advanceTimersByTimeAsync(3 * HEARTBEAT_REFRESH_INTERVAL_MS)
+        mockAdapter.consumeTokens.mockResolvedValue(allowedResult)
+        await vi.advanceTimersByTimeAsync(HEARTBEAT_REFRESH_INTERVAL_MS)
+        await promise
+
+        expect(mockQueue.refreshHeartbeat).toHaveBeenCalled()
+      } finally {
+        vi.useRealTimers()
+      }
+    })
+  })
+
   describe('acquireKey with custom rate limit', () => {
     const customRateLimit: CustomRateLimit = {
       mode: 'custom',