Merge origin/cli: use got.stream for downloads

Resolved conflict in assemblies-create.ts by adopting
the cleaner got.stream + pipeline approach for downloads
while preserving the concurrency limiting and fresh stream
creation from the local branch.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: kvz

src/cli/assemblies-create.ts

@@ -1,13 +1,13 @@
 import EventEmitter from 'node:events'
 import fs from 'node:fs'
 import fsp from 'node:fs/promises'
-import http from 'node:http'
-import https from 'node:https'
 import path from 'node:path'
 import process from 'node:process'
 import type { Readable, Writable } from 'node:stream'
+import { pipeline } from 'node:stream/promises'
 import tty from 'node:tty'
 import { promisify } from 'node:util'
+import got from 'got'
 import { tryCatch } from '../alphalib/tryCatch.ts'
 import type { StepsInput } from '../alphalib/types/template.ts'
 import type { CreateAssemblyParams } from '../apiTypes.ts'
@@ -785,25 +785,17 @@ export default async function run(
                 }
 
                 outputctl.debug(`DOWNLOADING ${stepResult.name} to ${outPath}`)
-                await new Promise<void>((dlResolve, dlReject) => {
-                  const get = resultUrl.startsWith('https') ? https.get : http.get
-                  const req = get(resultUrl, { signal: abortController.signal }, (res) => {
-                    if (res.statusCode !== 200) {
-                      const msg = `Server returned http status ${res.statusCode}`
-                      outputctl.error(msg)
-                      return dlReject(new Error(msg))
-                    }
-                    const outStream = fs.createWriteStream(outPath)
-                    res.pipe(outStream)
-                    outStream.on('finish', () => dlResolve())
-                    outStream.on('error', dlReject)
-                  })
-                  req.on('error', (err) => {
-                    if (err.name === 'AbortError') return dlResolve()
-                    outputctl.error(err.message)
-                    dlReject(err)
-                  })
-                })
+                const [dlErr] = await tryCatch(
+                  pipeline(
+                    got.stream(resultUrl, { signal: abortController.signal }),
+                    fs.createWriteStream(outPath),
+                  ),
+                )
+                if (dlErr) {
+                  if (dlErr.name === 'AbortError') continue
+                  outputctl.error(dlErr.message)
+                  throw dlErr
+                }
               }
             }
           }
@@ -814,6 +806,7 @@ export default async function run(
               await fsp.unlink(inPath)
             }
           }
+          return assembly
         })()
 
         jobsPromise.add(singleAssemblyPromise)
@@ -912,27 +905,15 @@ export default async function run(
 
           if (outStream != null && resulturl && !superceded) {
             outputctl.debug('DOWNLOADING')
-            await new Promise<void>((dlResolve, dlReject) => {
-              const get = resulturl.startsWith('https') ? https.get : http.get
-              const req = get(resulturl, { signal: abortController.signal }, (res) => {
-                if (res.statusCode !== 200) {
-                  const msg = `Server returned http status ${res.statusCode}`
-                  outputctl.error(msg)
-                  return dlReject(new Error(msg))
-                }
-
-                if (superceded) return dlResolve()
-
-                res.pipe(outStream)
-                outStream.on('finish', () => res.unpipe())
-                res.on('end', () => dlResolve())
-              })
-              req.on('error', (err) => {
-                if (err.name === 'AbortError') return dlResolve()
-                outputctl.error(err.message)
-                dlReject(err)
-              })
-            })
+            const [dlErr] = await tryCatch(
+              pipeline(got.stream(resulturl, { signal: abortController.signal }), outStream),
+            )
+            if (dlErr) {
+              if (dlErr.name !== 'AbortError') {
+                outputctl.error(dlErr.message)
+                throw dlErr
+              }
+            }
           }
 
           outputctl.debug(`COMPLETED ${inPath ?? 'null'} ${outPath ?? 'null'}`)

test/e2e/cli/assemblies.test.ts

@@ -1,3 +1,4 @@
+import crypto from 'node:crypto'
 import fsp from 'node:fs/promises'
 import process from 'node:process'
 import { promisify } from 'node:util'
@@ -230,6 +231,39 @@ describe('assemblies', () => {
       }),
     )
 
+    it(
+      'should download file with correct md5 hash',
+      testCase(async (client) => {
+        const infile = await imgPromise()
+        const steps = await stepsPromise()
+
+        const output = new OutputCtl()
+        const { results } = await assembliesCreate(output, client, {
+          steps,
+          inputs: [infile],
+          output: 'out-md5.jpg',
+        })
+
+        // Get the assembly result to find the expected md5hash
+        // The results array contains assembly statuses
+        const assemblyResult = results[0] as {
+          results?: Record<string, Array<{ md5hash?: string }>>
+        }
+        expect(assemblyResult).to.have.property('results')
+        const resultSteps = Object.values(assemblyResult.results ?? {})
+        expect(resultSteps.length).to.be.greaterThan(0)
+        const firstResult = resultSteps[0]?.[0]
+        expect(firstResult).to.have.property('md5hash')
+        const expectedMd5 = firstResult?.md5hash
+
+        // Calculate md5 of downloaded file
+        const downloadedBuffer = await fsp.readFile('out-md5.jpg')
+        const actualMd5 = crypto.createHash('md5').update(downloadedBuffer).digest('hex')
+
+        expect(actualMd5).to.equal(expectedMd5)
+      }),
+    )
+
     it(
       'should handle multiple inputs',
       testCase(async (client) => {