Centralize CLI flag set setup

docs/cli.md

@@ -149,7 +149,8 @@ curl -X POST http://127.0.0.1:8080/upload \
   --data-binary @bundle.tar.gz
 ```
 
-The upload response is accepted asynchronously:
+The upload response is returned after the archive is staged and validated; the
+destination fan-out continues asynchronously:
 
 ```json
 {"run_id":"reports.20260603T120000Z.abcdef12","status":"accepted"}

docs/config.md

@@ -130,12 +130,15 @@ Accepted upload content types:
 - `application/gzip`
 - `application/x-gzip`
 
-Accepted uploads return:
+Accepted uploads return after the archive is staged and validated:
 
 ```json
 {"run_id":"<id>","status":"accepted"}
 ```
 
+Malformed tar or gzip content and invalid staged bundles are rejected before a
+run id is issued.
+
 The run id can be queried through `GET /runs/<run_id>` while the status record
 is retained in memory. Completed records expire after `server.http.retention`;
 expiration also removes committed staged bundle directories for completed

docs/internal/app.md

@@ -5,7 +5,7 @@
 `internal/app` owns the top-level application use cases. It coordinates
 configuration loading, secret resolution, backend construction, source bundle
 discovery, destination selection, publish planning, publish execution,
-notification handoff, run reporting, and in-memory run coordination.
+notification handoff, run reporting, and upload coordination.
 
 The package is the boundary between callers and lower-level domain packages. It
 does not own manifest validation rules, destination state comparison, storage
@@ -31,13 +31,14 @@ root as a local backend, validates exactly that root bundle, and then uses the
 same destination fan-out path as normal runs.
 
 `Validate` and `Inspect` accept either a local path or one configured pipeline
-source. They share source backend construction with run workflows and never open
-destination backends.
+source. Configured-source mode uses the same runtime config and secret setup as
+run workflows, shares source backend construction, and never opens destination
+backends.
 
-`Serve` is the CLI-facing HTTP upload server entrypoint. It loads config,
-loads the configured secrets directory, resolves upload bearer tokens for
-configured `http_upload` sources, creates an `UploadCoordinator`, binds
-`server.http.bind`, and serves the upload API until its context is cancelled.
+`Serve` is the CLI-facing HTTP upload server entrypoint. It uses the app
+runtime setup, resolves upload bearer tokens for configured `http_upload`
+sources, creates an `UploadCoordinator`, binds `server.http.bind`, and serves
+the upload API until its context is cancelled.
 
 ## Run Reports
 
@@ -58,9 +59,11 @@ failures, return before a complete run report is available.
 
 The app runner:
 
-1. loads config from the supplied path or `config.DefaultConfigPath`;
-2. loads configured secret files into a config-owned environment resolver;
-3. builds the app-level backend factory and transform registry;
+1. builds runtime setup by resolving the config path, loading config, loading
+   configured secret files, and projecting secret-conflict warnings;
+2. builds the app-level backend factory from the config-owned environment
+   resolver;
+3. builds the app-level transform registry;
 4. opens each selected pipeline source backend;
 5. discovers validated source bundles from the source root;
 6. selects source bundles for each destination according to path mapping;
@@ -87,10 +90,10 @@ prepared.
 ## Upload Coordination
 
 `UploadCoordinator` owns in-memory coordination for asynchronous upload
-processing. It admits uploads for configured `http_upload` pipelines, generates
-run IDs, tracks status records, stages accepted archives through
-`internal/ingest`, and executes the selected pipeline through
-`RunPipelineWithLocalSource`.
+processing. It admits uploads for configured `http_upload` pipelines, reserves
+queue capacity before request-body staging, stages and validates archives
+through `internal/ingest`, tracks accepted status records, and executes the
+selected pipeline through `RunPipelineWithLocalSource`.
 
 Upload run IDs use:
 
@@ -111,10 +114,11 @@ The coordinator records these statuses:
 - `expired`
 
 Admission is bounded by `server.http.queue_size`. Full queues are rejected
-before the upload body is staged. Execution is bounded by
+before the upload body is read. Successfully reserved uploads are staged and
+validated before an accepted run record is created. Execution is bounded by
 `server.http.max_concurrency`, and only one upload for a given pipeline may run
-at a time. Later uploads for the same pipeline remain queued until the active
-run finishes.
+at a time. Later accepted uploads for the same pipeline remain queued until the
+active run finishes.
 
 Completed records retain the final run report or error text until
 `server.http.retention` elapses. Expiration removes completed status records and
@@ -137,32 +141,16 @@ pipeline ids, but not token values.
 Routes:
 
 - `GET /healthz`: returns `200` after config, secrets, tokens, coordinator, and route setup succeed.
-- `POST /upload`: accepts authenticated tar and tar.gz archives and returns an accepted run id.
+- `POST /upload`: stages and validates an authenticated tar or tar.gz archive, then returns an accepted run id.
 - `GET /runs/<run_id>`: returns the current in-memory upload status record or `404`.
 
 The upload token maps to exactly one configured pipeline. Producers do not
 submit pipeline ids, and submitted `pipeline` or `pipeline_id` query values are
-rejected. Full queues are rejected before the request body is read. Oversized
-uploads, unsupported content types, invalid bearer tokens, full queues, and
-unknown status records are mapped to stable HTTP status codes without returning
-secret token values.
-
-## Coordination
-
-`PipelineRunCoordinator` wraps `RunPipeline` with in-memory admission control.
-It allows different pipeline IDs to run concurrently and rejects a second active
-run for the same pipeline ID.
-
-Coordinator records contain a run ID, pipeline ID, status, timestamps, completed
-report, and error text when applicable. Active state is memory-only and is
-cleared after success, failure, unknown pipeline ID, or context cancellation.
-
-The admission context is checked before a run is accepted. Once accepted, the
-run uses the coordinator lifetime context, so caller cancellation can stop
-waiting for admission without owning the actual run lifetime.
-
-The coordinator does not queue duplicate runs, persist run records, or define
-transport endpoints.
+rejected. Full queues are rejected before the request body is read. Malformed
+archives and invalid staged bundles are rejected before a run id is issued.
+Oversized uploads, unsupported content types, invalid bearer tokens, full
+queues, and unknown status records are mapped to stable HTTP status codes
+without returning secret token values.
 
 ## Errors
 
@@ -182,10 +170,6 @@ aggregated into one run error after remaining destinations have been attempted.
 Destination diagnostics include pipeline ID, destination ID, backend, and
 bundle path.
 
-`PipelineRunCoordinator` returns `DuplicatePipelineRunError` when the same
-pipeline already has an active run. Callers can detect that condition with
-`IsDuplicatePipelineRun`.
-
 Stdout write errors are returned immediately because the caller's requested
 output stream can no longer be trusted.
 
@@ -193,17 +177,22 @@ output stream can no longer be trusted.
 
 Run helpers are grouped by responsibility:
 
+- `runtime.go`: runtime config path resolution, config loading, secret loading,
+  environment resolver handoff, and secret-conflict warning projection.
 - `run.go`: `Run`, `RunPipeline`, and shared run orchestration.
+- `run_destination.go`: destination-scoped planning, execution, action
+  recording, and failure bookkeeping.
 - `run_output.go`: `RunReport`, action/output records, and text/JSON report projection.
+- `output_projection.go`: shared bundle and manifest-file result projection for
+  command JSON output.
 - `run_summary.go`: summary counters.
 - `run_failures.go`: destination failure aggregation and partial-result detection.
 - `run_selection.go`: destination bundle selection, path mapping decisions, and fixed-path warnings.
 - `run_warnings.go`: secret and SSH warning records.
 - `run_notify.go`: notification event projection and action filtering.
-- `run_coordinator.go`: in-memory run admission, run IDs, status records, and duplicate-run errors.
-- `upload_coordinator.go`: in-memory upload admission, queueing, status tracking, staging handoff, and staged-source execution.
+- `upload_coordinator.go`: in-memory upload admission, queue reservation, staging handoff, status tracking, queueing, and staged-source execution.
 - `upload_http.go`: HTTP upload authentication, routes, JSON response projection, and HTTP error mapping.
-- `serve.go`: config/secrets loading and HTTP server startup.
+- `serve.go`: HTTP server startup.
 - `backends.go`: app-level backend factory wiring.
 - `transforms.go`: app-level transform registry wiring.
 - `source_select.go`: configured-source selection shared by `validate` and `inspect`.
@@ -238,8 +227,8 @@ Before changing app orchestration, inspect tests under:
 - `internal/publish`
 
 Use focused app tests for report structure, single-pipeline execution,
-coordinator admission, warning generation, notification behavior, and
-partial-result aggregation.
+upload admission, warning generation, notification behavior, and partial-result
+aggregation.
 
 ## Invariants
 
@@ -248,7 +237,5 @@ partial-result aggregation.
 - Destination-scoped failures still produce a structured report plus an aggregated error.
 - Dry-run must not mutate destination storage or invoke notifications.
 - `RunPipeline` must use the same run path as `Run` after pipeline selection.
-- Duplicate in-flight runs are rejected only for the same pipeline ID.
-- Different pipeline IDs may run concurrently.
 - Concrete backend and transform registration stays at the app layer.
 - The default notifier is `notify.Noop`.

docs/internal/ingest.md

@@ -6,6 +6,9 @@
 
 ## Archive staging
 
+`ValidateContentType` owns accepted upload content-type policy for archive
+staging callers.
+
 `StageArchive` accepts one upload body, content type, pipeline staging path, run id, and explicit size and file-count limits. It writes the request body to temporary storage while enforcing the configured upload size limit, extracts the archive into temporary local storage, validates the extracted source bundle, and then commits the validated bundle to:
 
 ```text

docs/operations.md

@@ -71,9 +71,9 @@ go run ./cmd/distributor validate --config examples/local-publish.yml --pipeline
 ## HTTP Upload Workflow
 
 `distributor serve` runs the HTTP upload API for pipelines whose source backend
-is `http_upload`. Each upload token maps to one configured pipeline, and each
-accepted archive is staged, validated, and published through the same
-destination fan-out path used by local source runs.
+is `http_upload`. Each upload token maps to one configured pipeline. A valid
+archive is staged and validated before a run id is returned, then published
+through the same destination fan-out path used by local source runs.
 
 Minimal local HTTP upload configuration:
 
@@ -119,7 +119,7 @@ curl -X POST http://127.0.0.1:8080/upload \
   --data-binary @bundle.tar.gz
 ```
 
-Successful admission returns a run id:
+Successful staging and admission returns a run id:
 
 ```json
 {"run_id":"reports.20260603T120000Z.abcdef12","status":"accepted"}
@@ -136,6 +136,9 @@ or error details on failure. Status is memory-only and expires after
 `server.http.retention`; completed staged bundle directories are removed on
 expiry. Restarting the process clears upload status and queue state.
 
+Malformed archives and invalid source bundles are rejected by `POST /upload`
+before a run id is issued.
+
 Use `GET /healthz` for readiness after config and tokens load:
 
 ```sh

docs/roadmap/audit.md

@@ -335,14 +335,12 @@ Risk level:
 
 - Low.
 
-### PipelineRunCoordinator overlaps conceptually with UploadCoordinator
+### Duplicate-run coordination overlaps conceptually with upload coordination
 
 Affected files/packages:
 
-- `internal/app/run_coordinator.go`
 - `internal/app/upload_coordinator.go`
 - `docs/internal/app.md`
-- `internal/app/run_coordinator_test.go`
 - `internal/app/upload_coordinator_test.go`
 
 Duplicated or near-duplicated behavior:
@@ -358,7 +356,9 @@ Why it matters:
 Recommended refactor:
 
 - Do not merge the coordinators now.
-- Review whether `PipelineRunCoordinator` is still needed as an exported app-level helper. If it is intended for future transports, document that role clearly. If not, remove it and its tests in a separate dead-code cleanup.
+- Review whether duplicate-run coordination is still needed as an exported
+  app-level helper. If it is intended for future transports, document that role
+  clearly. If not, remove it and its tests in a separate dead-code cleanup.
 - If both remain, extract only tiny shared timestamp/status helpers if a real third coordinator appears.
 
 Suggested tests:
@@ -474,7 +474,8 @@ Progress/status handling:
 
 - `RunReport` is the core run result model and supports JSON partial-result output.
 - HTTP upload status is memory-only and documented as such.
-- `PipelineRunCoordinator` and `UploadCoordinator` overlap conceptually but have different policies. Avoid merging unless product behavior converges.
+- Duplicate-run coordination and upload coordination overlap conceptually but
+  have different policies. Avoid merging unless product behavior converges.
 
 Gaps:
 
@@ -527,7 +528,7 @@ Avoid these changes in the cleanup pass:
    - Only centralize code if the helper does not blur archive directory semantics.
 
 7. Coordinator intent cleanup.
-   - Decide whether `PipelineRunCoordinator` is retained for internal future use.
+   - Decide whether duplicate-run coordination is retained for internal future use.
    - If retained, clarify comments/docs. If removed, do it as a separate dead-code commit.
 
 8. Test helper cleanup.

docs/roadmap/cleanup.md

@@ -290,20 +290,19 @@ Completion criteria:
 
 Goal:
 
-Remove the currently unused internal `PipelineRunCoordinator` to avoid
-maintaining two similar coordination concepts.
+Remove the currently unused duplicate-run coordinator to avoid maintaining two
+similar coordination concepts.
 
 Implementation scope:
 
-- Delete `PipelineRunCoordinator`, `PipelineRunRecord`,
-  `DuplicatePipelineRunError`, related helpers, and their tests.
+- Delete the duplicate-run coordinator, its run record and duplicate-run error
+  types, related helpers, and their tests.
 - Remove or rewrite `docs/internal/app.md` sections that describe the removed
   coordinator.
 - Keep `UploadCoordinator`; do not merge upload queueing with the removed
   duplicate-run coordinator.
-- Before deletion, confirm with `rg` that production code does not reference
-  `NewPipelineRunCoordinator`, `PipelineRunCoordinator`, or
-  `DuplicatePipelineRunError`.
+- Before deletion, confirm with search that production code does not reference
+  the duplicate-run coordinator constructor, type, or error.
 
 Current-behavior documentation updates:
 
@@ -313,7 +312,7 @@ Current-behavior documentation updates:
 Tests:
 
 - `go test ./internal/app ./internal/cli`
-- `rg -n "PipelineRunCoordinator|NewPipelineRunCoordinator|DuplicatePipelineRunError" internal docs`
+- Search `internal` and `docs` for the removed duplicate-run coordinator symbols;
   should show no stale references after removal.
 
 Completion criteria:
@@ -387,7 +386,7 @@ Recommended consistency checks:
 rg -n "LoadFile\\(|LoadSecretEnvironment\\(|DefaultConfigPath" internal/app internal/cli
 rg -n "application/x-tar|application/gzip|application/x-gzip" internal docs
 rg -n "2006-01-02T15:04:05Z07:00" internal pkg
