Match CLI downloads to Assembly results

.changeset/assembly-download-paths.md

@@ -0,0 +1,11 @@
+---
+'@transloadit/node': minor
+'@transloadit/mcp-server': patch
+'transloadit': minor
+---
+
+Fix CLI downloads to use Assembly result names and extensions, including bundled outputs and type-changing Steps. Preserve the existing input-relative directories and `<input>/<step>/…` layout for ordinary multi-result downloads, along with time-based skipping for directory reruns. Use `--reprocess-stale` to process unchanged inputs again.
+
+Fix HLS downloads by retaining the result directories referenced by playlists, keeping independent source bundles separate, and publishing linked watch updates together. Reserve download paths using the destination root’s case and Unicode naming rules, and avoid file/directory collisions and descendant symlinks.
+
+Add public HTTP(S) URL inputs to `assemblies create` and `run` using temporary uploads that are cleaned up afterward. Each download has a five-minute deadline, configurable in milliseconds with `--url-download-timeout`. Watch mode requires local inputs.

.changeset/clear-assembly-reports.md

@@ -0,0 +1,9 @@
+---
+'@transloadit/node': patch
+'@transloadit/mcp-server': patch
+'transloadit': patch
+---
+
+Report Assembly IDs and URLs in CLI diagnostics and JSON summaries, warn about ignored errors,
+and show concise validation and API errors with the reason for failed Assemblies.
+Validate diagnostic fields before printing them, excluding nested request metadata.

.changeset/close-upload-streams.md

@@ -0,0 +1,7 @@
+---
+'@transloadit/node': patch
+'@transloadit/mcp-server': patch
+'transloadit': patch
+---
+
+Close SDK-opened upload file streams before a failed or aborted `createAssembly()` rejects, so callers can immediately remove temporary files without leaking file handles.

packages/mcp-server/src/server.ts

