Function waitForPendingChatStream

apps/sim/lib/copilot/request/session/abort.ts:71–107

(
  chatId: string,
  timeoutMs = 5_000,
  expectedStreamId?: string
)

}

export async function waitForPendingChatStream(
  chatId: string,
  timeoutMs = 5_000,
  expectedStreamId?: string
): Promise<boolean> {
  const redis = getRedisClient()
  const deadline = Date.now() + timeoutMs

  for (;;) {
    const entry = pendingChatStreams.get(chatId)
    const localPending = !!entry && (!expectedStreamId || entry.streamId === expectedStreamId)

    if (redis) {
      try {
        const ownerStreamId = await redis.get(getChatStreamLockKey(chatId))
        const lockReleased =
          !ownerStreamId || (expectedStreamId !== undefined && ownerStreamId !== expectedStreamId)
        if (!localPending && lockReleased) {
          return true
        }
      } catch (error) {
        logger.warn('Failed to inspect chat stream lock while waiting', {
          chatId,
          expectedStreamId,
          error: toError(error).message,
        })
      }
    } else if (!localPending) {
      return true
    }

    if (Date.now() >= deadline) {
      return false
    }
    await sleep(200)
  }
}

export async function getPendingChatStreamId(chatId: string): Promise<string | null> {
  const localEntry = pendingChatStreams.get(chatId)