TL;DR Read Rob's answer, but the iterator code and and the partial download code are still handy so I'm leaving this answer with corrections.

Okay so I spent some time on this because I'm about to try to write my own cancellable url stream object. <strike>and it appears that asyncBytes.task.cancel() is more along the lines of URLSession's finishTasksAndInvalidate() than invalidateAndCancel().</strike> Since you are pointing your streaming task at a file that isn't really that large the URLSessionDataTask had already gotten the bytes in the buffer.

You can see this when you change up the function a bit (see Rob's example as well):

    func test_funcCondition(timeOut:TimeInterval, url:URL, session:URLSession) async throws {
             
        let (asyncBytes, _) = try await session.bytes(from: url)
    
        let deadLine = Date.now + timeOut
        var data = Data()
    
       func someConditionCheck(_ deadline:Date) -> Bool {
            Date.now > deadLine
        }
        
        for try await byte in asyncBytes {   
            if someConditionCheck(deadLine) {  
                asyncBytes.task.cancel()
                print("trying to cancel...")
            } 
            //Wrong type of task! Should not work. if Task.isCancelled { print ("cancelled") }
            data.append(byte) 
            //just to reduce the amount of printing
            if data.count % 100 == 0 {
                print(data.count) 
            }              
        }
    }

If you point the URL at "https://example.com/large_file.zip" like your example and make the time interval very short the function will print "trying to cancel..." between the time your marker hits and the file completes. <strike>It does NOT however, ever print "cancelled".</strike> (The task being cancelled is a URLSessionDataTask, not a Swift concurrency Task, that line never would have worked.)

<strike>If you point either what you wrote or this function at a Server-Sent-Event stream it will cancel out just fine.</strike> (While true, its not in contrast to the other behavior, which also works just fine. There are just bigger pauses in SSE data.)

<strike>If that isn't what you want, if you want to be able to start-stop streams mid-chunk, maybe explore a custom delegate (something I haven't done yet myself), or go work with AVFoundation if that's an option because they've thought a lot about working with large streaming files. I did not check making my own session and running session.invalidateAndCancel() on it instead, because that seems kind of extreme, but may be the way to go if you want to flush the buffer immediately. </strike>

The below will work to stop caring about the buffer immediately. It involves making a custom iterator. <strike>but it seems kind of quirky and may not in fact arrest the downloading (still cost users data rates and power). I haven't looked into how the stream protocol relates to the network protocol on that lower level, if you stop asking does it stop getting? I don't know. </strike> The cancel will arrest the stream allowing through the bytes that are already in the buffer, but your code won't get them. On my todo-list now is to look into how to change buffering policies.

Rob's code seems a nice way to go and advantage of a concurrency Task.

    func test_customIterator(timeOut:TimeInterval, url:URL, session:URLSession) async throws {
             
        let (asyncBytes, _) = try await session.bytes(from: url)
    
        let deadLine = Date.now + timeOut
        var data = Data()
    
       func someConditionCheck(_ deadline:Date) -> Bool {
            Date.now > deadLine
        }
    
        //could also be asyncBytes.lines.makeAsyncIterator(), etc.
        var iterator = asyncBytes.makeAsyncIterator()
        while !someConditionCheck(deadLine) {
            //await Task.yield()
            let byte = try await iterator.next()
            data.append(byte!) 
            print(data.count)            
        }
        //make sure to still tell URLSession you aren't listening anymore.
        //It may auto-close but that's not how I roll.
        asyncBytes.task.cancel()
    
    }

    let tap_out:TimeInterval = 0.0005
    try await test_customIterator(timeOut: tap_out, url: URL(string:"https://example.com/large_file.zip")!, session: URLSession.shared)

<strike>Interesting flavor of behavior. Thanks for pointing it out. Also I didn't know that the task was already available (asyncBytes.task). Thanks for that.</strike> Incorrect. The asyncBytes.task is a URLSessionDataTask not a concurrency Task

UPDATED TO ADD:

To get part of the file explicitly

    //https://developer.mozilla.org/en-US/docs/Web/HTTP/Range_requests
    func requestInChunks(data:inout Data, url:URL, session:URLSession, offset:Int, length:Int) async throws {
        var urlRequest = URLRequest(url: url)
        urlRequest.addValue("bytes=\(offset)-\(offset + length - 1)", forHTTPHeaderField: "Range")
    
        let (asyncBytes, response) = try await
            session.bytes(for: urlRequest, delegate: nil)
    
        guard (response as? HTTPURLResponse)?.statusCode == 206 else { //NOT 200!!
            throw APIngError("The server responded with an error.")
        }
        
        for try await byte in asyncBytes { 
            data.append(byte) 
            if data.count % 100 == 0 {
                print(data.count) 
            }  
        }
    }

Still think if my task on hand was about file downloading session.download would be my go to, but then there is file clean up, etc. so I get why not go there.