Centralize runtime config setup

docs/internal/app.md

@@ -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
+source. Configured-source mode uses the same runtime config and secret setup as
-destination backends.
+run workflows, shares source backend construction, and never opens destination
+backends.
 
-`Serve` is the CLI-facing HTTP upload server entrypoint. It loads config,
+`Serve` is the CLI-facing HTTP upload server entrypoint. It uses the app
-loads the configured secrets directory, resolves upload bearer tokens for
+runtime setup, resolves upload bearer tokens for configured `http_upload`
-configured `http_upload` sources, creates an `UploadCoordinator`, binds
+sources, creates an `UploadCoordinator`, binds `server.http.bind`, and serves
-`server.http.bind`, and serves the upload API until its context is cancelled.
+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`;
+1. builds runtime setup by resolving the config path, loading config, loading
-2. loads configured secret files into a config-owned environment resolver;
+   configured secret files, and projecting secret-conflict warnings;
-3. builds the app-level backend factory and transform registry;
+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;
@@ -195,6 +198,8 @@ 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_output.go`: `RunReport`, action/output records, and text/JSON report projection.
 - `run_summary.go`: summary counters.
@@ -205,7 +210,7 @@ Run helpers are grouped by responsibility:
 - `run_coordinator.go`: in-memory run admission, run IDs, status records, and duplicate-run errors.
 - `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`.

internal/app/inspect_test.go

@@ -3,6 +3,8 @@ package app
 import (
 	"bytes"
 	"context"
+	"encoding/json"
+	"os"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -85,6 +87,55 @@ func TestInspectConfiguredSourceJSON(t *testing.T) {
 	}
 }
 
+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/run.go

@@ -46,15 +46,11 @@ func Run(ctx context.Context, options RunOptions) error {
 		return err
 	}
 
-	configPath := options.ConfigPath
+	setup, err := loadRuntimeSetup(options.ConfigPath)
-	if configPath == "" {
-		configPath = config.DefaultConfigPath
-	}
-	cfg, err := config.LoadFile(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 +58,11 @@ func RunPipeline(ctx context.Context, options RunPipelineOptions) (RunReport, er
 		return RunReport{}, err
 	}
 
-	configPath := options.ConfigPath
+	setup, err := loadRuntimeSetup(options.ConfigPath)
-	if configPath == "" {
-		configPath = config.DefaultConfigPath
-	}
-	cfg, err := config.LoadFile(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 +73,81 @@ func RunPipelineWithLocalSource(ctx context.Context, options RunPipelineWithLoca
 		return RunReport{}, fmt.Errorf("source root is required")
 	}
 
-	configPath := options.ConfigPath
+	setup, err := loadRuntimeSetup(options.ConfigPath)
-	if configPath == "" {
-		configPath = config.DefaultConfigPath
-	}
-	cfg, err := config.LoadFile(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{
+	return buildRunReportWithSetup(ctx, setup.withPipelines([]config.Pipeline{pipeline}), RunOptions{
-		Server:    cfg.Server,
-		Secrets:   cfg.Secrets,
-		Pipelines: []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{
+	return buildRunReportWithSetup(ctx, setup.withPipelines([]config.Pipeline{pipeline}), RunOptions{
-		Server:    cfg.Server,
-		Secrets:   cfg.Secrets,
-		Pipelines: []config.Pipeline{pipeline},
-	}, RunOptions{
 		DryRun:   options.DryRun,
 		Force:    options.Force,
 		Notifier: options.Notifier,
@@ -142,7 +158,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 +181,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 +194,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 +213,12 @@ func buildRunReport(ctx context.Context, cfg config.Config, options RunOptions,
 		Actions:   []RunActionRecord{},
 	}
 	var failures runFailures
-	secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil)
+	report.PreambleWarnings = append(report.PreambleWarnings, setup.Warnings...)
-	if err != nil {
+	report.addWarnings(setup.Warnings)
-		return report, err
+	backends := provider(setup.Environment)
-	}
-	secretWarnings := secretConflictWarnings(secretLoad.Conflicts)
-	report.PreambleWarnings = append(report.PreambleWarnings, secretWarnings...)
-	report.addWarnings(secretWarnings)
-	backends := provider(secretLoad.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)

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
+	setup, err := loadRuntimeSetup(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)
 	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)
+	pipeline, ok := findPipeline(setup.Config, options.PipelineID)
-	if err != nil {
-		return sourceSelection{}, err
-	}
-	pipeline, ok := findPipeline(cfg, 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/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()