-rg -n "PipelineRunCoordinator|NewPipelineRunCoordinator|DuplicatePipelineRunError" internal docs
+rg -n "duplicate-run coordinator" internal docs
 ```
 
 The cleanup is complete when:

docs/troubleshooting.md

@@ -102,6 +102,24 @@ curl -i -X POST http://127.0.0.1:8080/upload \
 Safe fix: use the token value resolved by the configured `token_env`. Do not
 include token values in logs or tickets.
 
+## `POST /upload` returns `400`
+
+Likely cause: the archive content is malformed, the gzip body is invalid, the
+tar body cannot be extracted safely, or the extracted source bundle fails
+manifest and file validation.
+
+Diagnostic:
+
+```sh
+tar -tf bundle.tar
+tar -tzf bundle.tar.gz
+go run ./cmd/distributor validate <extracted-bundle-root>
+```
+
+Safe fix: rebuild the tar or tar.gz archive from one complete source bundle
+root. The archive must contain exactly one root-level `manifest.json`, and every
+manifest-listed file must exist as a regular file with matching size and digest.
+
 ## `POST /upload` returns `413`
 
 Likely cause: the request body exceeds the selected pipeline's

internal/app/inspect.go

@@ -65,55 +65,19 @@ func writeInspectResult(options InspectOptions, selection sourceSelection) error
 }
 
 type inspectResult struct {
-	PipelineID    string                `json:"pipeline_id,omitempty"`
-	SourceBackend string                `json:"source_backend,omitempty"`
-	BundleCount   int                   `json:"bundle_count"`
-	Bundles       []inspectBundleResult `json:"bundles"`
-}
-
-type inspectBundleResult struct {
-	Path      string              `json:"path"`
-	ID        string              `json:"id"`
-	Created   string              `json:"created"`
-	Digest    string              `json:"digest"`
-	FileCount int                 `json:"file_count"`
-	TotalSize int64               `json:"total_size"`
-	Files     []inspectFileResult `json:"files"`
-}
-
-type inspectFileResult struct {
-	Path   string `json:"path"`
-	SHA256 string `json:"sha256"`
-	Size   int64  `json:"size"`
+	PipelineID    string               `json:"pipeline_id,omitempty"`
+	SourceBackend string               `json:"source_backend,omitempty"`
+	BundleCount   int                  `json:"bundle_count"`
+	Bundles       []bundleDetailResult `json:"bundles"`
 }
 
 func inspectResultFromSelection(selection sourceSelection) inspectResult {
-	result := inspectResult{
+	return inspectResult{
 		PipelineID:    selection.PipelineID,
 		SourceBackend: selection.SourceBackend,
 		BundleCount:   len(selection.Bundles),
-		Bundles:       make([]inspectBundleResult, 0, len(selection.Bundles)),
+		Bundles:       bundleDetailsFromBundles(selection.Bundles),
 	}
-	for _, sourceBundle := range selection.Bundles {
-		bundleResult := inspectBundleResult{
-			Path:      storage.DisplayPath(sourceBundle.RootRelativePath),
-			ID:        sourceBundle.Manifest.ID,
-			Created:   sourceBundle.Manifest.Created.Format("2006-01-02T15:04:05Z07:00"),
-			Digest:    sourceBundle.Manifest.Digest,
-			FileCount: len(sourceBundle.Manifest.Files),
-			Files:     make([]inspectFileResult, 0, len(sourceBundle.Manifest.Files)),
-		}
-		for _, file := range sourceBundle.Manifest.Files {
-			bundleResult.TotalSize += file.Size
-			bundleResult.Files = append(bundleResult.Files, inspectFileResult{
-				Path:   file.Path,
-				SHA256: file.SHA256,
-				Size:   file.Size,
-			})
-		}
-		result.Bundles = append(result.Bundles, bundleResult)
-	}
-	return result
 }
 
 func writeInspection(w io.Writer, selection sourceSelection) error {
@@ -134,7 +98,7 @@ func writeInspection(w io.Writer, selection sourceSelection) error {
 			"- path=%s id=%s created=%s digest=%s files=%d\n",
 			storage.DisplayPath(sourceBundle.RootRelativePath),
 			sourceBundle.Manifest.ID,
-			sourceBundle.Manifest.Created.Format("2006-01-02T15:04:05Z07:00"),
+			formatManifestCreated(sourceBundle.Manifest.Created),
 			sourceBundle.Manifest.Digest,
 			len(sourceBundle.Manifest.Files),
 		); err != nil {

internal/app/inspect_test.go

@@ -3,9 +3,12 @@ package app
 import (
 	"bytes"
 	"context"
+	"encoding/json"
+	"os"
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
 )
@@ -85,6 +88,98 @@ func TestInspectConfiguredSourceJSON(t *testing.T) {
 	}
 }
 
+func TestInspectJSONPreservesCreatedOffsetAndFileMetadata(t *testing.T) {
+	sourceRoot := t.TempDir()
+	created := time.Date(2026, 6, 1, 6, 30, 0, 0, time.FixedZone("CDT", -5*60*60))
+	testutil.WriteSourceBundle(t, sourceRoot, "daily", testutil.BundleOptions{
+		ID:      "reports.offset",
+		Created: created,
+		Files: []testutil.SourceFile{
+			{Path: "report.md", Data: "# Report\n"},
+		},
+	})
+	var stdout bytes.Buffer
+
+	err := Inspect(context.Background(), InspectOptions{
+		Path:         sourceRoot,
+		Stdout:       &stdout,
+		OutputFormat: OutputFormatJSON,
+	})
+
+	if err != nil {
+		t.Fatalf("Inspect() error = %v", err)
+	}
+	result := decodeAppResult(t, stdout.String())
+	bundles, ok := result["bundles"].([]any)
+	if !ok || len(bundles) != 1 {
+		t.Fatalf("bundles = %#v, want one bundle", result["bundles"])
+	}
+	bundle, ok := bundles[0].(map[string]any)
+	if !ok {
+		t.Fatalf("bundle = %#v, want object", bundles[0])
+	}
+	if bundle["created"] != "2026-06-01T06:30:00-05:00" || bundle["file_count"] != float64(1) {
+		t.Fatalf("bundle = %#v, want offset timestamp and file count", bundle)
+	}
+	files, ok := bundle["files"].([]any)
+	if !ok || len(files) != 1 {
+		t.Fatalf("files = %#v, want one file", bundle["files"])
+	}
+	file, ok := files[0].(map[string]any)
+	if !ok || file["path"] != "report.md" || file["sha256"] == "" || file["size"] != float64(9) {
+		t.Fatalf("file = %#v, want projected file metadata", file)
+	}
+}
+
+func TestInspectConfiguredSourceJSONIncludesSecretConflictWarningWithoutValues(t *testing.T) {
+	name := "DISTRIBUTOR_TEST_INSPECT_SECRET"
+	t.Setenv(name, "process-value")
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	secretsRoot := t.TempDir()
+	if err := os.WriteFile(filepath.Join(secretsRoot, name), []byte("secret-value\n"), 0o600); err != nil {
+		t.Fatalf("write secret: %v", err)
+	}
+	testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{ID: "reports.json"})
+	configPath := writeConfigFile(t, `
+secrets:
+  directory: `+secretsRoot+`
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: `+sourceRoot+`
+    destinations:
+      - id: archive
+        backend: local
+        path: `+destinationRoot+`
+`)
+
+	var stdout bytes.Buffer
+	err := Inspect(context.Background(), InspectOptions{
+		ConfigPath:   configPath,
+		PipelineID:   "reports",
+		Stdout:       &stdout,
+		OutputFormat: OutputFormatJSON,
+	})
+	if err != nil {
+		t.Fatalf("Inspect() error = %v", err)
+	}
+	var envelope struct {
+		Warnings []OutputWarning `json:"warnings"`
+	}
+	if err := json.Unmarshal(stdout.Bytes(), &envelope); err != nil {
+		t.Fatalf("decode output: %v; output = %q", err, stdout.String())
+	}
+	if len(envelope.Warnings) != 1 || !strings.Contains(envelope.Warnings[0].Message, "secret "+name+" ignored") {
+		t.Fatalf("warnings = %#v, want secret conflict warning", envelope.Warnings)
+	}
+	output := stdout.String()
+	if strings.Contains(output, "process-value") || strings.Contains(output, "secret-value") {
+		t.Fatalf("stdout exposed secret values: %q", output)
+	}
+}
+
 func TestInspectRequiresPath(t *testing.T) {
 	err := Inspect(context.Background(), InspectOptions{})
 	if err == nil || !strings.Contains(err.Error(), "requires a path") {

internal/app/manifest.go

@@ -92,37 +92,23 @@ func normalizeManifestFiles(files []string) []string {
 }
 
 type manifestCreateResult struct {
-	ManifestPath string                     `json:"manifest_path"`
-	Root         string                     `json:"root"`
-	ID           string                     `json:"id"`
-	Created      string                     `json:"created"`
-	Digest       string                     `json:"digest"`
-	FileCount    int                        `json:"file_count"`
-	Files        []manifestCreateFileResult `json:"files"`
-}
-
-type manifestCreateFileResult struct {
-	Path   string `json:"path"`
-	SHA256 string `json:"sha256"`
-	Size   int64  `json:"size"`
+	ManifestPath string               `json:"manifest_path"`
+	Root         string               `json:"root"`
+	ID           string               `json:"id"`
+	Created      string               `json:"created"`
+	Digest       string               `json:"digest"`
+	FileCount    int                  `json:"file_count"`
+	Files        []manifestFileResult `json:"files"`
 }
 
 func manifestCreateResultFromManifest(root string, manifest producerbundle.Manifest) manifestCreateResult {
-	result := manifestCreateResult{
+	return manifestCreateResult{
 		ManifestPath: filepath.ToSlash(filepath.Join(root, producerbundle.ManifestName)),
 		Root:         filepath.ToSlash(root),
 		ID:           manifest.ID,
-		Created:      manifest.Created.Format(time.RFC3339),
+		Created:      formatManifestCreated(manifest.Created),
 		Digest:       manifest.Digest,
 		FileCount:    len(manifest.Files),
-		Files:        make([]manifestCreateFileResult, 0, len(manifest.Files)),
+		Files:        manifestFileResults(manifest.Files),
 	}
-	for _, file := range manifest.Files {
-		result.Files = append(result.Files, manifestCreateFileResult{
-			Path:   file.Path,
-			SHA256: file.SHA256,
-			Size:   file.Size,
-		})
-	}
-	return result
 }

internal/app/manifest_test.go

@@ -0,0 +1,42 @@
+package app
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestManifestCreateJSONPreservesCreatedOffsetAndFileMetadata(t *testing.T) {
+	root := t.TempDir()
+	if err := os.WriteFile(filepath.Join(root, "report.md"), []byte("# Report\n"), 0o600); err != nil {
+		t.Fatalf("write report: %v", err)
+	}
+	var stdout bytes.Buffer
+
+	err := ManifestCreate(context.Background(), ManifestCreateOptions{
+		Root:         root,
+		ID:           "reports.offset",
+		Created:      "2026-06-01T06:30:00-05:00",
+		Files:        []string{"report.md"},
+		Stdout:       &stdout,
+		OutputFormat: OutputFormatJSON,
+	})
+
+	if err != nil {
+		t.Fatalf("ManifestCreate() error = %v", err)
+	}
+	result := decodeAppResult(t, stdout.String())
+	if result["id"] != "reports.offset" || result["created"] != "2026-06-01T06:30:00-05:00" || result["file_count"] != float64(1) {
+		t.Fatalf("result = %#v, want manifest metadata", result)
+	}
+	files, ok := result["files"].([]any)
+	if !ok || len(files) != 1 {
+		t.Fatalf("files = %#v, want one file", result["files"])
+	}
+	file, ok := files[0].(map[string]any)
+	if !ok || file["path"] != "report.md" || file["sha256"] == "" || file["size"] != float64(9) {
+		t.Fatalf("file = %#v, want projected file metadata", file)
+	}
+}

internal/app/output_projection.go

@@ -0,0 +1,83 @@
+package app
+
+import (
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+)
+
+type bundleSummaryResult struct {
+	Path string `json:"path"`
+	ID   string `json:"id"`
+}
+
+type bundleDetailResult struct {
+	Path      string               `json:"path"`
+	ID        string               `json:"id"`
+	Created   string               `json:"created"`
+	Digest    string               `json:"digest"`
+	FileCount int                  `json:"file_count"`
+	TotalSize int64                `json:"total_size"`
+	Files     []manifestFileResult `json:"files"`
+}
+
+type manifestFileResult struct {
+	Path   string `json:"path"`
+	SHA256 string `json:"sha256"`
+	Size   int64  `json:"size"`
+}
+
+func bundleSummaryFromBundle(sourceBundle bundle.Bundle) bundleSummaryResult {
+	return bundleSummaryResult{
+		Path: storage.DisplayPath(sourceBundle.RootRelativePath),
+		ID:   sourceBundle.Manifest.ID,
+	}
+}
+
+func bundleSummariesFromBundles(sourceBundles []bundle.Bundle) []bundleSummaryResult {
+	results := make([]bundleSummaryResult, 0, len(sourceBundles))
+	for _, sourceBundle := range sourceBundles {
+		results = append(results, bundleSummaryFromBundle(sourceBundle))
+	}
+	return results
+}
+
+func bundleDetailFromBundle(sourceBundle bundle.Bundle) bundleDetailResult {
+	result := bundleDetailResult{
+		Path:      storage.DisplayPath(sourceBundle.RootRelativePath),
+		ID:        sourceBundle.Manifest.ID,
+		Created:   formatManifestCreated(sourceBundle.Manifest.Created),
+		Digest:    sourceBundle.Manifest.Digest,
+		FileCount: len(sourceBundle.Manifest.Files),
+		Files:     manifestFileResults(sourceBundle.Manifest.Files),
+	}
+	for _, file := range sourceBundle.Manifest.Files {
+		result.TotalSize += file.Size
+	}
+	return result
+}
+
+func bundleDetailsFromBundles(sourceBundles []bundle.Bundle) []bundleDetailResult {
+	results := make([]bundleDetailResult, 0, len(sourceBundles))
+	for _, sourceBundle := range sourceBundles {
+		results = append(results, bundleDetailFromBundle(sourceBundle))
+	}
+	return results
+}
+
+func manifestFileResults(files []bundle.ManifestFile) []manifestFileResult {
+	results := make([]manifestFileResult, 0, len(files))
+	for _, file := range files {
+		results = append(results, manifestFileResult{
+			Path:   file.Path,
+			SHA256: file.SHA256,
+			Size:   file.Size,
+		})
+	}
+	return results
+}
+
+func formatManifestCreated(created time.Time) string {
+	return created.Format(time.RFC3339)
+}

internal/app/run.go

@@ -8,7 +8,6 @@ import (
 	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
 	"gitea.maximumdirect.net/eric/distributor/internal/config"
 	"gitea.maximumdirect.net/eric/distributor/internal/notify"
-	"gitea.maximumdirect.net/eric/distributor/internal/publish"
 	"gitea.maximumdirect.net/eric/distributor/internal/storage"
 )
 
@@ -46,15 +45,11 @@ func Run(ctx context.Context, options RunOptions) error {
 		return err
 	}
 
-	configPath := options.ConfigPath
-	if configPath == "" {
-		configPath = config.DefaultConfigPath
-	}
-	cfg, err := config.LoadFile(configPath)
+	setup, err := loadRuntimeSetup(options.ConfigPath)
 	if err != nil {
 		return err
 	}
-	return runConfig(ctx, cfg, options)
+	return runSetup(ctx, setup, options)
 }
 
 func RunPipeline(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
@@ -62,15 +57,11 @@ func RunPipeline(ctx context.Context, options RunPipelineOptions) (RunReport, er
 		return RunReport{}, err
 	}
 
-	configPath := options.ConfigPath
-	if configPath == "" {
-		configPath = config.DefaultConfigPath
-	}
-	cfg, err := config.LoadFile(configPath)
+	setup, err := loadRuntimeSetup(options.ConfigPath)
 	if err != nil {
 		return RunReport{}, err
 	}
-	return runPipelineConfig(ctx, cfg, options)
+	return runPipelineSetup(ctx, setup, options)
 }
 
 func RunPipelineWithLocalSource(ctx context.Context, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
@@ -81,57 +72,81 @@ func RunPipelineWithLocalSource(ctx context.Context, options RunPipelineWithLoca
 		return RunReport{}, fmt.Errorf("source root is required")
 	}
 
-	configPath := options.ConfigPath
-	if configPath == "" {
-		configPath = config.DefaultConfigPath
-	}
-	cfg, err := config.LoadFile(configPath)
+	setup, err := loadRuntimeSetup(options.ConfigPath)
 	if err != nil {
 		return RunReport{}, err
 	}
-	return runPipelineConfigWithLocalSource(ctx, cfg, options)
+	return runPipelineSetupWithLocalSource(ctx, setup, options)
 }
 
 func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error {
-	return runConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment)
+	setup, err := runtimeSetupFromConfig("", cfg)
+	if err != nil {
+		return err
+	}
+	return runSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
 }
 
 type backendFactoryProvider func(config.Environment) *backendFactory
 
 func runPipelineConfig(ctx context.Context, cfg config.Config, options RunPipelineOptions) (RunReport, error) {
-	return runPipelineConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment)
+	setup, err := runtimeSetupFromConfig("", cfg)
+	if err != nil {
+		return RunReport{}, err
+	}
+	return runPipelineSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
 }
 
 func runPipelineConfigWithLocalSource(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
-	return runPipelineConfigWithLocalSourceAndBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment)
+	setup, err := runtimeSetupFromConfig("", cfg)
+	if err != nil {
+		return RunReport{}, err
+	}
+	return runPipelineSetupWithLocalSourceAndBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
 }
 
 func runPipelineConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunPipelineOptions, provider backendFactoryProvider) (RunReport, error) {
-	pipeline, ok := findPipeline(cfg, options.PipelineID)
+	setup, err := runtimeSetupFromConfig("", cfg)
+	if err != nil {
+		return RunReport{}, err
+	}
+	return runPipelineSetupWithBackendFactory(ctx, setup, options, provider)
+}
+
+func runPipelineSetup(ctx context.Context, setup runtimeSetup, options RunPipelineOptions) (RunReport, error) {
+	return runPipelineSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
+}
+
+func runPipelineSetupWithBackendFactory(ctx context.Context, setup runtimeSetup, options RunPipelineOptions, provider backendFactoryProvider) (RunReport, error) {
+	pipeline, ok := findPipeline(setup.Config, options.PipelineID)
 	if !ok {
 		return RunReport{}, PipelineNotFoundError{ID: options.PipelineID}
 	}
-	return buildRunReportWithBackendFactory(ctx, config.Config{
-		Server:    cfg.Server,
-		Secrets:   cfg.Secrets,
-		Pipelines: []config.Pipeline{pipeline},
-	}, RunOptions{
+	return buildRunReportWithSetup(ctx, setup.withPipelines([]config.Pipeline{pipeline}), RunOptions{
 		DryRun:   options.DryRun,
 		Force:    options.Force,
 		Notifier: options.Notifier,
-	}, provider)
+	}, provider, nil)
 }
 
 func runPipelineConfigWithLocalSourceAndBackendFactory(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions, provider backendFactoryProvider) (RunReport, error) {
-	pipeline, ok := findPipeline(cfg, options.PipelineID)
+	setup, err := runtimeSetupFromConfig("", cfg)
+	if err != nil {
+		return RunReport{}, err
+	}
+	return runPipelineSetupWithLocalSourceAndBackendFactory(ctx, setup, options, provider)
+}
+
+func runPipelineSetupWithLocalSource(ctx context.Context, setup runtimeSetup, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+	return runPipelineSetupWithLocalSourceAndBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
+}
+
+func runPipelineSetupWithLocalSourceAndBackendFactory(ctx context.Context, setup runtimeSetup, options RunPipelineWithLocalSourceOptions, provider backendFactoryProvider) (RunReport, error) {
+	pipeline, ok := findPipeline(setup.Config, options.PipelineID)
 	if !ok {
 		return RunReport{}, PipelineNotFoundError{ID: options.PipelineID}
 	}
-	return buildRunReport(ctx, config.Config{
-		Server:    cfg.Server,
-		Secrets:   cfg.Secrets,
-		Pipelines: []config.Pipeline{pipeline},
-	}, RunOptions{
+	return buildRunReportWithSetup(ctx, setup.withPipelines([]config.Pipeline{pipeline}), RunOptions{
 		DryRun:   options.DryRun,
 		Force:    options.Force,
 		Notifier: options.Notifier,
@@ -142,7 +157,19 @@ func runPipelineConfigWithLocalSourceAndBackendFactory(ctx context.Context, cfg
 }
 
 func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) error {
-	report, err := buildRunReportWithBackendFactory(ctx, cfg, options, provider)
+	setup, err := runtimeSetupFromConfig("", cfg)
+	if err != nil {
+		return err
+	}
+	return runSetupWithBackendFactory(ctx, setup, options, provider)
+}
+
+func runSetup(ctx context.Context, setup runtimeSetup, options RunOptions) error {
+	return runSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
+}
+
+func runSetupWithBackendFactory(ctx context.Context, setup runtimeSetup, options RunOptions, provider backendFactoryProvider) error {
+	report, err := buildRunReportWithSetup(ctx, setup, options, provider, nil)
 	if err != nil && !IsPartialResultError(err) {
 		return err
 	}
@@ -153,7 +180,11 @@ func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options
 }
 
 func buildRunReportWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) (RunReport, error) {
-	return buildRunReport(ctx, cfg, options, provider, nil)
+	setup, err := runtimeSetupFromConfig("", cfg)
+	if err != nil {
+		return RunReport{}, err
+	}
+	return buildRunReportWithSetup(ctx, setup, options, provider, nil)
 }
 
 type localSourceRoot struct {
@@ -162,6 +193,14 @@ type localSourceRoot struct {
 }
 
 func buildRunReport(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider, sourceRoot *localSourceRoot) (RunReport, error) {
+	setup, err := runtimeSetupFromConfig("", cfg)
+	if err != nil {
+		return RunReport{}, err
+	}
+	return buildRunReportWithSetup(ctx, setup, options, provider, sourceRoot)
+}
+
+func buildRunReportWithSetup(ctx context.Context, setup runtimeSetup, options RunOptions, provider backendFactoryProvider, sourceRoot *localSourceRoot) (RunReport, error) {
 	notifier := options.Notifier
 	if notifier == nil {
 		notifier = notify.Noop{}
@@ -173,17 +212,17 @@ func buildRunReport(ctx context.Context, cfg config.Config, options RunOptions,
 		Actions:   []RunActionRecord{},
 	}
 	var failures runFailures
-	secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil)
-	if err != nil {
-		return report, err
+	recorder := runReportRecorder{
+		report:   &report,
+		summary:  &summary,
+		failures: &failures,
 	}
-	secretWarnings := secretConflictWarnings(secretLoad.Conflicts)
-	report.PreambleWarnings = append(report.PreambleWarnings, secretWarnings...)
-	report.addWarnings(secretWarnings)
-	backends := provider(secretLoad.Environment)
+	report.PreambleWarnings = append(report.PreambleWarnings, setup.Warnings...)
+	report.addWarnings(setup.Warnings)
+	backends := provider(setup.Environment)
 	backends.readOnlyKnownHosts = options.DryRun
 	transforms := newTransformRegistry()
-	for _, pipeline := range cfg.Pipelines {
+	for _, pipeline := range setup.Config.Pipelines {
 		pipelineWarnings := sshWarnings(pipeline)
 		report.addWarnings(pipelineWarnings)
 		sourceBackend, bundles, sourceBackendName, err := openPipelineSource(ctx, backends, pipeline, sourceRoot)
@@ -199,103 +238,18 @@ func buildRunReport(ctx context.Context, cfg config.Config, options RunOptions,
 		})
 		pipelineIndex := len(report.Pipelines) - 1
 		for _, destination := range pipeline.Destinations {
-			selections := selectDestinationBundles(destination, bundles)
-			if isFixedPathDestination(destination) {
-				summary.recordFixedPath()
-				if options.DryRun {
-					warning := fixedPathSelectionWarning(pipeline.ID, destination.ID, selections, len(bundles))
-					report.addWarning(warning)
-					report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, warningEvent(warning))
-				}
-			}
-			if len(selections) == 0 {
-				continue
-			}
-			destinationBackend, err := backends.openDestination(ctx, destination)
-			if err != nil {
-				for _, selection := range selections {
-					failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(selection.SourceBundle.RootRelativePath), err)
-					summary.recordFailure()
-					report.Actions = append(report.Actions, errorAction(pipeline.ID, destination.ID, destination.Backend, selection.SourceBundle.RootRelativePath, err))
-					report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, actionEvent(len(report.Actions)-1))
-				}
-				continue
-			}
-			closeDestination := true
-			deferCloseDestination := func() {
-				if closeDestination {
-					closeBackend(destinationBackend)
-					closeDestination = false
-				}
-			}
-			for _, selection := range selections {
-				sourceBundle := selection.SourceBundle
-				req := publish.Request{
-					PipelineID:            pipeline.ID,
-					DestinationID:         destination.ID,
-					SourceBundle:          sourceBundle,
-					SourceBackend:         sourceBackend,
-					DestinationBackend:    destinationBackend,
-					DestinationBundlePath: selection.DestinationBundlePath,
-					PathMapping:           destination.PathMap.Mode,
-					Publish:               *destination.Publish,
-					Transform:             destination.Transform,
-					Links:                 destination.Links,
-					Transformers:          transforms,
-					Transfer:              destination.Transfer,
-					DistributorVersion:    Version,
-					Force:                 options.Force,
-				}
-				plan, err := publish.Build(ctx, req)
-				if err != nil {
-					if plan.PipelineID == "" {
-						plan.PipelineID = pipeline.ID
-					}
-					if plan.DestinationID == "" {
-						plan.DestinationID = destination.ID
-					}
-					if plan.BundleID == "" {
-						plan.BundleID = sourceBundle.Manifest.ID
-					}
-					if plan.BundlePath == "" {
-						plan.BundlePath = sourceBundle.RootRelativePath
-					}
-					if plan.DestinationBundlePath == "" {
-						plan.DestinationBundlePath = selection.DestinationBundlePath
-					}
-				}
-				if isFixedPathDestination(destination) {
-					plan.PathMapping = config.PathMappingFixed
-					if options.DryRun && isDestructiveFixedPathAction(plan.Action) {
-						warning := fixedPathReplacementWarning(plan)
-						report.addWarning(warning)
-						report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, warningEvent(warning))
-					}
-				}
-				report.Actions = append(report.Actions, runActionFromPlan(destination.Backend, plan, err))
-				report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, actionEvent(len(report.Actions)-1))
-				if err != nil {
-					failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
-					summary.recordFailure()
-					continue
-				}
-				summary.recordPlan(plan.Action)
-				if !options.DryRun {
-					if err := publish.Execute(ctx, req, plan); err != nil {
-						failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
-						summary.recordFailure()
-						continue
-					}
-					if shouldNotify(plan.Action) {
-						if err := notifier.Notify(ctx, notifyEvent(plan)); err != nil {
-							failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
-							summary.recordFailure()
-							continue
-						}
-					}
-				}
-			}
-			deferCloseDestination()
+			processDestination(ctx, runDestinationRequest{
+				options:       options,
+				notifier:      notifier,
+				backends:      backends,
+				transforms:    transforms,
+				pipeline:      pipeline,
+				pipelineIndex: pipelineIndex,
+				sourceBackend: sourceBackend,
+				bundles:       bundles,
+				destination:   destination,
+				recorder:      &recorder,
+			})
 		}
 		closeBackend(sourceBackend)
 	}
@@ -307,6 +261,12 @@ func buildRunReport(ctx context.Context, cfg config.Config, options RunOptions,
 	return report, nil
 }
 
+type runReportRecorder struct {
+	report   *RunReport
+	summary  *runSummary
+	failures *runFailures
+}
+
 func openPipelineSource(ctx context.Context, backends *backendFactory, pipeline config.Pipeline, sourceRoot *localSourceRoot) (storage.Backend, []bundle.Bundle, string, error) {
 	if sourceRoot != nil && sourceRoot.pipelineID == pipeline.ID {
 		sourceBackend, err := backends.openLocalPath(ctx, sourceRoot.root)

internal/app/run_coordinator.go

@@ -1,131 +0,0 @@
-package app
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"sync"
-	"time"
-)
-
-type PipelineRunID string
-
-type PipelineRunStatus string
-
-const (
-	PipelineRunRunning   PipelineRunStatus = "running"
-	PipelineRunSucceeded PipelineRunStatus = "succeeded"
-	PipelineRunFailed    PipelineRunStatus = "failed"
-)
-
-type PipelineRunRecord struct {
-	ID         PipelineRunID     `json:"id"`
-	PipelineID string            `json:"pipeline_id"`
-	Status     PipelineRunStatus `json:"status"`
-	StartedAt  time.Time         `json:"started_at"`
-	FinishedAt *time.Time        `json:"finished_at,omitempty"`
-	Report     RunReport         `json:"report,omitempty"`
-	Error      string            `json:"error,omitempty"`
-}
-
-type DuplicatePipelineRunError struct {
-	PipelineID string
-	RunID      PipelineRunID
-}
-
-func (err DuplicatePipelineRunError) Error() string {
-	if err.RunID == "" {
-		return fmt.Sprintf("pipeline %q already has an active run", err.PipelineID)
-	}
-	return fmt.Sprintf("pipeline %q already has active run %s", err.PipelineID, err.RunID)
-}
-
-func IsDuplicatePipelineRun(err error) bool {
-	var duplicate DuplicatePipelineRunError
-	return errors.As(err, &duplicate)
-}
-
-type PipelineRunCoordinator struct {
-	ctx    context.Context
-	run    pipelineRunFunc
-	now    func() time.Time
-	mu     sync.Mutex
-	nextID uint64
-	active map[string]PipelineRunRecord
-}
-
-type pipelineRunFunc func(context.Context, RunPipelineOptions) (RunReport, error)
-
-func NewPipelineRunCoordinator(ctx context.Context) *PipelineRunCoordinator {
-	return newPipelineRunCoordinator(ctx, RunPipeline)
-}
-
-func newPipelineRunCoordinator(ctx context.Context, run pipelineRunFunc) *PipelineRunCoordinator {
-	if ctx == nil {
-		ctx = context.Background()
-	}
-	return &PipelineRunCoordinator{
-		ctx:    ctx,
-		run:    run,
-		now:    time.Now,
-		active: map[string]PipelineRunRecord{},
-	}
-}
-
-func (coordinator *PipelineRunCoordinator) RunPipeline(ctx context.Context, options RunPipelineOptions) (PipelineRunRecord, error) {
-	if ctx == nil {
-		ctx = context.Background()
-	}
-	if err := ctx.Err(); err != nil {
-		return PipelineRunRecord{}, err
-	}
-	record, err := coordinator.admit(options.PipelineID)
-	if err != nil {
-		return PipelineRunRecord{}, err
-	}
-	defer coordinator.clear(options.PipelineID)
-
-	report, runErr := coordinator.run(coordinator.ctx, options)
-	record.Report = report
-	finishedAt := coordinator.now().UTC()
-	record.FinishedAt = &finishedAt
-	if runErr != nil {
-		record.Status = PipelineRunFailed
-		record.Error = runErr.Error()
-		return record, runErr
-	}
-	record.Status = PipelineRunSucceeded
-	return record, nil
-}
-
-func (coordinator *PipelineRunCoordinator) admit(pipelineID string) (PipelineRunRecord, error) {
-	coordinator.mu.Lock()
-	defer coordinator.mu.Unlock()
-	if active, ok := coordinator.active[pipelineID]; ok {
-		return PipelineRunRecord{}, DuplicatePipelineRunError{
-			PipelineID: pipelineID,
-			RunID:      active.ID,
-		}
-	}
-	coordinator.nextID++
-	record := PipelineRunRecord{
-		ID:         PipelineRunID(fmt.Sprintf("run-%016d", coordinator.nextID)),
-		PipelineID: pipelineID,
-		Status:     PipelineRunRunning,
-		StartedAt:  coordinator.now().UTC(),
-	}
-	coordinator.active[pipelineID] = record
-	return record, nil
-}
-
-func (coordinator *PipelineRunCoordinator) clear(pipelineID string) {
-	coordinator.mu.Lock()
-	defer coordinator.mu.Unlock()
-	delete(coordinator.active, pipelineID)
-}
-
-func (coordinator *PipelineRunCoordinator) activeCount() int {
-	coordinator.mu.Lock()
-	defer coordinator.mu.Unlock()
-	return len(coordinator.active)
-}

internal/app/run_coordinator_test.go

@@ -1,223 +0,0 @@
-package app
-
-import (
-	"context"
-	"errors"
-	"sync"
-	"testing"
-	"time"
-)
-
-func TestPipelineRunCoordinatorRejectsDuplicateActiveRun(t *testing.T) {
-	started := make(chan struct{})
-	release := make(chan struct{})
-	var startedOnce sync.Once
-	coordinator := newPipelineRunCoordinator(context.Background(), func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
-		startedOnce.Do(func() {
-			close(started)
-		})
-		<-release
-		return RunReport{}, nil
-	})
-	firstResult := make(chan runCoordinatorTestResult, 1)
-
-	go func() {
-		record, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
-		firstResult <- runCoordinatorTestResult{record: record, err: err}
-	}()
-	waitForSignal(t, started, "first run to start")
-
-	_, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
-	if err == nil || !IsDuplicatePipelineRun(err) {
-		t.Fatalf("RunPipeline() error = %v, want duplicate active run", err)
-	}
-	close(release)
-	result := waitForRunResult(t, firstResult)
-	if result.err != nil {
-		t.Fatalf("first RunPipeline() error = %v", result.err)
-	}
-	if result.record.Status != PipelineRunSucceeded || result.record.ID == "" || result.record.FinishedAt == nil {
-		t.Fatalf("first record = %#v, want succeeded completed record", result.record)
-	}
-	if got := coordinator.activeCount(); got != 0 {
-		t.Fatalf("active count = %d, want 0", got)
-	}
-}
-
-func TestPipelineRunCoordinatorAllowsDifferentActivePipelines(t *testing.T) {
-	started := make(chan string, 2)
-	release := make(chan struct{})
-	coordinator := newPipelineRunCoordinator(context.Background(), func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
-		started <- options.PipelineID
-		<-release
-		return RunReport{}, nil
-	})
-	firstResult := make(chan runCoordinatorTestResult, 1)
-	secondResult := make(chan runCoordinatorTestResult, 1)
-
-	go func() {
-		record, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports-one"})
-		firstResult <- runCoordinatorTestResult{record: record, err: err}
-	}()
-	go func() {
-		record, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports-two"})
-		secondResult <- runCoordinatorTestResult{record: record, err: err}
-	}()
-	startedPipelines := map[string]bool{
-		waitForPipelineID(t, started): true,
-		waitForPipelineID(t, started): true,
-	}
-	if !startedPipelines["reports-one"] || !startedPipelines["reports-two"] {
-		t.Fatalf("started pipelines = %#v, want both requested pipelines", startedPipelines)
-	}
-	if got := coordinator.activeCount(); got != 2 {
-		t.Fatalf("active count = %d, want 2", got)
-	}
-
-	close(release)
-	first := waitForRunResult(t, firstResult)
-	second := waitForRunResult(t, secondResult)
-	if first.err != nil || second.err != nil {
-		t.Fatalf("RunPipeline() errors = %v, %v; want nil", first.err, second.err)
-	}
-	if first.record.ID == second.record.ID {
-		t.Fatalf("run IDs matched: %q", first.record.ID)
-	}
-	if got := coordinator.activeCount(); got != 0 {
-		t.Fatalf("active count = %d, want 0", got)
-	}
-}
-
-func TestPipelineRunCoordinatorClearsActiveRunAfterSuccess(t *testing.T) {
-	coordinator := newPipelineRunCoordinator(context.Background(), func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
-		return RunReport{DryRun: options.DryRun}, nil
-	})
-
-	first, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports", DryRun: true})
-	if err != nil {
-		t.Fatalf("first RunPipeline() error = %v", err)
-	}
-	second, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
-	if err != nil {
-		t.Fatalf("second RunPipeline() error = %v", err)
-	}
-	if first.Status != PipelineRunSucceeded || second.Status != PipelineRunSucceeded {
-		t.Fatalf("statuses = %s, %s; want succeeded", first.Status, second.Status)
-	}
-	if !first.Report.DryRun {
-		t.Fatalf("first report dry_run = false, want true")
-	}
-	if got := coordinator.activeCount(); got != 0 {
-		t.Fatalf("active count = %d, want 0", got)
-	}
-}
-
-func TestPipelineRunCoordinatorClearsActiveRunAfterFailure(t *testing.T) {
-	runError := errors.New("run failed")
-	attempt := 0
-	coordinator := newPipelineRunCoordinator(context.Background(), func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
-		attempt++
-		if attempt == 1 {
-			return RunReport{}, runError
-		}
-		return RunReport{}, nil
-	})
-
-	first, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
-	if !errors.Is(err, runError) {
-		t.Fatalf("first RunPipeline() error = %v, want run failure", err)
-	}
-	if first.Status != PipelineRunFailed || first.Error != runError.Error() || first.FinishedAt == nil {
-		t.Fatalf("first record = %#v, want failed completed record", first)
-	}
-	second, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
-	if err != nil {
-		t.Fatalf("second RunPipeline() error = %v", err)
-	}
-	if second.Status != PipelineRunSucceeded {
-		t.Fatalf("second status = %s, want succeeded", second.Status)
-	}
-	if got := coordinator.activeCount(); got != 0 {
-		t.Fatalf("active count = %d, want 0", got)
-	}
-}
-
-func TestPipelineRunCoordinatorClearsActiveRunAfterCancellation(t *testing.T) {
-	runContext, cancel := context.WithCancel(context.Background())
-	coordinator := newPipelineRunCoordinator(runContext, func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
-		return RunReport{}, ctx.Err()
-	})
-	cancel()
-
-	record, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
-	if !errors.Is(err, context.Canceled) {
-		t.Fatalf("RunPipeline() error = %v, want context canceled", err)
-	}
-	if record.Status != PipelineRunFailed || record.Error != context.Canceled.Error() {
-		t.Fatalf("record = %#v, want failed cancellation record", record)
-	}
-	_, err = coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
-	if !errors.Is(err, context.Canceled) {
-		t.Fatalf("second RunPipeline() error = %v, want context canceled", err)
-	}
-	if IsDuplicatePipelineRun(err) {
-		t.Fatalf("second RunPipeline() error = %v, want cancellation instead of duplicate", err)
-	}
-	if got := coordinator.activeCount(); got != 0 {
-		t.Fatalf("active count = %d, want 0", got)
-	}
-}
-
-func TestPipelineRunCoordinatorUnknownPipelineDoesNotRemainActive(t *testing.T) {
-	coordinator := newPipelineRunCoordinator(context.Background(), func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
-		return RunReport{}, PipelineNotFoundError{ID: options.PipelineID}
-	})
-
-	_, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "missing"})
-	if err == nil || !IsPipelineNotFound(err) {
-		t.Fatalf("RunPipeline() error = %v, want pipeline not found", err)
-	}
-	_, err = coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "missing"})
-	if err == nil || !IsPipelineNotFound(err) || IsDuplicatePipelineRun(err) {
-		t.Fatalf("second RunPipeline() error = %v, want pipeline not found without duplicate", err)
-	}
-	if got := coordinator.activeCount(); got != 0 {
-		t.Fatalf("active count = %d, want 0", got)
-	}
-}
-
-type runCoordinatorTestResult struct {
-	record PipelineRunRecord
-	err    error
-}
-
-func waitForSignal(t *testing.T, signal <-chan struct{}, name string) {
-	t.Helper()
-	select {
-	case <-signal:
-	case <-time.After(time.Second):
-		t.Fatalf("timed out waiting for %s", name)
-	}
-}
-
-func waitForPipelineID(t *testing.T, pipelineIDs <-chan string) string {
-	t.Helper()
-	select {
-	case pipelineID := <-pipelineIDs:
-		return pipelineID
-	case <-time.After(time.Second):
-		t.Fatalf("timed out waiting for pipeline start")
-		return ""
-	}
-}
-
-func waitForRunResult(t *testing.T, results <-chan runCoordinatorTestResult) runCoordinatorTestResult {
-	t.Helper()
-	select {
-	case result := <-results:
-		return result
-	case <-time.After(time.Second):
-		t.Fatalf("timed out waiting for run result")
-		return runCoordinatorTestResult{}
-	}
-}

internal/app/run_destination.go

@@ -0,0 +1,163 @@
+package app
+
+import (
+	"context"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/notify"
+	"gitea.maximumdirect.net/eric/distributor/internal/publish"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+)
+
+type runDestinationRequest struct {
+	options       RunOptions
+	notifier      notify.Notifier
+	backends      *backendFactory
+	transforms    publish.TransformerResolver
+	pipeline      config.Pipeline
+	pipelineIndex int
+	sourceBackend storage.Backend
+	bundles       []bundle.Bundle
+	destination   config.Destination
+	recorder      *runReportRecorder
+}
+
+func processDestination(ctx context.Context, request runDestinationRequest) {
+	selections := selectDestinationBundles(request.destination, request.bundles)
+	if isFixedPathDestination(request.destination) {
+		request.recorder.summary.recordFixedPath()
+		if request.options.DryRun {
+			warning := fixedPathSelectionWarning(request.pipeline.ID, request.destination.ID, selections, len(request.bundles))
+			request.recorder.addPipelineWarning(request.pipelineIndex, warning)
+		}
+	}
+	if len(selections) == 0 {
+		return
+	}
+	destinationBackend, err := request.backends.openDestination(ctx, request.destination)
+	if err != nil {
+		for _, selection := range selections {
+			sourceBundle := selection.SourceBundle
+			request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{
+				pipelineID:    request.pipeline.ID,
+				destinationID: request.destination.ID,
+				backend:       request.destination.Backend,
+				bundlePath:    sourceBundle.RootRelativePath,
+				err:           err,
+			}, errorAction(request.pipeline.ID, request.destination.ID, request.destination.Backend, sourceBundle.RootRelativePath, err), true)
+		}
+		return
+	}
+	defer closeBackend(destinationBackend)
+
+	for _, selection := range selections {
+		processDestinationSelection(ctx, request, destinationBackend, selection)
+	}
+}
+
+func processDestinationSelection(ctx context.Context, request runDestinationRequest, destinationBackend storage.Backend, selection destinationBundleSelection) {
+	sourceBundle := selection.SourceBundle
+	publishRequest := publish.Request{
+		PipelineID:            request.pipeline.ID,
+		DestinationID:         request.destination.ID,
+		SourceBundle:          sourceBundle,
+		SourceBackend:         request.sourceBackend,
+		DestinationBackend:    destinationBackend,
+		DestinationBundlePath: selection.DestinationBundlePath,
+		PathMapping:           request.destination.PathMap.Mode,
+		Publish:               *request.destination.Publish,
+		Transform:             request.destination.Transform,
+		Links:                 request.destination.Links,
+		Transformers:          request.transforms,
+		Transfer:              request.destination.Transfer,
+		DistributorVersion:    Version,
+		Force:                 request.options.Force,
+	}
+	plan, err := publish.Build(ctx, publishRequest)
+	if err != nil {
+		plan = completePlanIdentity(plan, request.pipeline, request.destination, selection)
+	}
+	if isFixedPathDestination(request.destination) {
+		plan.PathMapping = config.PathMappingFixed
+		if request.options.DryRun && isDestructiveFixedPathAction(plan.Action) {
+			warning := fixedPathReplacementWarning(plan)
+			request.recorder.addPipelineWarning(request.pipelineIndex, warning)
+		}
+	}
+	action := runActionFromPlan(request.destination.Backend, plan, err)
+	if err != nil {
+		request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{
+			pipelineID:    request.pipeline.ID,
+			destinationID: request.destination.ID,
+			backend:       request.destination.Backend,
+			bundlePath:    sourceBundle.RootRelativePath,
+			err:           err,
+		}, action, true)
+		return
+	}
+	request.recorder.addPipelineAction(request.pipelineIndex, action)
+	request.recorder.summary.recordPlan(plan.Action)
+	if request.options.DryRun {
+		return
+	}
+	if err := publish.Execute(ctx, publishRequest, plan); err != nil {
+		request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{
+			pipelineID:    request.pipeline.ID,
+			destinationID: request.destination.ID,
+			backend:       request.destination.Backend,
+			bundlePath:    sourceBundle.RootRelativePath,
+			err:           err,
+		}, RunActionRecord{}, false)
+		return
+	}
+	if shouldNotify(plan.Action) {
+		if err := request.notifier.Notify(ctx, notifyEvent(plan)); err != nil {
+			request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{
+				pipelineID:    request.pipeline.ID,
+				destinationID: request.destination.ID,
+				backend:       request.destination.Backend,
+				bundlePath:    sourceBundle.RootRelativePath,
+				err:           err,
+			}, RunActionRecord{}, false)
+			return
+		}
+	}
+}
+
+func (recorder *runReportRecorder) addPipelineWarning(pipelineIndex int, warning OutputWarning) {
+	recorder.report.addWarning(warning)
+	recorder.report.Pipelines[pipelineIndex].events = append(recorder.report.Pipelines[pipelineIndex].events, warningEvent(warning))
+}
+
+func (recorder *runReportRecorder) addPipelineAction(pipelineIndex int, action RunActionRecord) {
+	recorder.report.Actions = append(recorder.report.Actions, action)
+	recorder.report.Pipelines[pipelineIndex].events = append(recorder.report.Pipelines[pipelineIndex].events, actionEvent(len(recorder.report.Actions)-1))
+}
+
+func (recorder *runReportRecorder) recordDestinationFailure(pipelineIndex int, failure runFailure, action RunActionRecord, includeAction bool) {
+	recorder.failures.add(failure.pipelineID, failure.destinationID, failure.backend, storage.DisplayPath(failure.bundlePath), failure.err)
+	recorder.summary.recordFailure()
+	if includeAction {
+		recorder.addPipelineAction(pipelineIndex, action)
+	}
+}
+
+func completePlanIdentity(plan publish.Plan, pipeline config.Pipeline, destination config.Destination, selection destinationBundleSelection) publish.Plan {
+	if plan.PipelineID == "" {
+		plan.PipelineID = pipeline.ID
+	}
+	if plan.DestinationID == "" {
+		plan.DestinationID = destination.ID
+	}
+	if plan.BundleID == "" {
+		plan.BundleID = selection.SourceBundle.Manifest.ID
+	}
+	if plan.BundlePath == "" {
+		plan.BundlePath = selection.SourceBundle.RootRelativePath
+	}
+	if plan.DestinationBundlePath == "" {
+		plan.DestinationBundlePath = selection.DestinationBundlePath
+	}
+	return plan
+}

internal/app/run_test.go

@@ -922,6 +922,50 @@ func TestBuildRunReportIncludesPartialFailures(t *testing.T) {
 	}
 }
 
+func TestBuildRunReportAlignsDestinationOpenFailuresForSelectedBundles(t *testing.T) {
+	sourceRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "daily/one", testBundleOptions{ID: "reports.one"})
+	writeSourceBundle(t, sourceRoot, "daily/two", testBundleOptions{ID: "reports.two", Created: testutil.DefaultCreated.Add(time.Hour)})
+	cfg := config.Config{Pipelines: []config.Pipeline{{
+		ID:     "reports",
+		Source: config.Backend{Backend: config.BackendLocal, Path: sourceRoot},
+		Destinations: []config.Destination{{
+			ID:       "object-archive",
+			Backend:  config.BackendS3,
+			Endpoint: "http://s3.test",
+			Bucket:   "missing-destination",
+		}},
+	}}}
+	config.ApplyDefaults(&cfg)
+
+	report, err := buildRunReportWithBackendFactory(context.Background(), cfg, RunOptions{}, fakeBackendFactoryProvider(t, nil))
+	if err == nil || !IsPartialResultError(err) {
+		t.Fatalf("buildRunReportWithBackendFactory() error = %v, want partial result error", err)
+	}
+	if report.Summary.Status != "failed" || report.Summary.Planned != 0 || report.Summary.Failed != 2 {
+		t.Fatalf("summary = %#v, want two destination open failures", report.Summary)
+	}
+	if got, want := len(report.Actions), 2; got != want {
+		t.Fatalf("action count = %d, want %d", got, want)
+	}
+	if got, want := len(report.OutputErrors), 2; got != want {
+		t.Fatalf("output error count = %d, want %d", got, want)
+	}
+	if got, want := len(report.Pipelines[0].events), 2; got != want {
+		t.Fatalf("pipeline event count = %d, want %d", got, want)
+	}
+	for index, bundlePath := range []string{"daily/one", "daily/two"} {
+		action := report.Actions[index]
+		if action.PipelineID != "reports" || action.DestinationID != "object-archive" || action.Backend != config.BackendS3 || action.BundlePath != bundlePath || action.Action != "error" {
+			t.Fatalf("action[%d] = %#v, want %s destination open error", index, action, bundlePath)
+		}
+		outputError := report.OutputErrors[index]
+		if outputError.PipelineID != action.PipelineID || outputError.DestinationID != action.DestinationID || outputError.Backend != action.Backend || outputError.BundlePath != action.BundlePath {
+			t.Fatalf("output error[%d] = %#v, action = %#v, want aligned identity", index, outputError, action)
+		}
+	}
+}
+
 func TestRunPipelineRunsOnlyRequestedPipeline(t *testing.T) {
 	firstSource := t.TempDir()
 	secondSource := t.TempDir()

internal/app/runtime.go

@@ -0,0 +1,44 @@
+package app
+
+import "gitea.maximumdirect.net/eric/distributor/internal/config"
+
+type runtimeSetup struct {
+	ConfigPath  string
+	Config      config.Config
+	Environment config.Environment
+	Warnings    []OutputWarning
+}
+
+func loadRuntimeSetup(configPath string) (runtimeSetup, error) {
+	resolvedPath := runtimeConfigPath(configPath)
+	cfg, err := config.LoadFile(resolvedPath)
+	if err != nil {
+		return runtimeSetup{}, err
+	}
+	return runtimeSetupFromConfig(resolvedPath, cfg)
+}
+
+func runtimeSetupFromConfig(configPath string, cfg config.Config) (runtimeSetup, error) {
+	secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil)
+	if err != nil {
+		return runtimeSetup{}, err
+	}
+	return runtimeSetup{
+		ConfigPath:  configPath,
+		Config:      cfg,
+		Environment: secretLoad.Environment,
+		Warnings:    secretConflictWarnings(secretLoad.Conflicts),
+	}, nil
+}
+
+func runtimeConfigPath(configPath string) string {
+	if configPath == "" {
+		return config.DefaultConfigPath
+	}
+	return configPath
+}
+
+func (setup runtimeSetup) withPipelines(pipelines []config.Pipeline) runtimeSetup {
+	setup.Config.Pipelines = pipelines
+	return setup
+}

internal/app/runtime_test.go

@@ -0,0 +1,32 @@
+package app
+
+import (
+	"testing"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
+)
+
+func TestRuntimeConfigPathDefaultsEmptyPath(t *testing.T) {
+	if got, want := runtimeConfigPath(""), config.DefaultConfigPath; got != want {
+		t.Fatalf("runtimeConfigPath(\"\") = %q, want %q", got, want)
+	}
+	if got, want := runtimeConfigPath("/tmp/distributor.yml"), "/tmp/distributor.yml"; got != want {
+		t.Fatalf("runtimeConfigPath(explicit) = %q, want %q", got, want)
+	}
+}
+
+func TestLoadRuntimeSetupReturnsLoadedConfigPath(t *testing.T) {
+	configPath := testutil.WriteMinimalLocalConfig(t, t.TempDir(), t.TempDir())
+
+	setup, err := loadRuntimeSetup(configPath)
+	if err != nil {
+		t.Fatalf("loadRuntimeSetup() error = %v", err)
+	}
+	if setup.ConfigPath != configPath {
+		t.Fatalf("ConfigPath = %q, want %q", setup.ConfigPath, configPath)
+	}
+	if len(setup.Config.Pipelines) != 1 {
+		t.Fatalf("pipeline count = %d, want 1", len(setup.Config.Pipelines))
+	}
+}

internal/app/serve.go

@@ -6,8 +6,6 @@ import (
 	"fmt"
 	"net"
 	"net/http"
-
-	"gitea.maximumdirect.net/eric/distributor/internal/config"
 )
 
 type ServeOptions struct {
@@ -22,26 +20,18 @@ func Serve(ctx context.Context, options ServeOptions) error {
 		return err
 	}
 
-	configPath := options.ConfigPath
-	if configPath == "" {
-		configPath = config.DefaultConfigPath
-	}
-	cfg, err := config.LoadFile(configPath)
-	if err != nil {
-		return err
-	}
-	secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil)
+	setup, err := loadRuntimeSetup(options.ConfigPath)
 	if err != nil {
 		return err
 	}
 
-	handler, err := newUploadHTTPHandler(ctx, cfg, secretLoad.Environment)
+	handler, err := newUploadHTTPHandler(ctx, setup.Config, setup.Environment)
 	if err != nil {
 		return err
 	}
-	listener, err := net.Listen("tcp", cfg.Server.HTTP.Bind)
+	listener, err := net.Listen("tcp", setup.Config.Server.HTTP.Bind)
 	if err != nil {
-		return fmt.Errorf("bind HTTP server %q: %w", cfg.Server.HTTP.Bind, err)
+		return fmt.Errorf("bind HTTP server %q: %w", setup.Config.Server.HTTP.Bind, err)
 	}
 	defer listener.Close()
 

internal/app/serve_test.go

@@ -0,0 +1,88 @@
+package app
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+func TestServeFailsForUnsafeUploadTokensWithoutLeakingValues(t *testing.T) {
+	duplicateSecret := "duplicate-secret"
+	tests := []struct {
+		name       string
+		configPath func(*testing.T) string
+		env        map[string]string
+		want       string
+	}{
+		{
+			name: "missing token",
+			configPath: func(t *testing.T) string {
+				return writeServeUploadConfig(t, []string{"DISTRIBUTOR_TEST_MISSING_UPLOAD_TOKEN"})
+			},
+			want: "DISTRIBUTOR_TEST_MISSING_UPLOAD_TOKEN",
+		},
+		{
+			name: "empty token",
+			configPath: func(t *testing.T) string {
+				return writeServeUploadConfig(t, []string{"DISTRIBUTOR_TEST_EMPTY_UPLOAD_TOKEN"})
+			},
+			env:  map[string]string{"DISTRIBUTOR_TEST_EMPTY_UPLOAD_TOKEN": ""},
+			want: "DISTRIBUTOR_TEST_EMPTY_UPLOAD_TOKEN",
+		},
+		{
+			name: "duplicate token",
+			configPath: func(t *testing.T) string {
+				return writeServeUploadConfig(t, []string{
+					"DISTRIBUTOR_TEST_FIRST_UPLOAD_TOKEN",
+					"DISTRIBUTOR_TEST_SECOND_UPLOAD_TOKEN",
+				})
+			},
+			env: map[string]string{
+				"DISTRIBUTOR_TEST_FIRST_UPLOAD_TOKEN":  duplicateSecret,
+				"DISTRIBUTOR_TEST_SECOND_UPLOAD_TOKEN": duplicateSecret,
+			},
+			want: "same value",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			for name, value := range tt.env {
+				t.Setenv(name, value)
+			}
+
+			err := Serve(context.Background(), ServeOptions{ConfigPath: tt.configPath(t)})
+			if err == nil {
+				t.Fatal("Serve() error = nil, want token startup error")
+			}
+			if !strings.Contains(err.Error(), tt.want) {
+				t.Fatalf("Serve() error = %v, want %q", err, tt.want)
+			}
+			if strings.Contains(err.Error(), duplicateSecret) {
+				t.Fatalf("Serve() error exposed token value: %v", err)
+			}
+		})
+	}
+}
+
+func writeServeUploadConfig(t *testing.T, tokenEnvs []string) string {
+	t.Helper()
+	body := `
+server:
+  http:
+    bind: 127.0.0.1:0
+pipelines:
+`
+	for index, tokenEnv := range tokenEnvs {
+		body += `
+  - id: reports-` + string(rune('a'+index)) + `
+    source:
+      backend: http_upload
+      token_env: ` + tokenEnv + `
+    destinations:
+      - id: archive
+        backend: local
+        path: ` + t.TempDir() + `
+`
+	}
+	return writeConfigFile(t, body)
+}

internal/app/source_select.go

@@ -44,11 +44,11 @@ func selectSourceBundles(ctx context.Context, options sourceCommandOptions, prov
 		return sourceSelection{}, err
 	}
 	if options.ConfigPath != "" {
-		cfg, err := config.LoadFile(options.ConfigPath)
+		setup, err := loadRuntimeSetup(options.ConfigPath)
 		if err != nil {
 			return sourceSelection{}, err
 		}
-		return selectSourceBundlesFromConfig(ctx, cfg, options, provider)
+		return selectSourceBundlesFromSetup(ctx, setup, options, provider)
 	}
 	if options.PipelineID != "" {
 		return sourceSelection{}, fmt.Errorf("configured source mode requires --config")
@@ -72,21 +72,25 @@ func selectSourceBundles(ctx context.Context, options sourceCommandOptions, prov
 }
 
 func selectSourceBundlesFromConfig(ctx context.Context, cfg config.Config, options sourceCommandOptions, provider backendFactoryProvider) (sourceSelection, error) {
+	setup, err := runtimeSetupFromConfig("", cfg)
+	if err != nil {
+		return sourceSelection{}, err
+	}
+	return selectSourceBundlesFromSetup(ctx, setup, options, provider)
+}
+
+func selectSourceBundlesFromSetup(ctx context.Context, setup runtimeSetup, options sourceCommandOptions, provider backendFactoryProvider) (sourceSelection, error) {
 	if options.Path != "" {
 		return sourceSelection{}, fmt.Errorf("configured source mode does not accept a local path")
 	}
 	if options.PipelineID == "" {
 		return sourceSelection{}, fmt.Errorf("configured source mode requires --pipeline")
 	}
-	secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil)
-	if err != nil {
-		return sourceSelection{}, err
-	}
-	pipeline, ok := findPipeline(cfg, options.PipelineID)
+	pipeline, ok := findPipeline(setup.Config, options.PipelineID)
 	if !ok {
 		return sourceSelection{}, PipelineNotFoundError{ID: options.PipelineID}
 	}
-	backends := provider(secretLoad.Environment)
+	backends := provider(setup.Environment)
 	sourceBackend, err := backends.openSource(ctx, pipeline.Source)
 	if err != nil {
 		return sourceSelection{}, fmt.Errorf("pipeline %s source backend %s: %w", pipeline.ID, pipeline.Source.Backend, err)
@@ -111,7 +115,7 @@ func selectSourceBundlesFromConfig(ctx context.Context, cfg config.Config, optio
 		PipelineID:    pipeline.ID,
 		SourceBackend: pipeline.Source.Backend,
 		ConfigMode:    true,
-		Warnings:      append(secretConflictWarnings(secretLoad.Conflicts), sourceSSHWarnings(pipeline)...),
+		Warnings:      append(setup.Warnings, sourceSSHWarnings(pipeline)...),
 	}, nil
 }
 

internal/app/upload_coordinator.go

@@ -78,6 +78,7 @@ type UploadCoordinator struct {
 	queueSize      int
 	maxConcurrency int
 	runningCount   int
+	reservedCount  int
 	activePipeline map[string]bool
 	pending        []*uploadJob
 	records        map[UploadRunID]UploadRunRecord
@@ -88,9 +89,10 @@ type uploadStageFunc func(context.Context, ingest.StageOptions) (ingest.StagedBu
 type uploadRunFunc func(context.Context, config.Config, RunPipelineWithLocalSourceOptions) (RunReport, error)
 
 type uploadJob struct {
-	recordID UploadRunID
-	request  UploadRequest
-	pipeline config.Pipeline
+	recordID   UploadRunID
+	request    UploadRequest
+	pipeline   config.Pipeline
+	stagedRoot string
 }
 
 type uploadCoordinatorHooks struct {
@@ -164,24 +166,49 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
 	if err != nil {
 		return UploadRunRecord{}, err
 	}
+	if err := ingest.ValidateContentType(request.ContentType); err != nil {
+		return UploadRunRecord{}, err
+	}
+
+	coordinator.mu.Lock()
+	coordinator.expireLocked(coordinator.now().UTC())
+	if coordinator.queueFullLocked() {
+		coordinator.mu.Unlock()
+		return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize}
+	}
+	coordinator.reservedCount++
+	coordinator.mu.Unlock()
+
+	staged, err := coordinator.stage(ctx, ingest.StageOptions{
+		Body:                request.Body,
+		ContentType:         request.ContentType,
+		PipelineStagingPath: pipeline.Source.Upload.StagingPath,
+		RunID:               string(runID),
+		MaxUploadSize:       int64(*pipeline.Source.Upload.MaxUploadSize),
+		MaxExtractedSize:    int64(*pipeline.Source.Upload.MaxUploadSize),
+		MaxFileCount:        uploadMaxFileCount(request.MaxFileCount),
+	})
+	if err != nil {
+		coordinator.releaseReservation()
+		return UploadRunRecord{}, err
+	}
 
 	coordinator.mu.Lock()
 	defer coordinator.mu.Unlock()
-	coordinator.expireLocked(coordinator.now().UTC())
-	if len(coordinator.pending) >= coordinator.queueSize {
-		return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize}
-	}
+	coordinator.reservedCount--
 	record := UploadRunRecord{
 		ID:         runID,
 		PipelineID: pipeline.ID,
 		Status:     UploadStatusAccepted,
 		AcceptedAt: coordinator.now().UTC(),
+		StagedRoot: staged.Root,
 	}
 	coordinator.records[runID] = record
 	coordinator.pending = append(coordinator.pending, &uploadJob{
-		recordID: runID,
-		request:  request,
-		pipeline: pipeline,
+		recordID:   runID,
+		request:    request,
+		pipeline:   pipeline,
+		stagedRoot: staged.Root,
 	})
 	coordinator.notify()
 	return record, nil
@@ -205,13 +232,13 @@ func (coordinator *UploadCoordinator) CanAccept() bool {
 	coordinator.mu.Lock()
 	defer coordinator.mu.Unlock()
 	coordinator.expireLocked(coordinator.now().UTC())
-	return len(coordinator.pending) < coordinator.queueSize
+	return !coordinator.queueFullLocked()
 }
 
 func (coordinator *UploadCoordinator) QueueDepth() int {
 	coordinator.mu.Lock()
 	defer coordinator.mu.Unlock()
-	return len(coordinator.pending)
+	return len(coordinator.pending) + coordinator.reservedCount
 }
 
 func (coordinator *UploadCoordinator) RunningCount() int {
@@ -290,47 +317,30 @@ func (coordinator *UploadCoordinator) markPendingQueuedLocked() {
 }
 
 func (coordinator *UploadCoordinator) runJob(job *uploadJob) {
-	record := coordinator.currentRecord(job.recordID)
-	maxFileCount := job.request.MaxFileCount
-	if maxFileCount <= 0 {
-		maxFileCount = DefaultUploadMaxFileCount
-	}
-	staged, err := coordinator.stage(coordinator.ctx, ingest.StageOptions{
-		Body:                job.request.Body,
-		ContentType:         job.request.ContentType,
-		PipelineStagingPath: job.pipeline.Source.Upload.StagingPath,
-		RunID:               string(record.ID),
-		MaxUploadSize:       int64(*job.pipeline.Source.Upload.MaxUploadSize),
-		MaxExtractedSize:    int64(*job.pipeline.Source.Upload.MaxUploadSize),
-		MaxFileCount:        maxFileCount,
+	report, err := coordinator.run(coordinator.ctx, coordinator.cfg, RunPipelineWithLocalSourceOptions{
+		PipelineID: job.pipeline.ID,
+		SourceRoot: job.stagedRoot,
+		DryRun:     job.request.DryRun,
+		Force:      job.request.Force,
 	})
-	if err == nil {
-		coordinator.setStagedRoot(job.recordID, staged.Root)
-		var report RunReport
-		report, err = coordinator.run(coordinator.ctx, coordinator.cfg, RunPipelineWithLocalSourceOptions{
-			PipelineID: job.pipeline.ID,
-			SourceRoot: staged.Root,
-			DryRun:     job.request.DryRun,
-			Force:      job.request.Force,
-		})
-		coordinator.complete(job, &report, err)
-		return
+	coordinator.complete(job, &report, err)
+}
+
+func (coordinator *UploadCoordinator) releaseReservation() {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	coordinator.reservedCount--
+}
+
+func (coordinator *UploadCoordinator) queueFullLocked() bool {
+	return len(coordinator.pending)+coordinator.reservedCount >= coordinator.queueSize
+}
+
+func uploadMaxFileCount(value int) int {
+	if value > 0 {
+		return value
 	}
-	coordinator.complete(job, nil, err)
-}
-
-func (coordinator *UploadCoordinator) currentRecord(runID UploadRunID) UploadRunRecord {
-	coordinator.mu.Lock()
-	defer coordinator.mu.Unlock()
-	return coordinator.records[runID]
-}
-
-func (coordinator *UploadCoordinator) setStagedRoot(runID UploadRunID, root string) {
-	coordinator.mu.Lock()
-	defer coordinator.mu.Unlock()
-	record := coordinator.records[runID]
-	record.StagedRoot = root
-	coordinator.records[runID] = record
+	return DefaultUploadMaxFileCount
 }
 
 func (coordinator *UploadCoordinator) complete(job *uploadJob, report *RunReport, runErr error) {

internal/app/upload_http.go

@@ -1,13 +1,10 @@
 package app
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
-	"mime"
 	"net/http"
 	"strings"
 
@@ -24,7 +21,6 @@ type uploadCoordinator interface {
 type uploadHTTPHandler struct {
 	coordinator uploadCoordinator
 	tokens      map[string]string
-	limits      map[string]int64
 }
 
 type uploadAcceptedResponse struct {
@@ -38,20 +34,18 @@ type httpErrorResponse struct {
 
 func newUploadHTTPHandler(ctx context.Context, cfg config.Config, environment config.Environment) (http.Handler, error) {
 	config.ApplyDefaults(&cfg)
-	tokens, limits, err := resolveUploadTokens(cfg, environment)
+	tokens, err := resolveUploadTokens(cfg, environment)
 	if err != nil {
 		return nil, err
 	}
 	return uploadHTTPHandler{
 		coordinator: NewUploadCoordinator(ctx, cfg),
 		tokens:      tokens,
-		limits:      limits,
 	}, nil
 }
 
-func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]string, map[string]int64, error) {
+func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]string, error) {
 	tokens := make(map[string]string)
-	limits := make(map[string]int64)
 	for _, pipeline := range cfg.Pipelines {
 		if pipeline.Source.Backend != config.BackendHTTPUpload {
 			continue
@@ -59,18 +53,17 @@ func resolveUploadTokens(cfg config.Config, environment config.Environment) (map
 		tokenName := pipeline.Source.Upload.TokenEnv
 		token, ok := environment.Lookup(tokenName)
 		if !ok {
-			return nil, nil, fmt.Errorf("upload token environment variable %s is not set", tokenName)
+			return nil, fmt.Errorf("upload token environment variable %s is not set", tokenName)
 		}
 		if token == "" {
-			return nil, nil, fmt.Errorf("upload token environment variable %s is empty", tokenName)
+			return nil, fmt.Errorf("upload token environment variable %s is empty", tokenName)
 		}
 		if existing, exists := tokens[token]; exists {
-			return nil, nil, fmt.Errorf("upload token environment variables for pipelines %s and %s resolve to the same value", existing, pipeline.ID)
+			return nil, fmt.Errorf("upload token environment variables for pipelines %s and %s resolve to the same value", existing, pipeline.ID)
 		}
 		tokens[token] = pipeline.ID
-		limits[pipeline.ID] = int64(*pipeline.Source.Upload.MaxUploadSize)
 	}
-	return tokens, limits, nil
+	return tokens, nil
 }
 
 func (handler uploadHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -101,7 +94,7 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req
 		return
 	}
 	contentType := r.Header.Get("Content-Type")
-	if !supportedUploadContentType(contentType) {
+	if err := ingest.ValidateContentType(contentType); err != nil {
 		writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type")
 		return
 	}
@@ -109,19 +102,10 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req
 		writeHTTPError(w, http.StatusServiceUnavailable, "upload queue is full")
 		return
 	}
-	body, err := readUploadBody(r.Body, handler.limits[pipelineID])
-	if err != nil {
-		if errors.Is(err, ingest.ErrUploadTooLarge) {
-			writeHTTPError(w, http.StatusRequestEntityTooLarge, "upload exceeds maximum size")
-			return
-		}
-		writeHTTPError(w, http.StatusBadRequest, "read upload body failed")
-		return
-	}
 	record, err := handler.coordinator.Submit(r.Context(), UploadRequest{
 		PipelineID:  pipelineID,
 		ContentType: contentType,
-		Body:        bytes.NewReader(body),
+		Body:        r.Body,
 	})
 	if err != nil {
 		writeUploadSubmitError(w, err)
@@ -160,31 +144,6 @@ func (handler uploadHTTPHandler) authenticate(header string) (string, bool) {
 	return pipelineID, ok
 }
 
-func supportedUploadContentType(contentType string) bool {
-	mediaType, _, err := mime.ParseMediaType(contentType)
-	if err != nil {
-		mediaType = contentType
-	}
-	switch mediaType {
-	case ingest.ContentTypeTar, ingest.ContentTypeGzip, ingest.ContentTypeXGzip:
-		return true
-	default:
-		return false
-	}
-}
-
-func readUploadBody(body io.Reader, maxSize int64) ([]byte, error) {
-	limited := &io.LimitedReader{R: body, N: maxSize + 1}
-	data, err := io.ReadAll(limited)
-	if err != nil {
-		return nil, err
-	}
-	if int64(len(data)) > maxSize {
-		return nil, ingest.ErrUploadTooLarge
-	}
-	return data, nil
-}
-
 func writeUploadSubmitError(w http.ResponseWriter, err error) {
 	switch {
 	case IsUploadQueueFull(err):

internal/app/upload_http_integration_test.go

@@ -13,6 +13,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -68,32 +69,65 @@ func TestHTTPUploadPublishesTarAndGzipFanout(t *testing.T) {
 	}
 }
 
-func TestHTTPUploadInvalidArchiveFailsWithoutPublishing(t *testing.T) {
+func TestHTTPUploadInvalidArchiveIsRejectedWithoutRunID(t *testing.T) {
 	destination := t.TempDir()
-	cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
+	coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
 		id:           "reports",
 		tokenEnv:     "REPORTS_TOKEN",
 		stagingPath:  filepath.Join(t.TempDir(), "reports"),
 		destinations: []string{destination},
-	}}, 4, 1)
-	handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{
-		"REPORTS_TOKEN": "reports-secret",
-	}))
-	if err != nil {
-		t.Fatalf("newUploadHTTPHandler() error = %v", err)
+	}}, 4, 1))
+	handler := uploadHTTPHandler{
+		coordinator: coordinator,
+		tokens:      map[string]string{"reports-secret": "reports"},
 	}
 	server := httptest.NewServer(handler)
 	defer server.Close()
 
-	runID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("not a tar archive"))
-	record := waitForHTTPUploadStatus(t, server, runID, UploadStatusFailed)
+	status, body := postHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("not a tar archive"))
+	if status != http.StatusBadRequest {
+		t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusBadRequest, body)
+	}
+	if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") {
+		t.Fatalf("invalid archive response exposed run id or token: %s", body)
+	}
+	if got := coordinator.QueueDepth(); got != 0 {
+		t.Fatalf("queue depth = %d, want 0", got)
+	}
+	assertDirectoryEmpty(t, destination)
+}
 
-	if record.Error == "" {
-		t.Fatal("failed status error is empty")
+func TestHTTPUploadOversizedArchiveIsRejectedWithoutRunID(t *testing.T) {
+	destination := t.TempDir()
+	stagingPath := filepath.Join(t.TempDir(), "reports")
+	cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
+		id:           "reports",
+		tokenEnv:     "REPORTS_TOKEN",
+		stagingPath:  stagingPath,
+		destinations: []string{destination},
+	}}, 4, 1)
+	size := config.ByteSize(4)
+	cfg.Server.HTTP.MaxUploadSize = &size
+	cfg.Pipelines[0].Source.Upload.MaxUploadSize = &size
+	coordinator := NewUploadCoordinator(context.Background(), cfg)
+	handler := uploadHTTPHandler{
+		coordinator: coordinator,
+		tokens:      map[string]string{"reports-secret": "reports"},
 	}
-	if record.Report != nil {
-		t.Fatalf("failed staging report = %#v, want nil", record.Report)
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	status, body := postHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, bundleArchive(t, false, testutil.BundleOptions{}))
+	if status != http.StatusRequestEntityTooLarge {
+		t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusRequestEntityTooLarge, body)
 	}
+	if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") {
+		t.Fatalf("oversized response exposed run id or token: %s", body)
+	}
+	if got := coordinator.QueueDepth(); got != 0 {
+		t.Fatalf("queue depth = %d, want 0", got)
+	}
+	assertDirectoryEmpty(t, stagingPath)
 	assertDirectoryEmpty(t, destination)
 }
 
@@ -122,7 +156,6 @@ func TestHTTPUploadSamePipelineRequestsSerialize(t *testing.T) {
 	handler := uploadHTTPHandler{
 		coordinator: coordinator,
 		tokens:      map[string]string{"reports-secret": "reports"},
-		limits:      map[string]int64{"reports": 1024},
 	}
 	server := httptest.NewServer(handler)
 	defer server.Close()
@@ -177,10 +210,6 @@ func TestHTTPUploadDifferentPipelinesRunConcurrently(t *testing.T) {
 			"one-secret": "reports-one",
 			"two-secret": "reports-two",
 		},
-		limits: map[string]int64{
-			"reports-one": 1024,
-			"reports-two": 1024,
-		},
 	}
 	server := httptest.NewServer(handler)
 	defer server.Close()
@@ -247,6 +276,22 @@ func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpe
 }
 
 func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) UploadRunID {
+	t.Helper()
+	status, responseBody := postHTTPUpload(t, server, token, contentType, body)
+	if status != http.StatusAccepted {
+		t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusAccepted, responseBody)
+	}
+	var accepted uploadAcceptedResponse
+	if err := json.Unmarshal([]byte(responseBody), &accepted); err != nil {
+		t.Fatalf("decode accepted response: %v", err)
+	}
+	if accepted.RunID == "" || accepted.Status != UploadStatusAccepted {
+		t.Fatalf("accepted response = %#v, want run id and accepted status", accepted)
+	}
+	return accepted.RunID
+}
+
+func postHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) (int, string) {
 	t.Helper()
 	request, err := http.NewRequest(http.MethodPost, server.URL+"/upload", bytes.NewReader(body))
 	if err != nil {
@@ -259,17 +304,11 @@ func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType
 		t.Fatalf("POST /upload error = %v", err)
 	}
 	defer response.Body.Close()
-	if response.StatusCode != http.StatusAccepted {
-		t.Fatalf("POST /upload status = %d, want %d", response.StatusCode, http.StatusAccepted)
+	data, err := io.ReadAll(response.Body)
+	if err != nil {
+		t.Fatalf("read response body: %v", err)
 	}
-	var accepted uploadAcceptedResponse
-	if err := json.NewDecoder(response.Body).Decode(&accepted); err != nil {
-		t.Fatalf("decode accepted response: %v", err)
-	}
-	if accepted.RunID == "" || accepted.Status != UploadStatusAccepted {
-		t.Fatalf("accepted response = %#v, want run id and accepted status", accepted)
-	}
-	return accepted.RunID
+	return response.StatusCode, string(data)
 }
 
 func waitForHTTPUploadStatus(t *testing.T, server *httptest.Server, runID UploadRunID, status UploadStatus) UploadRunRecord {

internal/app/upload_http_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
 )
 
 type fakeUploadCoordinator struct {
@@ -41,7 +42,7 @@ func (fake fakeUploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bo
 func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) {
 	cfg := uploadHTTPTestConfig()
 
-	_, _, err := resolveUploadTokens(cfg, config.NewEnvironment(nil, func(string) (string, bool) {
+	_, err := resolveUploadTokens(cfg, config.NewEnvironment(nil, func(string) (string, bool) {
 		return "", false
 	}))
 	if err == nil || !strings.Contains(err.Error(), "UPLOAD_TOKEN") {
@@ -58,7 +59,7 @@ func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) {
 	})
 	config.ApplyDefaults(&cfg)
 	secret := "super-secret-token"
-	_, _, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{
+	_, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{
 		"UPLOAD_TOKEN":       secret,
 		"OTHER_UPLOAD_TOKEN": secret,
 	}))
@@ -110,7 +111,6 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
 			},
 		},
 		tokens: map[string]string{"valid-token": "reports"},
-		limits: map[string]int64{"reports": 1024},
 	}
 	recorder := httptest.NewRecorder()
 	request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
@@ -141,7 +141,6 @@ func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) {
 	handler := uploadHTTPHandler{
 		coordinator: fakeUploadCoordinator{canAccept: true},
 		tokens:      map[string]string{"valid-token": "reports"},
-		limits:      map[string]int64{"reports": 1024},
 	}
 
 	for _, authHeader := range []string{"", "Bearer wrong-token"} {
@@ -178,14 +177,6 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t
 			body:        strings.NewReader("archive"),
 			wantStatus:  http.StatusUnsupportedMediaType,
 		},
-		{
-			name:        "oversized",
-			canAccept:   true,
-			url:         "/upload",
-			contentType: "application/x-tar",
-			body:        strings.NewReader("too-large"),
-			wantStatus:  http.StatusRequestEntityTooLarge,
-		},
 		{
 			name:        "full queue",
 			canAccept:   false,
@@ -214,7 +205,6 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t
 					},
 				},
 				tokens: map[string]string{"valid-token": "reports"},
-				limits: map[string]int64{"reports": 4},
 			}
 			recorder := httptest.NewRecorder()
 			request := httptest.NewRequest(http.MethodPost, tt.url, tt.body)
@@ -233,6 +223,41 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t
 	}
 }
 
+func TestUploadHTTPHandlerMapsSubmitErrors(t *testing.T) {
+	tests := []struct {
+		name       string
+		err        error
+		wantStatus int
+	}{
+		{name: "oversized", err: ingest.ErrUploadTooLarge, wantStatus: http.StatusRequestEntityTooLarge},
+		{name: "unsupported", err: ingest.ErrUnsupportedContentType, wantStatus: http.StatusUnsupportedMediaType},
+		{name: "malformed", err: errors.New("malformed archive"), wantStatus: http.StatusBadRequest},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			handler := uploadHTTPHandler{
+				coordinator: fakeUploadCoordinator{
+					canAccept: true,
+					submit: func(context.Context, UploadRequest) (UploadRunRecord, error) {
+						return UploadRunRecord{}, tt.err
+					},
+				},
+				tokens: map[string]string{"valid-token": "reports"},
+			}
+			recorder := httptest.NewRecorder()
+			request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
+			request.Header.Set("Authorization", "Bearer valid-token")
+			request.Header.Set("Content-Type", "application/x-tar")
+
+			handler.ServeHTTP(recorder, request)
+
+			if recorder.Code != tt.wantStatus {
+				t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String())
+			}
+		})
+	}
+}
+
 func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) {
 	finishedAt := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC)
 	handler := uploadHTTPHandler{
@@ -251,7 +276,6 @@ func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) {
 			},
 		},
 		tokens: map[string]string{"valid-token": "reports"},
-		limits: map[string]int64{"reports": 1024},
 	}
 
 	recorder := httptest.NewRecorder()

internal/app/validate.go

@@ -6,7 +6,6 @@ import (
 	"io"
 
 	"gitea.maximumdirect.net/eric/distributor/internal/config"
-	"gitea.maximumdirect.net/eric/distributor/internal/storage"
 )
 
 type ValidateOptions struct {
@@ -73,29 +72,17 @@ func writeValidateResult(options ValidateOptions, selection sourceSelection) err
 }
 
 type validateResult struct {
-	PipelineID    string                 `json:"pipeline_id,omitempty"`
-	SourceBackend string                 `json:"source_backend,omitempty"`
-	BundleCount   int                    `json:"bundle_count"`
-	Bundles       []validateBundleResult `json:"bundles"`
-}
-
-type validateBundleResult struct {
-	Path string `json:"path"`
-	ID   string `json:"id"`
+	PipelineID    string                `json:"pipeline_id,omitempty"`
+	SourceBackend string                `json:"source_backend,omitempty"`
+	BundleCount   int                   `json:"bundle_count"`
+	Bundles       []bundleSummaryResult `json:"bundles"`
 }
 
 func validateResultFromSelection(selection sourceSelection) validateResult {
-	result := validateResult{
+	return validateResult{
 		PipelineID:    selection.PipelineID,
 		SourceBackend: selection.SourceBackend,
 		BundleCount:   len(selection.Bundles),
-		Bundles:       make([]validateBundleResult, 0, len(selection.Bundles)),
+		Bundles:       bundleSummariesFromBundles(selection.Bundles),
 	}
-	for _, sourceBundle := range selection.Bundles {
-		result.Bundles = append(result.Bundles, validateBundleResult{
-			Path: storage.DisplayPath(sourceBundle.RootRelativePath),
-			ID:   sourceBundle.Manifest.ID,
-		})
-	}
-	return result
 }

internal/app/validate_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"os"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -183,6 +184,51 @@ pipelines:
 	}
 }
 
+func TestValidateConfiguredSourcePrintsSecretConflictWarningWithoutValues(t *testing.T) {
+	name := "DISTRIBUTOR_TEST_VALIDATE_SECRET"
+	t.Setenv(name, "process-value")
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	secretsRoot := t.TempDir()
+	if err := os.WriteFile(filepath.Join(secretsRoot, name), []byte("secret-value\n"), 0o600); err != nil {
+		t.Fatalf("write secret: %v", err)
+	}
+	testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{})
+	configPath := writeConfigFile(t, `
+secrets:
+  directory: `+secretsRoot+`
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: `+sourceRoot+`
+    destinations:
+      - id: archive
+        backend: local
+        path: `+destinationRoot+`
+`)
+
+	var stdout bytes.Buffer
+	err := Validate(context.Background(), ValidateOptions{
+		ConfigPath: configPath,
+		PipelineID: "reports",
+		Stdout:     &stdout,
+	})
+	if err != nil {
+		t.Fatalf("Validate() error = %v", err)
+	}
+	output := stdout.String()
+	if !strings.Contains(output, "secret "+name+" ignored because the real environment already has that variable") {
+		t.Fatalf("stdout = %q, want secret conflict warning", output)
+	}
+	if !strings.Contains(output, "Validated 1 bundle(s) for pipeline reports source local") {
+		t.Fatalf("stdout = %q, want validate summary", output)
+	}
+	if strings.Contains(output, "process-value") || strings.Contains(output, "secret-value") {
+		t.Fatalf("stdout exposed secret values: %q", output)
+	}
+}
+
 func TestValidateConfiguredSourceRequiresPipeline(t *testing.T) {
 	sourceRoot := t.TempDir()
 	destinationRoot := t.TempDir()

internal/bundle/manifest_test.go

@@ -70,11 +70,19 @@ func TestParseManifestRejectsInvalidDigestFormat(t *testing.T) {
 
 func TestParseManifestRejectsUnsafeFilePaths(t *testing.T) {
 	tests := []string{
+		`"path": ""`,
+		`"path": "."`,
+		`"path": "./report.md"`,
 		`"path": "../report.md"`,
 		`"path": "/report.md"`,
 		`"path": "nested/../report.md"`,
+		`"path": "nested/./report.md"`,
+		`"path": "nested//report.md"`,
+		`"path": "nested\\report.md"`,
 		`"path": "manifest.json"`,
+		`"path": "nested/manifest.json"`,
 		`"path": "` + storage.StateFileName + `"`,
+		`"path": "nested/` + storage.StateFileName + `"`,
 	}
 	for _, replacement := range tests {
 		t.Run(replacement, func(t *testing.T) {
@@ -128,6 +136,16 @@ func TestValidateManifestRejectsInvalidManifest(t *testing.T) {
 			manifest.Digest = BundleDigest(manifest.Files)
 			return manifest
 		},
+		"nested manifest path": func(manifest Manifest) Manifest {
+			manifest.Files[0].Path = "nested/manifest.json"
+			manifest.Digest = BundleDigest(manifest.Files)
+			return manifest
+		},
+		"nested state path": func(manifest Manifest) Manifest {
+			manifest.Files[0].Path = "nested/" + storage.StateFileName
+			manifest.Digest = BundleDigest(manifest.Files)
+			return manifest
+		},
 		"duplicate path": func(manifest Manifest) Manifest {
 			manifest.Files[1].Path = manifest.Files[0].Path
 			manifest.Digest = BundleDigest(manifest.Files)

internal/bundle/validate_test.go

@@ -65,6 +65,38 @@ func TestValidateRejectsSymlinkFile(t *testing.T) {
 	assertErrorContains(t, err, "regular file")
 }
 
+func TestValidateRejectsUnsafeManifestPaths(t *testing.T) {
+	tests := []string{
+		"",
+		".",
+		"./report.md",
+		"../report.md",
+		"/report.md",
+		"nested/../report.md",
+		"nested/./report.md",
+		"nested//report.md",
+		`nested\report.md`,
+		ManifestName,
+		storage.StateFileName,
+		"nested/" + ManifestName,
+		"nested/" + storage.StateFileName,
+	}
+	for _, path := range tests {
+		t.Run(path, func(t *testing.T) {
+			backend := validFakeBundle(t)
+			manifest := validFixtureManifest(t)
+			manifest.Files[0].Path = path
+			manifest.Digest = BundleDigest(manifest.Files)
+			writeManifest(t, backend, manifest)
+
+			_, err := Validate(context.Background(), backend, "")
+			if err == nil {
+				t.Fatal("Validate() error = nil, want unsafe path error")
+			}
+		})
+	}
+}
+
 func validFakeBundle(t *testing.T) *fake.Backend {
 	t.Helper()
 	backend := fake.New()

internal/cli/flags.go

@@ -0,0 +1,12 @@
+package cli
+
+import (
+	"flag"
+	"io"
+)
+
+func newFlagSet(name string, stderr io.Writer) *flag.FlagSet {
+	flags := flag.NewFlagSet(name, flag.ContinueOnError)
+	flags.SetOutput(stderr)
+	return flags
+}

internal/cli/manifest.go

@@ -2,7 +2,6 @@ package cli
 
 import (
 	"context"
-	"flag"
 	"fmt"
 	"io"
 	"strings"
@@ -30,8 +29,7 @@ func manifestCreateCommand(ctx context.Context, args []string, stdout, stderr io
 		printManifestCreateHelp(stdout)
 		return exitOK
 	}
-	flags := flag.NewFlagSet("manifest create", flag.ContinueOnError)
-	flags.SetOutput(stderr)
+	flags := newFlagSet("manifest create", stderr)
 	id := flags.String("id", "", "source bundle id")
 	created := flags.String("created", "", "source created timestamp")
 	overwrite := flags.Bool("overwrite", false, "replace an existing manifest.json")

internal/cli/run.go

@@ -2,7 +2,6 @@ package cli
 
 import (
 	"context"
-	"flag"
 	"fmt"
 	"io"
 
@@ -15,8 +14,7 @@ func runCommand(ctx context.Context, args []string, stdout, stderr io.Writer) in
 		return exitOK
 	}
 
-	flags := flag.NewFlagSet("run", flag.ContinueOnError)
-	flags.SetOutput(stderr)
+	flags := newFlagSet("run", stderr)
 	configPath := flags.String("config", "", "path to config file")
 	dryRun := flags.Bool("dry-run", false, "load and validate config without publishing")
 	force := flags.Bool("force", false, "allow explicit destructive replacement for supported conflicts")

internal/cli/serve.go

@@ -2,7 +2,6 @@ package cli
 
 import (
 	"context"
-	"flag"
 	"fmt"
 	"io"
 
@@ -17,8 +16,7 @@ func serveCommand(ctx context.Context, args []string, stdout, stderr io.Writer)
 		return exitOK
 	}
 
-	flags := flag.NewFlagSet("serve", flag.ContinueOnError)
-	flags.SetOutput(stderr)
+	flags := newFlagSet("serve", stderr)
 	configPath := flags.String("config", "", "path to config file")
 	if err := flags.Parse(args); err != nil {
 		return exitUsage

internal/cli/source_mode.go

@@ -1,7 +1,6 @@
 package cli
 
 import (
-	"flag"
 	"fmt"
 	"io"
 
@@ -17,8 +16,7 @@ type sourceDiagnosticArgs struct {
 }
 
 func parseSourceDiagnosticArgs(stderr io.Writer, command string, args []string) (sourceDiagnosticArgs, bool) {
-	flags := flag.NewFlagSet(command, flag.ContinueOnError)
-	flags.SetOutput(stderr)
+	flags := newFlagSet(command, stderr)
 	configPath := flags.String("config", "", "path to config file")
 	pipelineID := flags.String("pipeline", "", "pipeline id")
 	bundlePath := flags.String("bundle", "", "source-root-relative bundle path")

internal/cli/version.go

@@ -2,7 +2,6 @@ package cli
 
 import (
 	"context"
-	"flag"
 	"fmt"
 	"io"
 
@@ -14,8 +13,7 @@ func versionCommand(_ context.Context, args []string, stdout, stderr io.Writer)
 		printVersionHelp(stdout)
 		return exitOK
 	}
-	flags := flag.NewFlagSet("version", flag.ContinueOnError)
-	flags.SetOutput(stderr)
+	flags := newFlagSet("version", stderr)
 	formatFlag := addFormatFlag(flags)
 	if err := flags.Parse(args); err != nil {
 		return exitUsage

internal/config/backend_view.go

@@ -0,0 +1,50 @@
+package config
+
+type backendView struct {
+	Backend   string
+	Host      string
+	User      string
+	Port      int
+	Path      string
+	Endpoint  string
+	Bucket    string
+	Prefix    string
+	Region    string
+	ForcePath *bool
+	Creds     Credentials
+	SSH       SSH
+}
+
+func backendViewFromSource(source Backend) backendView {
+	return backendView{
+		Backend:   source.Backend,
+		Host:      source.Host,
+		User:      source.User,
+		Port:      source.Port,
+		Path:      source.Path,
+		Endpoint:  source.Endpoint,
+		Bucket:    source.Bucket,
+		Prefix:    source.Prefix,
+		Region:    source.Region,
+		ForcePath: source.ForcePath,
+		Creds:     source.Creds,
+		SSH:       source.SSH,
+	}
+}
+
+func backendViewFromDestination(destination Destination) backendView {
+	return backendView{
+		Backend:   destination.Backend,
+		Host:      destination.Host,
+		User:      destination.User,
+		Port:      destination.Port,
+		Path:      destination.Path,
+		Endpoint:  destination.Endpoint,
+		Bucket:    destination.Bucket,
+		Prefix:    destination.Prefix,
+		Region:    destination.Region,
+		ForcePath: destination.ForcePath,
+		Creds:     destination.Creds,
+		SSH:       destination.SSH,
+	}
+}

internal/config/backend_view_test.go

@@ -0,0 +1,187 @@
+package config
+
+import "testing"
+
+func TestBackendViewsPreserveEquivalentStorageFields(t *testing.T) {
+	forcePathStyle := false
+	source := Backend{
+		Backend:   BackendS3,
+		Host:      "storage.example.com",
+		User:      "reports",
+		Port:      2222,
+		Path:      "/reports",
+		Endpoint:  "https://s3.example.com",
+		Bucket:    "source",
+		Prefix:    "incoming",
+		Region:    "us-west-2",
+		ForcePath: &forcePathStyle,
+		Creds: Credentials{
+			AccessKeyIDEnv:     "ACCESS_KEY_ID",
+			SecretAccessKeyEnv: "SECRET_ACCESS_KEY",
+		},
+		SSH: SSH{
+			KeyFile:       "/home/reports/.ssh/id_ed25519",
+			KnownHosts:    "/home/reports/.ssh/known_hosts",
+			HostKeyPolicy: HostKeyPolicyStrict,
+		},
+	}
+	destination := Destination{
+		Backend:   source.Backend,
+		Host:      source.Host,
+		User:      source.User,
+		Port:      source.Port,
+		Path:      source.Path,
+		Endpoint:  source.Endpoint,
+		Bucket:    source.Bucket,
+		Prefix:    source.Prefix,
+		Region:    source.Region,
+		ForcePath: source.ForcePath,
+		Creds:     source.Creds,
+		SSH:       source.SSH,
+	}
+
+	sourceView := backendViewFromSource(source)
+	destinationView := backendViewFromDestination(destination)
+
+	if sourceView != destinationView {
+		t.Fatalf("source view = %#v, destination view = %#v, want equivalent storage fields", sourceView, destinationView)
+	}
+}
+
+func TestBackendViewValidationKeepsHTTPUploadSourceOnly(t *testing.T) {
+	cfg := Config{Pipelines: []Pipeline{{
+		ID: "reports",
+		Source: Backend{
+			Backend: BackendHTTPUpload,
+			Upload:  HTTPUpload{TokenEnv: "UPLOAD_TOKEN"},
+		},
+		Destinations: []Destination{{
+			ID:      "archive",
+			Backend: BackendHTTPUpload,
+		}},
+	}}}
+	ApplyDefaults(&cfg)
+
+	err := Validate(cfg)
+	if err == nil {
+		t.Fatal("Validate() error = nil, want destination http_upload error")
+	}
+	if got, want := err.Error(), "pipelines[0].destinations[0].backend http_upload is only supported for sources"; got != want {
+		t.Fatalf("Validate() error = %q, want %q", got, want)
+	}
+}
+
+func TestBackendViewValidationAppliesStorageRulesToSourcesAndDestinations(t *testing.T) {
+	forcePathStyle := false
+	tests := []struct {
+		name        string
+		source      Backend
+		destination Destination
+		wantErr     bool
+	}{
+		{
+			name: "local valid",
+			source: Backend{
+				Backend: BackendLocal,
+				Path:    "/source",
+			},
+			destination: Destination{
+				Backend: BackendLocal,
+				Path:    "/destination",
+			},
+		},
+		{
+			name: "local missing path",
+			source: Backend{
+				Backend: BackendLocal,
+			},
+			destination: Destination{
+				Backend: BackendLocal,
+			},
+			wantErr: true,
+		},
+		{
+			name: "ssh valid",
+			source: Backend{
+				Backend: BackendSSH,
+				Host:    "source.example.com",
+				Port:    22,
+				Path:    "/source",
+				SSH:     SSH{HostKeyPolicy: HostKeyPolicyAcceptNew},
+			},
+			destination: Destination{
+				Backend: BackendSSH,
+				Host:    "destination.example.com",
+				Port:    22,
+				Path:    "/destination",
+				SSH:     SSH{HostKeyPolicy: HostKeyPolicyAcceptNew},
+			},
+		},
+		{
+			name: "ssh missing host",
+			source: Backend{
+				Backend: BackendSSH,
+				Port:    22,
+				Path:    "/source",
+				SSH:     SSH{HostKeyPolicy: HostKeyPolicyAcceptNew},
+			},
+			destination: Destination{
+				Backend: BackendSSH,
+				Port:    22,
+				Path:    "/destination",
+				SSH:     SSH{HostKeyPolicy: HostKeyPolicyAcceptNew},
+			},
+			wantErr: true,
+		},
+		{
+			name: "s3 valid",
+			source: Backend{
+				Backend:   BackendS3,
+				Endpoint:  "https://s3.example.com",
+				Bucket:    "source",
+				Prefix:    "incoming",
+				Region:    DefaultS3Region,
+				ForcePath: &forcePathStyle,
+			},
+			destination: Destination{
+				Backend:   BackendS3,
+				Endpoint:  "https://s3.example.com",
+				Bucket:    "destination",
+				Prefix:    "archive",
+				Region:    DefaultS3Region,
+				ForcePath: &forcePathStyle,
+			},
+		},
+		{
+			name: "s3 partial credentials",
+			source: Backend{
+				Backend:  BackendS3,
+				Endpoint: "https://s3.example.com",
+				Bucket:   "source",
+				Region:   DefaultS3Region,
+				Creds:    Credentials{AccessKeyIDEnv: "ACCESS_KEY_ID"},
+			},
+			destination: Destination{
+				Backend:  BackendS3,
+				Endpoint: "https://s3.example.com",
+				Bucket:   "destination",
+				Region:   DefaultS3Region,
+				Creds:    Credentials{AccessKeyIDEnv: "ACCESS_KEY_ID"},
+			},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			sourceErrors := validateBackend(nil, "source", backendViewFromSource(tt.source))
+			destinationErrors := validateBackend(nil, "destination", backendViewFromDestination(tt.destination))
+			if got := len(sourceErrors) > 0; got != tt.wantErr {
+				t.Fatalf("source validation errors = %#v, wantErr %t", sourceErrors, tt.wantErr)
+			}
+			if got := len(destinationErrors) > 0; got != tt.wantErr {
+				t.Fatalf("destination validation errors = %#v, wantErr %t", destinationErrors, tt.wantErr)
+			}
+		})
+	}
+}

internal/config/defaults.go

@@ -134,30 +134,24 @@ func duration(value Duration) *Duration {
 }
 
 func applyBackendDefaults(backend *Backend) {
-	if backend.Backend == BackendSSH {
-		if backend.Port == 0 {
-			backend.Port = 22
-		}
-		if backend.SSH.HostKeyPolicy == "" {
-			backend.SSH.HostKeyPolicy = HostKeyPolicyAcceptNew
-		}
-	}
-	if backend.Backend == BackendS3 {
-		applyS3Defaults(&backend.Region, &backend.Prefix, &backend.ForcePath)
-	}
+	applyStorageBackendDefaults(backend.Backend, &backend.Port, &backend.SSH, &backend.Region, &backend.Prefix, &backend.ForcePath)
 }
 
 func applyDestinationDefaults(destination *Destination) {
-	if destination.Backend == BackendSSH {
-		if destination.Port == 0 {
-			destination.Port = 22
+	applyStorageBackendDefaults(destination.Backend, &destination.Port, &destination.SSH, &destination.Region, &destination.Prefix, &destination.ForcePath)
+}
+
+func applyStorageBackendDefaults(backend string, port *int, ssh *SSH, region, prefix *string, forcePath **bool) {
+	if backend == BackendSSH {
+		if *port == 0 {
+			*port = 22
 		}
-		if destination.SSH.HostKeyPolicy == "" {
-			destination.SSH.HostKeyPolicy = HostKeyPolicyAcceptNew
+		if ssh.HostKeyPolicy == "" {
+			ssh.HostKeyPolicy = HostKeyPolicyAcceptNew
 		}
 	}
-	if destination.Backend == BackendS3 {
-		applyS3Defaults(&destination.Region, &destination.Prefix, &destination.ForcePath)
+	if backend == BackendS3 {
+		applyS3Defaults(region, prefix, forcePath)
 	}
 }
 

internal/config/validate.go

@@ -100,7 +100,7 @@ func validateSourceBackend(errs ValidationErrors, context string, backend Backen
 	if backend.Backend == BackendHTTPUpload {
 		return validateHTTPUploadSource(errs, context, backend.Upload)
 	}
-	return validateBackend(errs, context, backend.Backend, backend.Host, backend.Port, backend.Path, backend.Endpoint, backend.Bucket, backend.Prefix, backend.SSH.HostKeyPolicy, backend.Creds)
+	return validateBackend(errs, context, backendViewFromSource(backend))
 }
 
 func validateDestinationBackend(errs ValidationErrors, context string, destination Destination) ValidationErrors {
@@ -108,7 +108,7 @@ func validateDestinationBackend(errs ValidationErrors, context string, destinati
 		errs = append(errs, context+".backend "+BackendHTTPUpload+" is only supported for sources")
 		return errs
 	}
-	return validateBackend(errs, context, destination.Backend, destination.Host, destination.Port, destination.Path, destination.Endpoint, destination.Bucket, destination.Prefix, destination.SSH.HostKeyPolicy, destination.Creds)
+	return validateBackend(errs, context, backendViewFromDestination(destination))
 }
 
 func validateHTTPUploadSource(errs ValidationErrors, context string, upload HTTPUpload) ValidationErrors {
@@ -124,47 +124,47 @@ func validateHTTPUploadSource(errs ValidationErrors, context string, upload HTTP
 	return errs
 }
 
-func validateBackend(errs ValidationErrors, context, backend, host string, port int, path, endpoint, bucket, prefix string, hostKeyPolicy HostKeyPolicy, creds Credentials) ValidationErrors {
-	switch backend {
+func validateBackend(errs ValidationErrors, context string, backend backendView) ValidationErrors {
+	switch backend.Backend {
 	case "":
 		errs = append(errs, context+".backend is required")
 	case BackendLocal:
-		if path == "" {
+		if backend.Path == "" {
 			errs = append(errs, context+".path is required for local backend")
 		}
 	case BackendSSH:
-		if host == "" {
+		if backend.Host == "" {
 			errs = append(errs, context+".host is required for ssh backend")
 		}
-		if path == "" {
+		if backend.Path == "" {
 			errs = append(errs, context+".path is required for ssh backend")
 		}
-		if port < 0 || port > 65535 {
+		if backend.Port < 0 || backend.Port > 65535 {
 			errs = append(errs, context+".port must be between 1 and 65535")
 		}
-		if port == 0 {
+		if backend.Port == 0 {
 			errs = append(errs, context+".port is required for ssh backend after defaults are applied")
 		}
-		if hostKeyPolicy != "" {
-			if _, ok := NormalizeHostKeyPolicy(string(hostKeyPolicy)); !ok {
+		if backend.SSH.HostKeyPolicy != "" {
+			if _, ok := NormalizeHostKeyPolicy(string(backend.SSH.HostKeyPolicy)); !ok {
 				errs = append(errs, context+".host_key_policy must be strict, true, accept-new, off, or false")
 			}
 		}
 	case BackendS3:
-		if endpoint == "" {
+		if backend.Endpoint == "" {
 			errs = append(errs, context+".endpoint is required for s3 backend")
 		}
-		if bucket == "" {
+		if backend.Bucket == "" {
 			errs = append(errs, context+".bucket is required for s3 backend")
 		}
-		if err := ValidateS3Prefix(prefix); err != nil {
+		if err := ValidateS3Prefix(backend.Prefix); err != nil {
 			errs = append(errs, context+".prefix must be a clean relative slash-separated path")
 		}
-		if (creds.AccessKeyIDEnv == "") != (creds.SecretAccessKeyEnv == "") {
+		if (backend.Creds.AccessKeyIDEnv == "") != (backend.Creds.SecretAccessKeyEnv == "") {
 			errs = append(errs, context+".credentials.access_key_id_env and credentials.secret_access_key_env must be configured together")
 		}
 	default:
-		errs = append(errs, context+".backend "+backend+" is unsupported")
+		errs = append(errs, context+".backend "+backend.Backend+" is unsupported")
 	}
 	return errs
 }

internal/ingest/archive.go

@@ -138,6 +138,11 @@ func validateRunID(value string) error {
 	return nil
 }
 
+func ValidateContentType(contentType string) error {
+	_, err := archiveFormat(contentType)
+	return err
+}
+
 type archiveKind int
 
 const (

internal/ingest/archive_test.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"compress/gzip"
 	"context"
+	"encoding/json"
 	"errors"
 	"io/fs"
 	"os"
@@ -47,6 +48,25 @@ func TestStageArchiveRejectsUnsupportedContentType(t *testing.T) {
 	}
 }
 
+func TestValidateContentType(t *testing.T) {
+	for _, contentType := range []string{
+		ContentTypeTar,
+		ContentTypeGzip,
+		ContentTypeXGzip,
+		ContentTypeGzip + "; charset=binary",
+	} {
+		t.Run(contentType, func(t *testing.T) {
+			if err := ValidateContentType(contentType); err != nil {
+				t.Fatalf("ValidateContentType() error = %v", err)
+			}
+		})
+	}
+
+	if err := ValidateContentType("application/zip"); !errors.Is(err, ErrUnsupportedContentType) {
+		t.Fatalf("ValidateContentType() error = %v, want ErrUnsupportedContentType", err)
+	}
+}
+
 func TestStageArchiveEnforcesMaxUploadSize(t *testing.T) {
 	archive := validArchive(t, false)
 	err := stageArchiveError(t, archive, ContentTypeTar, func(opts *StageOptions) {
@@ -94,9 +114,19 @@ func TestStageArchiveRejectsUnsafeEntries(t *testing.T) {
 		"path traversal": {
 			fileEntry("../report.md", "report"),
 		},
+		"dot path": {
+			fileEntry("./report.md", "report"),
+		},
+		"dot segment": {
+			fileEntry("nested/./report.md", "report"),
+		},
 		"backslash path": {
 			fileEntry(`nested\report.md`, "report"),
 		},
+		"duplicate file": {
+			fileEntry("report.md", "report"),
+			fileEntry("report.md", "report"),
+		},
 		"symlink": {
 			{name: "link.md", typeflag: tar.TypeSymlink, linkname: "report.md"},
 		},
@@ -106,6 +136,12 @@ func TestStageArchiveRejectsUnsafeEntries(t *testing.T) {
 		"device": {
 			{name: "device", typeflag: tar.TypeChar},
 		},
+		"fifo": {
+			{name: "socket", typeflag: tar.TypeFifo},
+		},
+		"socket": {
+			{name: "socket", typeflag: 'S'},
+		},
 	}
 	for name, entries := range tests {
 		t.Run(name, func(t *testing.T) {
@@ -127,6 +163,14 @@ func TestStageArchiveRejectsBundleValidationFailures(t *testing.T) {
 			fileEntry("nested/manifest.json", "{}"),
 			fileEntry("report.md", "report"),
 		},
+		"listed nested manifest": {
+			fileEntry("manifest.json", uncheckedManifestJSON(t, manifestFor("reports.listed.nested", fileSpec{path: "nested/manifest.json", body: "{}"}))),
+			fileEntry("nested/manifest.json", "{}"),
+		},
+		"listed state file": {
+			fileEntry("manifest.json", uncheckedManifestJSON(t, manifestFor("reports.listed.state", fileSpec{path: ".distributor.json", body: "{}"}))),
+			fileEntry(".distributor.json", "{}"),
+		},
 		"missing listed file": {
 			fileEntry("manifest.json", manifestJSON(t, manifestFor("reports.missing", fileSpec{path: "missing.md", body: "missing"}))),
 		},
@@ -149,6 +193,20 @@ func TestStageArchiveRejectsBundleValidationFailures(t *testing.T) {
 	}
 }
 
+func TestStageArchiveAcceptsSafeDirectories(t *testing.T) {
+	archive := makeArchive(t, false,
+		tarEntry{name: "nested", typeflag: tar.TypeDir},
+		tarEntry{name: "nested/assets", typeflag: tar.TypeDir},
+		fileEntry("manifest.json", manifestJSON(t, manifestFor("reports.directories", fileSpec{path: "nested/assets/report.md", body: "report"}))),
+		fileEntry("nested/assets/report.md", "report"),
+	)
+	staged := stageArchive(t, archive, ContentTypeTar)
+
+	if got := readFile(t, staged.Root, "nested/assets/report.md"); got != "report" {
+		t.Fatalf("report = %q", got)
+	}
+}
+
 func TestStageArchiveCleansUpFailedExtraction(t *testing.T) {
 	stagingPath := filepath.Join(t.TempDir(), "staging")
 	archive := makeArchive(t, false, fileEntry("../report.md", "report"))
@@ -336,6 +394,15 @@ func manifestJSON(t *testing.T, manifest sourcebundle.Manifest) string {
 	return string(data)
 }
 
+func uncheckedManifestJSON(t *testing.T, manifest sourcebundle.Manifest) string {
+	t.Helper()
+	data, err := json.MarshalIndent(manifest, "", "  ")
+	if err != nil {
+		t.Fatalf("MarshalIndent() error = %v", err)
+	}
+	return string(append(data, '\n'))
+}
+
 func writeFile(t *testing.T, root, relative, body string) {
 	t.Helper()
 	fullPath := filepath.Join(root, filepath.FromSlash(relative))

internal/storage/path_test.go

@@ -12,6 +12,10 @@ func TestValidatePath(t *testing.T) {
 		"report.md",
 		"daily/report.md",
 		"a-b_1.2/report.html",
+		"manifest.json",
+		StateFileName,
+		"nested/manifest.json",
+		"nested/" + StateFileName,
 	}
 	for _, path := range valid {
 		t.Run("valid "+path, func(t *testing.T) {
@@ -23,10 +27,14 @@ func TestValidatePath(t *testing.T) {
 
 	invalid := []string{
 		"",
+		".",
+		"./report.md",
 		"/absolute",
 		"../outside",
 		"nested/../outside",
+		"nested/.",
 		"nested/./file",
+		"nested/",
 		"nested//file",
 		`nested\file`,
 	}

pkg/bundle/bundle_test.go

@@ -90,17 +90,33 @@ func TestBuildManifestRequiresOneFileMode(t *testing.T) {
 	}
 }
 
-func TestBuildManifestRejectsUnsafePath(t *testing.T) {
+func TestBuildManifestRejectsUnsafeExplicitPaths(t *testing.T) {
 	root := t.TempDir()
 	writeFile(t, root, "report.txt", "report")
 
-	_, err := BuildManifest(BuildOptions{
-		Root:  root,
-		ID:    "reports.unsafe",
-		Files: []string{"../report.txt"},
-	})
-	if err == nil {
-		t.Fatal("BuildManifest() error = nil, want unsafe path error")
+	tests := []string{
+		"",
+		"../report.txt",
+		"/report.txt",
+		"nested/../report.txt",
+		"nested/./report.txt",
+		`nested\report.txt`,
+		ManifestName,
+		distributorStateName,
+		"nested/" + ManifestName,
+		"nested/" + distributorStateName,
+	}
+	for _, sourcePath := range tests {
+		t.Run(sourcePath, func(t *testing.T) {
+			_, err := BuildManifest(BuildOptions{
+				Root:  root,
+				ID:    "reports.unsafe",
+				Files: []string{sourcePath},
+			})
+			if err == nil {
+				t.Fatal("BuildManifest() error = nil, want unsafe path error")
+			}
+		})
 	}
 }
 
@@ -330,7 +346,21 @@ func TestValidateSourcePath(t *testing.T) {
 			t.Fatalf("ValidateSourcePath(%q) error = %v", path, err)
 		}
 	}
-	invalid := []string{"", "../report.md", "/report.md", "nested/../report.md", `nested\report.md`, ManifestName, distributorStateName}
+	invalid := []string{
+		"",
+		".",
+		"./report.md",
+		"../report.md",
+		"/report.md",
+		"nested/../report.md",
+		"nested/./report.md",
+		"nested//report.md",
+		`nested\report.md`,
+		ManifestName,
+		distributorStateName,
+		"nested/" + ManifestName,
+		"nested/" + distributorStateName,
+	}
 	for _, path := range invalid {
 		if err := ValidateSourcePath(path); err == nil {
 			t.Fatalf("ValidateSourcePath(%q) error = nil, want error", path)

pkg/bundle/path.go

@@ -23,8 +23,7 @@ func ValidateSourcePath(value string) error {
 			return fmt.Errorf("source path %q must be a clean relative slash-separated path", value)
 		}
 	}
-	switch value {
-	case ManifestName, distributorStateName:
+	if path.Base(value) == ManifestName || path.Base(value) == distributorStateName {
 		return fmt.Errorf("%q is reserved", value)
 	}
 	return nil