@@ -1717,9 +1717,7 @@ export const createTransloaditMcpServer = (
               const known = await settleWithin(Promise.race([created, work]), statusReadTimeoutMs)
               if (known === deadlineReached) {
                 // Without the request, nothing can create this Assembly later, so a status check
-                // that finds no Assembly makes creating it again safe. The SDK keeps the read
-                // streams it opened for the files until they are collected, as on any failed
-                // creation; closing them needs an SDK change.
+                // that finds no Assembly makes creating it again safe.
                 creationAbort.abort()
                 return creationAssemblyId
                   ? buildCreationPending(

packages/node/README.md

@@ -146,7 +146,8 @@ npx -y @transloadit/node run "extract a waveform image from this audio" --input
 ```
 
 Repeat `--input` for multiple files, or pass a directory. Use `--single-assembly` to process the
-inputs together. `run` also supports `--watch`, `--recursive`, and `--concurrency`.
+inputs together. `run` also supports `--watch`, `--recursive`, `--concurrency`, and
+`--url-download-timeout` (milliseconds per URL input, default: `300000`).
 
 ### Compile Assembly Instructions
 
@@ -265,6 +266,19 @@ npx -y transloadit auth token --aud mcp --scope assemblies:write,templates:read
 
 ### Processing Media
 
+Assembly runs print their ID and status URL to stderr, including when processing fails.
+`--json` writes newline-delimited JSON summaries for terminal Assembly statuses, with their
+reference, final status, and `ignored_errors`. Ignored errors also produce a warning with their
+count and up to five details.
+If polling fails before a terminal status is retrieved, the error and known Assembly reference
+are reported on stderr.
+With `--print-urls`, JSON remains a single object containing `urls` and `assemblies`.
+This document is written after the batch finishes; an input or watch error that aborts the batch
+can prevent it from being written.
+`--output -` keeps Assembly summaries off stdout; the reference stays on stderr.
+When creating Assemblies, invalid Robot parameters and API failures produce a short diagnostic and
+a nonzero exit status.
+
 For common one-off tasks, prefer the intent-first commands:
 
 The full generated intent reference also lives in [`docs/intent-commands.md`](./docs/intent-commands.md).
@@ -1502,11 +1516,22 @@ The `assemblies create` command additionally supports:
   Without uploads (for example, `/vimeo/import`, `/http/import`, or prompt-only `/image/generate`
   Steps), results are saved as `<dir>/<step>/<result name>`, with no input folder, including with
   `--single-assembly`.
-- `--input, -i` - Provide a local upload, or use `-i -` to read stdin explicitly. When `--steps`
-  is used without `-i`, stdin is read only when it is a pipe or a redirected file. Other non-TTY
-  stdin (such as an agent shell or `/dev/null`) runs the Assembly without uploads. With `--template`,
-  stdin always requires `-i -`.
+  Downloaded filenames come from the Assembly results, including their extensions. Ordinary
+  downloads retain the existing input-relative directories and `<input>/<step>/…` layout for
+  multiple results. Directory inputs retain subfolders relative to that directory. Existing
+  time-based checks skip inputs with newer outputs; use `--reprocess-stale` to process them again.
+  HLS results retain the directories in `meta.relative_path`, so playlists can find their
+  renditions and segments. Multiple HLS bundles in one Step get separate source folders. On
+  collisions, a linked bundle moves together to a new folder, preserving playlist references.
+  Watch updates refresh their existing result files.
+- `--input, -i` - Provide a local upload or a public HTTP(S) URL, or use `-i -` to read stdin explicitly.
+  URLs are downloaded to temporary files for upload and cleaned up after processing; `--watch`
+  requires local inputs. When `--steps` is used without `-i`, stdin is read only when it is a pipe
+  or a redirected file. Other non-TTY stdin (such as an agent shell or `/dev/null`) runs the Assembly
+  without uploads. With `--template`, stdin always requires `-i -`.
 - `--single-assembly` - Pass all input files to a single assembly instead of one assembly per file
+- `--url-download-timeout` - Limit each URL input download to this many milliseconds (default:
+  `300000`, or five minutes). DNS checks and redirects share the same deadline.
 
 #### Log Levels
 

packages/node/src/Transloadit.ts

@@ -1,3 +1,4 @@
+import type { ReadStream } from 'node:fs'
 import type { Readable } from 'node:stream'
 
 import type {
@@ -72,9 +73,11 @@ import type { Stream, UploadBehavior } from './tus.ts'
 
 import * as assert from 'node:assert'
 import { randomUUID } from 'node:crypto'
+import { addAbortListener, setMaxListeners } from 'node:events'
 import { constants, createReadStream } from 'node:fs'
 import { access, stat } from 'node:fs/promises'
 import { basename } from 'node:path'
+import { finished } from 'node:stream/promises'
 import { setTimeout as delay } from 'node:timers/promises'
 
 import { compileAssemblyInstructionsFromPrompt } from '@transloadit/utils'
@@ -241,11 +244,15 @@ const toReadableUpload = (label: string, value: Readable | IntoStreamInput): Rea
   return readable ? value : intoStream(value)
 }
 
-const buildStreamsMap = (descriptors: UploadDescriptor[]): Record<string, Stream> =>
+const buildStreamsMap = (
+  descriptors: UploadDescriptor[],
+  onFileStreamOpened?: (stream: ReadStream) => void,
+): Record<string, Stream> =>
   Object.fromEntries(
     descriptors.map((descriptor) => {
       if (descriptor.path) {
         const stream = createReadStream(descriptor.path)
+        onFileStreamOpened?.(stream)
         return [descriptor.label, { stream, path: descriptor.path }]
       }
 
@@ -705,7 +712,7 @@ export class Transloadit {
       assemblyId,
       expectedUploads,
       onAssemblyCreated,
-      signal,
+      signal: callerSignal,
       uploadBehavior = 'await',
     } = opts
 
@@ -722,6 +729,34 @@ export class Transloadit {
       effectiveAssemblyId = randomUUID().replace(/-/g, '')
     }
     const urlSuffix = `/assemblies/${effectiveAssemblyId}`
+    const uploadAbort = new AbortController()
+    const signal = uploadAbort.signal
+    // This private signal coordinates every concurrent upload and its queue.
+    setMaxListeners(0, signal)
+    if (callerSignal?.aborted) uploadAbort.abort(callerSignal.reason)
+    // Older supported Node releases retain AbortSignal.any() dependencies indefinitely.
+    const callerAbort =
+      callerSignal && !callerSignal.aborted
+        ? addAbortListener(callerSignal, () => uploadAbort.abort(callerSignal.reason))
+        : undefined
+    const detachCallerAbort = (): void => {
+      callerAbort?.[Symbol.dispose]()
+    }
+    let backgroundUploadsComplete: Promise<unknown> | undefined
+    const openedFileStreams = new Set<ReadStream>()
+    const pendingFileSlices = new Set<Promise<unknown>>()
+    const onFileStreamOpened = (stream: ReadStream): void => {
+      // Register before I/O can emit close, so cleanup cannot miss that event.
+      openedFileStreams.add(stream)
+      stream.once('close', () => openedFileStreams.delete(stream))
+    }
+    const onFileSliceStarted = (slice: Promise<unknown>): void => {
+      pendingFileSlices.add(slice)
+      const remove = (): void => {
+        pendingFileSlices.delete(slice)
+      }
+      slice.then(remove, remove)
+    }
 
     // We want to be able to return the promise immediately with custom data
     const promise = (async () => {
@@ -746,7 +781,7 @@ export class Transloadit {
         })),
       ]
 
-      const allStreamsMap = buildStreamsMap(descriptors)
+      const allStreamsMap = buildStreamsMap(descriptors, onFileStreamOpened)
 
       const allStreams = Object.values(allStreamsMap)
 
@@ -779,15 +814,18 @@ export class Transloadit {
         onAssemblyCreated?.(result)
 
         if (Object.keys(allStreamsMap).length > 0) {
-          const { uploadUrls } = await sendTusRequest({
+          const { uploadUrls, uploadsComplete } = await sendTusRequest({
             streamsMap: allStreamsMap,
             assembly: result,
             onProgress: onUploadProgress,
             requestedChunkSize,
             uploadConcurrency,
             signal,
             uploadBehavior,
+            onFileStreamOpened,
+            onFileSliceStarted,
           })
+          if (uploadBehavior === 'background') backgroundUploadsComplete = uploadsComplete
           if (uploadBehavior !== 'await' && Object.keys(uploadUrls).length > 0) {
             result.upload_urls = uploadUrls
           }
@@ -813,6 +851,29 @@ export class Transloadit {
 
       return Promise.race([createAssemblyAndUpload(), streamErrorPromise])
     })()
+      .catch(async (error: unknown) => {
+        uploadAbort.abort()
+        // File descriptors must be closed before callers can remove their temporary files.
+        // Destroyed, unread streams reject finished(); cleanup must preserve the creation error.
+        while (openedFileStreams.size > 0 || pendingFileSlices.size > 0) {
+          await Promise.allSettled(pendingFileSlices)
+          await Promise.allSettled(
+            [...openedFileStreams].map((stream) => {
+              const closed = finished(stream, { cleanup: true })
+              stream.destroy()
+              return closed
+            }),
+          )
+        }
+        throw error
+      })
+      .finally(() => {
+        if (backgroundUploadsComplete && !signal.aborted) {
+          backgroundUploadsComplete.then(detachCallerAbort, detachCallerAbort)
+          return
+        }
+        detachCallerAbort()
+      })
 
     // This allows the user to use or log the assemblyId even before it has been created for easier debugging
     return Object.assign(promise, { assemblyId: effectiveAssemblyId })

packages/node/src/cli/OutputCtl.ts

@@ -60,6 +60,7 @@ export interface IOutputCtl {
   info(msg: unknown): void
   debug(msg: unknown): void
   trace(msg: unknown): void
+  /** An undefined simple value emits JSON only, preserving binary or URL stdout in text mode. */
   print(simple: unknown, json: unknown): void
 }
 
@@ -110,6 +111,6 @@ export default class OutputCtl implements IOutputCtl {
   print(simple: unknown, json: unknown): void {
     if (this.json) console.log(JSON.stringify(json))
     else if (typeof simple === 'string') console.log(simple)
-    else console.dir(simple, { depth: null })
+    else if (simple !== undefined) console.dir(simple, { depth: null })
   }
 }

packages/node/src/cli/commands/BaseCommand.ts

@@ -2,9 +2,14 @@ import type { ResolvedCliConfig } from '../helpers.ts'
 import type { IOutputCtl } from '../OutputCtl.ts'
 
 import { Command, Option } from 'clipanion'
+import { RequestError } from 'got'
+import { z } from 'zod'
 
+import { ApiError } from '../../ApiError.ts'
+import { tryCatch } from '../../alphalib/tryCatch.ts'
 import { Transloadit as TransloaditClient } from '../../Transloadit.ts'
-import { buildMissingAuthMessage, resolveCliConfig } from '../helpers.ts'
+import { CliInputError, formatSchemaIssues } from '../errors.ts'
+import { buildMissingAuthMessage, formatAPIError, resolveCliConfig } from '../helpers.ts'
 import OutputCtl, { LOG_LEVEL_DEFAULT, LOG_LEVEL_NAMES, parseLogLevel } from '../OutputCtl.ts'
 
 abstract class BaseCommand extends Command {
@@ -50,6 +55,27 @@ abstract class BaseCommand extends Command {
   }
 
   abstract override execute(): Promise<number | undefined>
+
+  protected abstract run(): Promise<number | undefined>
+
+  /** Report schema, API, and transport failures once; other errors retain Clipanion's diagnostics. */
+  protected async runWithErrorReporting(): Promise<number | undefined> {
+    const [error, result] = await tryCatch(this.run())
+    if (!error) return result
+    if (error instanceof z.ZodError) {
+      this.output.error(`Invalid options: ${formatSchemaIssues(error.issues)}`)
+      return 1
+    }
+    if (error instanceof RequestError) {
+      this.output.error(formatAPIError(error))
+      return 1
+    }
+    if (error instanceof CliInputError || error instanceof ApiError) {
+      this.output.error(formatAPIError(error))
+      return 1
+    }
+    throw error
+  }
 }
 
 export abstract class AuthenticatedCommand extends BaseCommand {
@@ -58,17 +84,13 @@ export abstract class AuthenticatedCommand extends BaseCommand {
     if (!this.setupClient()) {
       return 1
     }
-    return await this.run()
+    return await this.runWithErrorReporting()
   }
-
-  protected abstract run(): Promise<number | undefined>
 }
 
 export abstract class UnauthenticatedCommand extends BaseCommand {
   override async execute(): Promise<number | undefined> {
     this.setupOutput()
-    return await this.run()
+    return await this.runWithErrorReporting()
   }
-
-  protected abstract run(): Promise<number | undefined>
 }