Add configured source validation and inspection

2026-06-01 21:11:35 +00:00
parent 8b1e5abf68
commit 29fd0e494c
13 changed files with 728 additions and 50 deletions
--- a/internal/app/inspect.go
+++ b/internal/app/inspect.go
@@ -5,40 +5,70 @@ import (
 	"fmt"
 	"io"

-	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
 	"gitea.maximumdirect.net/eric/distributor/internal/storage"
 )

 type InspectOptions struct {
 	Path         string
+	ConfigPath   string
+	PipelineID   string
+	BundlePath   string
 	Stdout       io.Writer
 	OutputFormat OutputFormat
 }

 func Inspect(ctx context.Context, options InspectOptions) error {
+	return inspectWithBackendFactory(ctx, options, newBackendFactoryWithEnvironment)
+}
+
+func inspectWithBackendFactory(ctx context.Context, options InspectOptions, provider backendFactoryProvider) error {
 	if err := ValidateOutputFormat(options.OutputFormat); err != nil {
 		return err
 	}
-	if options.Path == "" {
-		return fmt.Errorf("inspect command requires a path")
-	}
-	backend, err := newBackendFactory().openLocalPath(ctx, options.Path)
+	selection, err := selectSourceBundles(ctx, sourceCommandOptions{
+		CommandName: "inspect",
+		Path:        options.Path,
+		ConfigPath:  options.ConfigPath,
+		PipelineID:  options.PipelineID,
+		BundlePath:  options.BundlePath,
+	}, provider)
 	if err != nil {
 		return err
 	}
-	bundles, err := bundle.Discover(ctx, backend, "")
+	return writeInspectResult(options, selection)
+}
+
+func inspectConfigWithBackendFactory(ctx context.Context, cfg config.Config, options InspectOptions, provider backendFactoryProvider) error {
+	if err := ValidateOutputFormat(options.OutputFormat); err != nil {
+		return err
+	}
+	selection, err := selectSourceBundlesFromConfig(ctx, cfg, sourceCommandOptions{
+		CommandName: "inspect",
+		PipelineID:  options.PipelineID,
+		BundlePath:  options.BundlePath,
+	}, provider)
 	if err != nil {
 		return err
 	}
+	return writeInspectResult(options, selection)
+}
+
+func writeInspectResult(options InspectOptions, selection sourceSelection) error {
 	if IsJSONOutput(options.OutputFormat) {
-		return WriteJSONEnvelope(options.Stdout, "inspect", true, nil, inspectResultFromBundles(bundles), nil)
+		return WriteJSONEnvelope(options.Stdout, "inspect", true, selection.Warnings, inspectResultFromSelection(selection), nil)
 	}
-	return writeInspection(options.Stdout, bundles)
+	if err := writeWarnings(options.Stdout, selection.Warnings); err != nil {
+		return err
+	}
+	return writeInspection(options.Stdout, selection)
 }

 type inspectResult struct {
-	BundleCount int                   `json:"bundle_count"`
-	Bundles     []inspectBundleResult `json:"bundles"`
+	PipelineID    string                `json:"pipeline_id,omitempty"`
+	SourceBackend string                `json:"source_backend,omitempty"`
+	BundleCount   int                   `json:"bundle_count"`
+	Bundles       []inspectBundleResult `json:"bundles"`
 }

 type inspectBundleResult struct {
@@ -57,12 +87,14 @@ type inspectFileResult struct {
 	Size   int64  `json:"size"`
 }

-func inspectResultFromBundles(bundles []bundle.Bundle) inspectResult {
+func inspectResultFromSelection(selection sourceSelection) inspectResult {
 	result := inspectResult{
-		BundleCount: len(bundles),
-		Bundles:     make([]inspectBundleResult, 0, len(bundles)),
+		PipelineID:    selection.PipelineID,
+		SourceBackend: selection.SourceBackend,
+		BundleCount:   len(selection.Bundles),
+		Bundles:       make([]inspectBundleResult, 0, len(selection.Bundles)),
 	}
-	for _, sourceBundle := range bundles {
+	for _, sourceBundle := range selection.Bundles {
 		bundleResult := inspectBundleResult{
 			Path:      storage.DisplayPath(sourceBundle.RootRelativePath),
 			ID:        sourceBundle.Manifest.ID,
@@ -84,14 +116,19 @@ func inspectResultFromBundles(bundles []bundle.Bundle) inspectResult {
 	return result
 }

-func writeInspection(w io.Writer, bundles []bundle.Bundle) error {
+func writeInspection(w io.Writer, selection sourceSelection) error {
 	if w == nil {
 		return nil
 	}
-	if _, err := fmt.Fprintf(w, "Bundles: %d\n", len(bundles)); err != nil {
+	if selection.ConfigMode {
+		if _, err := fmt.Fprintf(w, "Pipeline: %s\nSource: %s\n", selection.PipelineID, selection.SourceBackend); err != nil {
+			return err
+		}
+	}
+	if _, err := fmt.Fprintf(w, "Bundles: %d\n", len(selection.Bundles)); err != nil {
 		return err
 	}
-	for _, sourceBundle := range bundles {
+	for _, sourceBundle := range selection.Bundles {
 		if _, err := fmt.Fprintf(
 			w,
 			"- path=%s id=%s created=%s digest=%s files=%d\n",
--- a/internal/app/inspect_test.go
+++ b/internal/app/inspect_test.go
@@ -6,6 +6,8 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
 )

 func TestInspectPrintsBundleSummary(t *testing.T) {
@@ -32,6 +34,57 @@ func TestInspectPrintsBundleSummary(t *testing.T) {
 	}
 }

+func TestInspectConfiguredLocalSource(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	testutil.WriteSourceBundle(t, sourceRoot, "daily", testutil.BundleOptions{ID: "reports.daily"})
+	var stdout bytes.Buffer
+
+	err := Inspect(context.Background(), InspectOptions{
+		ConfigPath: testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot),
+		PipelineID: "reports",
+		Stdout:     &stdout,
+	})
+
+	if err != nil {
+		t.Fatalf("Inspect() configured source error = %v", err)
+	}
+	output := stdout.String()
+	for _, want := range []string{
+		"Pipeline: reports",
+		"Source: local",
+		"Bundles: 1",
+		"path=daily",
+		"id=reports.daily",
+	} {
+		if !strings.Contains(output, want) {
+			t.Fatalf("Inspect() output = %q, want substring %q", output, want)
+		}
+	}
+}
+
+func TestInspectConfiguredSourceJSON(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{ID: "reports.json"})
+	var stdout bytes.Buffer
+
+	err := Inspect(context.Background(), InspectOptions{
+		ConfigPath:   testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot),
+		PipelineID:   "reports",
+		Stdout:       &stdout,
+		OutputFormat: OutputFormatJSON,
+	})
+
+	if err != nil {
+		t.Fatalf("Inspect() configured JSON error = %v", err)
+	}
+	result := decodeAppResult(t, stdout.String())
+	if result["pipeline_id"] != "reports" || result["source_backend"] != "local" || result["bundle_count"] != float64(1) {
+		t.Fatalf("result = %#v, want configured inspect metadata", result)
+	}
+}
+
 func TestInspectRequiresPath(t *testing.T) {
 	err := Inspect(context.Background(), InspectOptions{})
 	if err == nil || !strings.Contains(err.Error(), "requires a path") {
--- a/internal/app/run.go
+++ b/internal/app/run.go
@@ -310,6 +310,9 @@ func sshWarnings(pipeline config.Pipeline) []OutputWarning {
 }

 func writeWarnings(w io.Writer, warnings []OutputWarning) error {
+	if w == nil {
+		return nil
+	}
 	for _, warning := range warnings {
 		if _, err := fmt.Fprintf(w, "Warning: %s\n", warning.Message); err != nil {
 			return err
--- a/internal/app/source_select.go
+++ b/internal/app/source_select.go
@@ -0,0 +1,120 @@
+package app
+
+import (
+	"context"
+	"fmt"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+)
+
+type sourceCommandOptions struct {
+	CommandName string
+	Path        string
+	ConfigPath  string
+	PipelineID  string
+	BundlePath  string
+}
+
+type sourceSelection struct {
+	Bundles       []bundle.Bundle
+	PipelineID    string
+	SourceBackend string
+	ConfigMode    bool
+	Warnings      []OutputWarning
+}
+
+func selectSourceBundles(ctx context.Context, options sourceCommandOptions, provider backendFactoryProvider) (sourceSelection, error) {
+	if err := ctx.Err(); err != nil {
+		return sourceSelection{}, err
+	}
+	if options.ConfigPath != "" {
+		cfg, err := config.LoadFile(options.ConfigPath)
+		if err != nil {
+			return sourceSelection{}, err
+		}
+		return selectSourceBundlesFromConfig(ctx, cfg, options, provider)
+	}
+	if options.PipelineID != "" {
+		return sourceSelection{}, fmt.Errorf("configured source mode requires --config")
+	}
+	if options.BundlePath != "" {
+		return sourceSelection{}, fmt.Errorf("configured source mode requires --config")
+	}
+	if options.Path == "" {
+		return sourceSelection{}, fmt.Errorf("%s command requires a path", options.CommandName)
+	}
+	backend, err := newBackendFactory().openLocalPath(ctx, options.Path)
+	if err != nil {
+		return sourceSelection{}, err
+	}
+	defer closeBackend(backend)
+	bundles, err := bundle.Discover(ctx, backend, "")
+	if err != nil {
+		return sourceSelection{}, err
+	}
+	return sourceSelection{Bundles: bundles}, nil
+}
+
+func selectSourceBundlesFromConfig(ctx context.Context, cfg config.Config, 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)
+	if !ok {
+		return sourceSelection{}, fmt.Errorf("pipeline %q not found", options.PipelineID)
+	}
+	backends := provider(secretLoad.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)
+	}
+	defer closeBackend(sourceBackend)
+
+	var bundles []bundle.Bundle
+	if options.BundlePath != "" {
+		sourceBundle, err := bundle.Validate(ctx, sourceBackend, options.BundlePath)
+		if err != nil {
+			return sourceSelection{}, fmt.Errorf("pipeline %s source backend %s bundle %s: %w", pipeline.ID, pipeline.Source.Backend, storage.DisplayPath(options.BundlePath), err)
+		}
+		bundles = []bundle.Bundle{sourceBundle}
+	} else {
+		bundles, err = bundle.Discover(ctx, sourceBackend, "")
+		if err != nil {
+			return sourceSelection{}, fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err)
+		}
+	}
+	return sourceSelection{
+		Bundles:       bundles,
+		PipelineID:    pipeline.ID,
+		SourceBackend: pipeline.Source.Backend,
+		ConfigMode:    true,
+		Warnings:      append(secretConflictWarnings(secretLoad.Conflicts), sourceSSHWarnings(pipeline)...),
+	}, nil
+}
+
+func findPipeline(cfg config.Config, id string) (config.Pipeline, bool) {
+	for _, pipeline := range cfg.Pipelines {
+		if pipeline.ID == id {
+			return pipeline, true
+		}
+	}
+	return config.Pipeline{}, false
+}
+
+func sourceSSHWarnings(pipeline config.Pipeline) []OutputWarning {
+	if pipeline.Source.Backend != config.BackendSSH || pipeline.Source.SSH.HostKeyPolicy != config.HostKeyPolicyOff {
+		return nil
+	}
+	return []OutputWarning{{
+		Message: fmt.Sprintf("pipeline=%s source host_key_policy=off disables SSH host key checking", pipeline.ID),
+	}}
+}
--- a/internal/app/validate.go
+++ b/internal/app/validate.go
@@ -5,43 +5,78 @@ import (
 	"fmt"
 	"io"

-	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
 	"gitea.maximumdirect.net/eric/distributor/internal/storage"
 )

 type ValidateOptions struct {
 	Path         string
+	ConfigPath   string
+	PipelineID   string
+	BundlePath   string
 	Stdout       io.Writer
 	OutputFormat OutputFormat
 }

 func Validate(ctx context.Context, options ValidateOptions) error {
+	return validateWithBackendFactory(ctx, options, newBackendFactoryWithEnvironment)
+}
+
+func validateWithBackendFactory(ctx context.Context, options ValidateOptions, provider backendFactoryProvider) error {
 	if err := ValidateOutputFormat(options.OutputFormat); err != nil {
 		return err
 	}
-	if options.Path == "" {
-		return fmt.Errorf("validate command requires a path")
-	}
-	backend, err := newBackendFactory().openLocalPath(ctx, options.Path)
+	selection, err := selectSourceBundles(ctx, sourceCommandOptions{
+		CommandName: "validate",
+		Path:        options.Path,
+		ConfigPath:  options.ConfigPath,
+		PipelineID:  options.PipelineID,
+		BundlePath:  options.BundlePath,
+	}, provider)
 	if err != nil {
 		return err
 	}
-	bundles, err := bundle.Discover(ctx, backend, "")
+	return writeValidateResult(options, selection)
+}
+
+func validateConfigWithBackendFactory(ctx context.Context, cfg config.Config, options ValidateOptions, provider backendFactoryProvider) error {
+	if err := ValidateOutputFormat(options.OutputFormat); err != nil {
+		return err
+	}
+	selection, err := selectSourceBundlesFromConfig(ctx, cfg, sourceCommandOptions{
+		CommandName: "validate",
+		PipelineID:  options.PipelineID,
+		BundlePath:  options.BundlePath,
+	}, provider)
 	if err != nil {
 		return err
 	}
+	return writeValidateResult(options, selection)
+}
+
+func writeValidateResult(options ValidateOptions, selection sourceSelection) error {
 	if IsJSONOutput(options.OutputFormat) {
-		return WriteJSONEnvelope(options.Stdout, "validate", true, nil, validateResultFromBundles(bundles), nil)
+		return WriteJSONEnvelope(options.Stdout, "validate", true, selection.Warnings, validateResultFromSelection(selection), nil)
 	}
+	var err error
 	if options.Stdout != nil {
-		_, err = fmt.Fprintf(options.Stdout, "Validated %d bundle(s)\n", len(bundles))
+		if err := writeWarnings(options.Stdout, selection.Warnings); err != nil {
+			return err
+		}
+		if selection.ConfigMode {
+			_, err = fmt.Fprintf(options.Stdout, "Validated %d bundle(s) for pipeline %s source %s\n", len(selection.Bundles), selection.PipelineID, selection.SourceBackend)
+		} else {
+			_, err = fmt.Fprintf(options.Stdout, "Validated %d bundle(s)\n", len(selection.Bundles))
+		}
 	}
 	return err
 }

 type validateResult struct {
-	BundleCount int                    `json:"bundle_count"`
-	Bundles     []validateBundleResult `json:"bundles"`
+	PipelineID    string                 `json:"pipeline_id,omitempty"`
+	SourceBackend string                 `json:"source_backend,omitempty"`
+	BundleCount   int                    `json:"bundle_count"`
+	Bundles       []validateBundleResult `json:"bundles"`
 }

 type validateBundleResult struct {
@@ -49,12 +84,14 @@ type validateBundleResult struct {
 	ID   string `json:"id"`
 }

-func validateResultFromBundles(bundles []bundle.Bundle) validateResult {
+func validateResultFromSelection(selection sourceSelection) validateResult {
 	result := validateResult{
-		BundleCount: len(bundles),
-		Bundles:     make([]validateBundleResult, 0, len(bundles)),
+		PipelineID:    selection.PipelineID,
+		SourceBackend: selection.SourceBackend,
+		BundleCount:   len(selection.Bundles),
+		Bundles:       make([]validateBundleResult, 0, len(selection.Bundles)),
 	}
-	for _, sourceBundle := range bundles {
+	for _, sourceBundle := range selection.Bundles {
 		result.Bundles = append(result.Bundles, validateBundleResult{
 			Path: storage.DisplayPath(sourceBundle.RootRelativePath),
 			ID:   sourceBundle.Manifest.ID,
--- a/internal/app/validate_test.go
+++ b/internal/app/validate_test.go
@@ -3,9 +3,15 @@ package app
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"path/filepath"
 	"strings"
 	"testing"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
+	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
 )

 func TestValidateLocalBundle(t *testing.T) {
@@ -31,9 +37,182 @@ func TestValidateExampleSourceBundle(t *testing.T) {
 	}
 }

+func TestValidateConfiguredLocalSource(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{})
+	var stdout bytes.Buffer
+
+	err := Validate(context.Background(), ValidateOptions{
+		ConfigPath: testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot),
+		PipelineID: "reports",
+		Stdout:     &stdout,
+	})
+
+	if err != nil {
+		t.Fatalf("Validate() configured source error = %v", err)
+	}
+	if got, want := stdout.String(), "Validated 1 bundle(s) for pipeline reports source local\n"; got != want {
+		t.Fatalf("stdout = %q, want %q", got, want)
+	}
+}
+
+func TestValidateConfiguredSourceBundlePath(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	testutil.WriteSourceBundle(t, sourceRoot, "daily/one", testutil.BundleOptions{ID: "reports.one"})
+	testutil.WriteSourceBundle(t, sourceRoot, "daily/two", testutil.BundleOptions{ID: "reports.two"})
+	var stdout bytes.Buffer
+
+	err := Validate(context.Background(), ValidateOptions{
+		ConfigPath:   testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot),
+		PipelineID:   "reports",
+		BundlePath:   "daily/two",
+		Stdout:       &stdout,
+		OutputFormat: OutputFormatJSON,
+	})
+
+	if err != nil {
+		t.Fatalf("Validate() configured bundle error = %v", err)
+	}
+	result := decodeAppResult(t, stdout.String())
+	if result["pipeline_id"] != "reports" || result["source_backend"] != "local" || result["bundle_count"] != float64(1) {
+		t.Fatalf("result = %#v, want configured source summary", result)
+	}
+	bundles, ok := result["bundles"].([]any)
+	if !ok || len(bundles) != 1 {
+		t.Fatalf("bundles = %#v, want one bundle", result["bundles"])
+	}
+	sourceBundle, ok := bundles[0].(map[string]any)
+	if !ok || sourceBundle["path"] != "daily/two" || sourceBundle["id"] != "reports.two" {
+		t.Fatalf("bundle = %#v, want narrowed bundle", sourceBundle)
+	}
+}
+
+func TestValidateConfiguredRemoteSourcesThroughStorageAbstraction(t *testing.T) {
+	s3Source := fake.New()
+	testutil.WriteFakeSourceBundle(t, s3Source, "", testutil.BundleOptions{ID: "reports.s3"})
+	sshSource := fake.New()
+	testutil.WriteFakeSourceBundle(t, sshSource, "daily", testutil.BundleOptions{ID: "reports.ssh"})
+	cfg := config.Config{Pipelines: []config.Pipeline{
+		{
+			ID: "s3-reports",
+			Source: config.Backend{
+				Backend:  config.BackendS3,
+				Endpoint: "http://s3.test",
+				Bucket:   "source-bucket",
+			},
+			Destinations: []config.Destination{{
+				ID:      "archive",
+				Backend: config.BackendLocal,
+				Path:    t.TempDir(),
+			}},
+		},
+		{
+			ID: "ssh-reports",
+			Source: config.Backend{
+				Backend: config.BackendSSH,
+				Host:    "ssh.test",
+				Path:    "/source",
+			},
+			Destinations: []config.Destination{{
+				ID:      "archive",
+				Backend: config.BackendLocal,
+				Path:    t.TempDir(),
+			}},
+		},
+	}}
+	config.ApplyDefaults(&cfg)
+	provider := fakeBackendFactoryProvider(t, map[string]storage.Backend{
+		"s3:source-bucket": s3Source,
+		"ssh:/source":      sshSource,
+	})
+
+	var s3Stdout bytes.Buffer
+	if err := validateConfigWithBackendFactory(context.Background(), cfg, ValidateOptions{
+		PipelineID:   "s3-reports",
+		Stdout:       &s3Stdout,
+		OutputFormat: OutputFormatJSON,
+	}, provider); err != nil {
+		t.Fatalf("validate s3 source error = %v", err)
+	}
+	s3Result := decodeAppResult(t, s3Stdout.String())
+	if s3Result["source_backend"] != "s3" || s3Result["bundle_count"] != float64(1) {
+		t.Fatalf("s3 result = %#v, want one s3 bundle", s3Result)
+	}
+
+	var sshStdout bytes.Buffer
+	if err := validateConfigWithBackendFactory(context.Background(), cfg, ValidateOptions{
+		PipelineID: "ssh-reports",
+		BundlePath: "daily",
+		Stdout:     &sshStdout,
+	}, provider); err != nil {
+		t.Fatalf("validate ssh source error = %v", err)
+	}
+	if !strings.Contains(sshStdout.String(), "pipeline ssh-reports source ssh") {
+		t.Fatalf("ssh stdout = %q, want ssh source summary", sshStdout.String())
+	}
+}
+
+func TestValidateConfiguredSourceLoadsSecretsBeforeOpeningBackend(t *testing.T) {
+	sourceRoot := filepath.Join(t.TempDir(), "missing-source")
+	destinationRoot := t.TempDir()
+	configPath := writeConfigFile(t, `
+secrets:
+  directory: `+filepath.Join(t.TempDir(), "missing-secrets")+`
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: `+sourceRoot+`
+    destinations:
+      - id: archive
+        backend: local
+        path: `+destinationRoot+`
+`)
+
+	err := Validate(context.Background(), ValidateOptions{ConfigPath: configPath, PipelineID: "reports"})
+	if err == nil {
+		t.Fatal("Validate() error = nil, want secrets directory error")
+	}
+	if !strings.Contains(err.Error(), "load secrets directory") {
+		t.Fatalf("Validate() error = %v, want secrets directory error", err)
+	}
+	if strings.Contains(err.Error(), "missing-source") {
+		t.Fatalf("Validate() error = %v, opened source before loading secrets", err)
+	}
+}
+
+func TestValidateConfiguredSourceRequiresPipeline(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{})
+
+	err := Validate(context.Background(), ValidateOptions{
+		ConfigPath: testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot),
+	})
+
+	if err == nil || !strings.Contains(err.Error(), "requires --pipeline") {
+		t.Fatalf("Validate() error = %v, want required pipeline", err)
+	}
+}
+
 func TestValidateRequiresPath(t *testing.T) {
 	err := Validate(context.Background(), ValidateOptions{})
 	if err == nil || !strings.Contains(err.Error(), "requires a path") {
 		t.Fatalf("Validate() error = %v, want required path", err)
 	}
 }
+
+func decodeAppResult(t *testing.T, output string) map[string]any {
+	t.Helper()
+	var envelope map[string]any
+	if err := json.Unmarshal([]byte(output), &envelope); err != nil {
+		t.Fatalf("decode output: %v; output = %q", err, output)
+	}
+	result, ok := envelope["result"].(map[string]any)
+	if !ok {
+		t.Fatalf("result = %#v, want object", envelope["result"])
+	}
+	return result
+}
--- a/internal/cli/inspect.go
+++ b/internal/cli/inspect.go
@@ -16,6 +16,9 @@ func inspectCommand(ctx context.Context, args []string, stdout, stderr io.Writer
 	}
 	flags := flag.NewFlagSet("inspect", flag.ContinueOnError)
 	flags.SetOutput(stderr)
+	configPath := flags.String("config", "", "path to config file")
+	pipelineID := flags.String("pipeline", "", "pipeline id")
+	bundlePath := flags.String("bundle", "", "source-root-relative bundle path")
 	formatFlag := addFormatFlag(flags)
 	if err := flags.Parse(args); err != nil {
 		return exitUsage
@@ -28,7 +31,17 @@ func inspectCommand(ctx context.Context, args []string, stdout, stderr io.Writer
 	if !ok {
 		return exitUsage
 	}
-	if err := app.Inspect(ctx, app.InspectOptions{Path: path, Stdout: stdout, OutputFormat: format}); err != nil {
+	if !validateInspectModeOK(stderr, "inspect", path, *configPath, *pipelineID, *bundlePath) {
+		return exitUsage
+	}
+	if err := app.Inspect(ctx, app.InspectOptions{
+		Path:         path,
+		ConfigPath:   *configPath,
+		PipelineID:   *pipelineID,
+		BundlePath:   *bundlePath,
+		Stdout:       stdout,
+		OutputFormat: format,
+	}); err != nil {
 		return fail(stderr, err)
 	}
 	return exitOK
@@ -37,10 +50,15 @@ func inspectCommand(ctx context.Context, args []string, stdout, stderr io.Writer
 func printInspectHelp(w io.Writer) {
 	fmt.Fprint(w, `Usage:
  distributor inspect [--format text|json] <path>
+  distributor inspect --config <path> --pipeline <id> [--bundle <path>] [--format text|json]

 Options:
+  --config <path>     Path to config file for configured source inspection
+  --pipeline <id>     Pipeline id to inspect in config mode
+  --bundle <path>     Source-root-relative bundle path to inspect
  --format text|json  Output format

-Print a normalized summary of local source bundles.
+Print a normalized summary of local source bundles or a configured pipeline
+source.
 `)
 }
--- a/internal/cli/root_test.go
+++ b/internal/cli/root_test.go
@@ -139,6 +139,45 @@ func TestExecuteValidateJSON(t *testing.T) {
 	}
 }

+func TestExecuteValidateConfiguredSource(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{})
+	configPath := testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
+	var stdout, stderr bytes.Buffer
+
+	code := Execute(context.Background(), []string{"validate", "--config", configPath, "--pipeline", "reports"}, &stdout, &stderr)
+
+	if code != exitOK {
+		t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
+	}
+	if got, want := stdout.String(), "Validated 1 bundle(s) for pipeline reports source local\n"; got != want {
+		t.Fatalf("stdout = %q, want %q", got, want)
+	}
+}
+
+func TestExecuteValidateConfiguredSourceJSON(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	testutil.WriteSourceBundle(t, sourceRoot, "daily", testutil.BundleOptions{ID: "reports.daily"})
+	configPath := testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
+	var stdout, stderr bytes.Buffer
+
+	code := Execute(context.Background(), []string{"validate", "--config", configPath, "--pipeline", "reports", "--bundle", "daily", "--format", "json"}, &stdout, &stderr)
+
+	if code != exitOK {
+		t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
+	}
+	envelope := decodeEnvelope(t, &stdout)
+	if envelope["command"] != "validate" || envelope["ok"] != true {
+		t.Fatalf("envelope = %#v, want validate ok", envelope)
+	}
+	result := envelopeResult(t, envelope)
+	if result["pipeline_id"] != "reports" || result["source_backend"] != "local" || result["bundle_count"] != float64(1) {
+		t.Fatalf("result = %#v, want configured source metadata", result)
+	}
+}
+
 func TestExecuteValidateArgs(t *testing.T) {
 	validPath := filepath.Join("..", "bundle", "testdata", "valid_bundle")
 	tests := []struct {
@@ -166,6 +205,24 @@ func TestExecuteValidateArgs(t *testing.T) {
 			wantCode:   exitUsage,
 			wantStderr: "accepts at most one path",
 		},
+		{
+			name:       "path plus config",
+			args:       []string{"validate", "--config", "config.yml", "--pipeline", "reports", validPath},
+			wantCode:   exitUsage,
+			wantStderr: "does not accept a local path",
+		},
+		{
+			name:       "pipeline without config",
+			args:       []string{"validate", "--pipeline", "reports"},
+			wantCode:   exitUsage,
+			wantStderr: "requires --config",
+		},
+		{
+			name:       "config without pipeline",
+			args:       []string{"validate", "--config", "config.yml"},
+			wantCode:   exitUsage,
+			wantStderr: "requires --pipeline",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -223,6 +280,30 @@ func TestExecuteInspectJSON(t *testing.T) {
 	}
 }

+func TestExecuteInspectConfiguredSource(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	testutil.WriteSourceBundle(t, sourceRoot, "daily", testutil.BundleOptions{ID: "reports.daily"})
+	configPath := testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
+	var stdout, stderr bytes.Buffer
+
+	code := Execute(context.Background(), []string{"inspect", "--config", configPath, "--pipeline", "reports"}, &stdout, &stderr)
+
+	if code != exitOK {
+		t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
+	}
+	for _, want := range []string{
+		"Pipeline: reports",
+		"Source: local",
+		"path=daily",
+		"id=reports.daily",
+	} {
+		if !strings.Contains(stdout.String(), want) {
+			t.Fatalf("stdout = %q, want substring %q", stdout.String(), want)
+		}
+	}
+}
+
 func TestExecuteInspectArgs(t *testing.T) {
 	validPath := filepath.Join("..", "bundle", "testdata", "valid_bundle")
 	tests := []struct {
@@ -250,6 +331,24 @@ func TestExecuteInspectArgs(t *testing.T) {
 			wantCode:   exitUsage,
 			wantStderr: "accepts at most one path",
 		},
+		{
+			name:       "path plus config",
+			args:       []string{"inspect", "--config", "config.yml", "--pipeline", "reports", validPath},
+			wantCode:   exitUsage,
+			wantStderr: "does not accept a local path",
+		},
+		{
+			name:       "pipeline without config",
+			args:       []string{"inspect", "--pipeline", "reports"},
+			wantCode:   exitUsage,
+			wantStderr: "requires --config",
+		},
+		{
+			name:       "config without pipeline",
+			args:       []string{"inspect", "--config", "config.yml"},
+			wantCode:   exitUsage,
+			wantStderr: "requires --pipeline",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
--- a/internal/cli/source_mode.go
+++ b/internal/cli/source_mode.go
@@ -0,0 +1,26 @@
+package cli
+
+import (
+	"fmt"
+	"io"
+)
+
+func validateInspectModeOK(stderr io.Writer, command, path, configPath, pipelineID, bundlePath string) bool {
+	configMode := configPath != "" || pipelineID != "" || bundlePath != ""
+	if !configMode {
+		return true
+	}
+	if path != "" {
+		fmt.Fprintf(stderr, "distributor: %s does not accept a local path with --config, --pipeline, or --bundle\n", command)
+		return false
+	}
+	if configPath == "" {
+		fmt.Fprintf(stderr, "distributor: %s requires --config when --pipeline or --bundle is set\n", command)
+		return false
+	}
+	if pipelineID == "" {
+		fmt.Fprintf(stderr, "distributor: %s requires --pipeline in config mode\n", command)
+		return false
+	}
+	return true
+}
--- a/internal/cli/validate.go
+++ b/internal/cli/validate.go
@@ -16,6 +16,9 @@ func validateCommand(ctx context.Context, args []string, stdout, stderr io.Write
 	}
 	flags := flag.NewFlagSet("validate", flag.ContinueOnError)
 	flags.SetOutput(stderr)
+	configPath := flags.String("config", "", "path to config file")
+	pipelineID := flags.String("pipeline", "", "pipeline id")
+	bundlePath := flags.String("bundle", "", "source-root-relative bundle path")
 	formatFlag := addFormatFlag(flags)
 	if err := flags.Parse(args); err != nil {
 		return exitUsage
@@ -28,7 +31,17 @@ func validateCommand(ctx context.Context, args []string, stdout, stderr io.Write
 	if !ok {
 		return exitUsage
 	}
-	if err := app.Validate(ctx, app.ValidateOptions{Path: path, Stdout: stdout, OutputFormat: format}); err != nil {
+	if !validateInspectModeOK(stderr, "validate", path, *configPath, *pipelineID, *bundlePath) {
+		return exitUsage
+	}
+	if err := app.Validate(ctx, app.ValidateOptions{
+		Path:         path,
+		ConfigPath:   *configPath,
+		PipelineID:   *pipelineID,
+		BundlePath:   *bundlePath,
+		Stdout:       stdout,
+		OutputFormat: format,
+	}); err != nil {
 		return fail(stderr, err)
 	}
 	return exitOK
@@ -37,10 +50,15 @@ func validateCommand(ctx context.Context, args []string, stdout, stderr io.Write
 func printValidateHelp(w io.Writer) {
 	fmt.Fprint(w, `Usage:
  distributor validate [--format text|json] <path>
+  distributor validate --config <path> --pipeline <id> [--bundle <path>] [--format text|json]

 Options:
+  --config <path>     Path to config file for configured source validation
+  --pipeline <id>     Pipeline id to validate in config mode
+  --bundle <path>     Source-root-relative bundle path to validate
  --format text|json  Output format

-Validate a local source bundle directory or a tree containing source bundles.
+Validate a local source bundle directory, a local source bundle tree, or a
+configured pipeline source.
 `)
 }