Add local source pipeline execution

docs/internal/app.md

@@ -23,6 +23,13 @@ config path, pipeline ID, dry-run flag, force flag, and optional notifier. It
 loads the same config as `Run`, narrows execution to exactly one configured
 pipeline, and returns a `RunReport` without writing command output.
 
+`RunPipelineWithLocalSource` is the app-layer single-pipeline entrypoint for an
+already prepared local source bundle root. It accepts the same pipeline
+selection and execution options as `RunPipeline` plus a local source root path.
+It loads config, selects one configured pipeline, opens the supplied source
+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.
@@ -64,6 +71,14 @@ pipeline. It uses the same backend factory, secret loading, transform registry,
 warning generation, destination planning, publish execution, notification
 behavior, and failure aggregation as `Run`.
 
+`RunPipelineWithLocalSource` follows the same flow after pipeline selection
+except for source opening and source discovery. It opens the supplied local
+source root directly, validates the root bundle before opening any destinations,
+and passes the resulting local source backend and bundle into the same
+destination planning and execution loop. Destination code receives the normal
+storage backend and bundle values and does not depend on how the source root was
+prepared.
+
 ## Coordination
 
 `PipelineRunCoordinator` wraps `RunPipeline` with in-memory admission control.
@@ -89,6 +104,11 @@ work starts, source open errors, and source discovery errors.
 `RunPipeline` returns `PipelineNotFoundError` when the requested pipeline ID is
 not configured. Callers can detect that condition with `IsPipelineNotFound`.
 
+`RunPipelineWithLocalSource` also returns `PipelineNotFoundError` for an unknown
+pipeline ID. It returns before destination opening when the supplied local
+source root is missing, cannot be opened, or does not validate as one complete
+source bundle.
+
 Per-destination backend, planning, execution, and notification errors are
 aggregated into one run error after remaining destinations have been attempted.
 Destination diagnostics include pipeline ID, destination ID, backend, and

internal/app/run.go

@@ -29,6 +29,15 @@ type RunPipelineOptions struct {
 	Notifier   notify.Notifier
 }
 
+type RunPipelineWithLocalSourceOptions struct {
+	ConfigPath string
+	PipelineID string
+	SourceRoot string
+	DryRun     bool
+	Force      bool
+	Notifier   notify.Notifier
+}
+
 func Run(ctx context.Context, options RunOptions) error {
 	if err := ValidateOutputFormat(options.OutputFormat); err != nil {
 		return err
@@ -64,6 +73,25 @@ func RunPipeline(ctx context.Context, options RunPipelineOptions) (RunReport, er
 	return runPipelineConfig(ctx, cfg, options)
 }
 
+func RunPipelineWithLocalSource(ctx context.Context, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+	if err := ctx.Err(); err != nil {
+		return RunReport{}, err
+	}
+	if options.SourceRoot == "" {
+		return RunReport{}, fmt.Errorf("source root is required")
+	}
+
+	configPath := options.ConfigPath
+	if configPath == "" {
+		configPath = config.DefaultConfigPath
+	}
+	cfg, err := config.LoadFile(configPath)
+	if err != nil {
+		return RunReport{}, err
+	}
+	return runPipelineConfigWithLocalSource(ctx, cfg, options)
+}
+
 func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error {
 	return runConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment)
 }
@@ -74,12 +102,17 @@ func runPipelineConfig(ctx context.Context, cfg config.Config, options RunPipeli
 	return runPipelineConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment)
 }
 
+func runPipelineConfigWithLocalSource(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+	return runPipelineConfigWithLocalSourceAndBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment)
+}
+
 func runPipelineConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunPipelineOptions, provider backendFactoryProvider) (RunReport, error) {
 	pipeline, ok := findPipeline(cfg, 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{
@@ -89,6 +122,25 @@ func runPipelineConfigWithBackendFactory(ctx context.Context, cfg config.Config,
 	}, provider)
 }
 
+func runPipelineConfigWithLocalSourceAndBackendFactory(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions, provider backendFactoryProvider) (RunReport, error) {
+	pipeline, ok := findPipeline(cfg, 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{
+		DryRun:   options.DryRun,
+		Force:    options.Force,
+		Notifier: options.Notifier,
+	}, provider, &localSourceRoot{
+		pipelineID: options.PipelineID,
+		root:       options.SourceRoot,
+	})
+}
+
 func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) error {
 	report, err := buildRunReportWithBackendFactory(ctx, cfg, options, provider)
 	if err != nil && !IsPartialResultError(err) {
@@ -101,6 +153,15 @@ 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)
+}
+
+type localSourceRoot struct {
+	pipelineID string
+	root       string
+}
+
+func buildRunReport(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider, sourceRoot *localSourceRoot) (RunReport, error) {
 	notifier := options.Notifier
 	if notifier == nil {
 		notifier = notify.Noop{}
@@ -125,18 +186,13 @@ func buildRunReportWithBackendFactory(ctx context.Context, cfg config.Config, op
 	for _, pipeline := range cfg.Pipelines {
 		pipelineWarnings := sshWarnings(pipeline)
 		report.addWarnings(pipelineWarnings)
-		sourceBackend, err := backends.openSource(ctx, pipeline.Source)
+		sourceBackend, bundles, sourceBackendName, err := openPipelineSource(ctx, backends, pipeline, sourceRoot)
 		if err != nil {
-			return report, fmt.Errorf("pipeline %s source backend %s: %w", pipeline.ID, pipeline.Source.Backend, err)
+			return report, err
-		}
-		bundles, err := bundle.Discover(ctx, sourceBackend, "")
-		if err != nil {
-			closeBackend(sourceBackend)
-			return report, fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err)
 		}
 		report.Pipelines = append(report.Pipelines, RunPipelineSummary{
 			ID:            pipeline.ID,
-			SourceBackend: pipeline.Source.Backend,
+			SourceBackend: sourceBackendName,
 			BundleCount:   len(bundles),
 			Destinations:  destinationIDs(pipeline.Destinations),
 			Warnings:      pipelineWarnings,
@@ -251,6 +307,32 @@ func buildRunReportWithBackendFactory(ctx context.Context, cfg config.Config, op
 	return report, nil
 }
 
+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)
+		if err != nil {
+			return nil, nil, config.BackendLocal, fmt.Errorf("pipeline %s source backend %s: %w", pipeline.ID, config.BackendLocal, err)
+		}
+		sourceBundle, err := bundle.Validate(ctx, sourceBackend, "")
+		if err != nil {
+			closeBackend(sourceBackend)
+			return nil, nil, config.BackendLocal, fmt.Errorf("pipeline %s source backend %s validate source bundle: %w", pipeline.ID, config.BackendLocal, err)
+		}
+		return sourceBackend, []bundle.Bundle{sourceBundle}, config.BackendLocal, nil
+	}
+
+	sourceBackend, err := backends.openSource(ctx, pipeline.Source)
+	if err != nil {
+		return nil, nil, pipeline.Source.Backend, fmt.Errorf("pipeline %s source backend %s: %w", pipeline.ID, pipeline.Source.Backend, err)
+	}
+	bundles, err := bundle.Discover(ctx, sourceBackend, "")
+	if err != nil {
+		closeBackend(sourceBackend)
+		return nil, nil, pipeline.Source.Backend, fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err)
+	}
+	return sourceBackend, bundles, pipeline.Source.Backend, nil
+}
+
 type closeableBackend interface {
 	Close() error
 }

internal/app/run_test.go

@@ -3,6 +3,7 @@ package app
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -220,6 +221,116 @@ func TestRunPublishesNewLocalBundle(t *testing.T) {
 	}
 }
 
+func TestRunPipelineWithLocalSourcePublishesConfiguredDestination(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+
+	report, err := RunPipelineWithLocalSource(context.Background(), RunPipelineWithLocalSourceOptions{
+		ConfigPath: writeUploadPipelineConfig(t, destinationRoot),
+		PipelineID: "reports",
+		SourceRoot: sourceRoot,
+	})
+	if err != nil {
+		t.Fatalf("RunPipelineWithLocalSource() error = %v", err)
+	}
+
+	if got, want := report.Summary.Status, "ok"; got != want {
+		t.Fatalf("report status = %q, want %q", got, want)
+	}
+	if got, want := len(report.Pipelines), 1; got != want {
+		t.Fatalf("pipeline count = %d, want %d", got, want)
+	}
+	if got, want := report.Pipelines[0].SourceBackend, config.BackendLocal; got != want {
+		t.Fatalf("source backend = %q, want %q", got, want)
+	}
+	if got, want := report.Pipelines[0].BundleCount, 1; got != want {
+		t.Fatalf("bundle count = %d, want %d", got, want)
+	}
+	if got, want := len(report.Actions), 1; got != want {
+		t.Fatalf("action count = %d, want %d", got, want)
+	}
+	if report.Actions[0].PipelineID != "reports" || report.Actions[0].DestinationID != "archive" {
+		t.Fatalf("action = %#v, want reports/archive action", report.Actions[0])
+	}
+	testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
+	testutil.AssertFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n")
+}
+
+func TestRunPipelineWithLocalSourceValidatesBeforeDestinationWrites(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	writeJSONManifest(t, sourceRoot, testutil.ValidManifest(testutil.BundleOptions{}))
+
+	_, err := RunPipelineWithLocalSource(context.Background(), RunPipelineWithLocalSourceOptions{
+		ConfigPath: writeUploadPipelineConfig(t, destinationRoot),
+		PipelineID: "reports",
+		SourceRoot: sourceRoot,
+	})
+	if err == nil {
+		t.Fatal("RunPipelineWithLocalSource() error = nil, want validation error")
+	}
+	if !strings.Contains(err.Error(), "validate source bundle") {
+		t.Fatalf("RunPipelineWithLocalSource() error = %v, want source validation context", err)
+	}
+	entries, readErr := os.ReadDir(destinationRoot)
+	if readErr != nil {
+		t.Fatalf("ReadDir() error = %v", readErr)
+	}
+	if len(entries) != 0 {
+		t.Fatalf("destination entries = %d, want no writes", len(entries))
+	}
+}
+
+func TestRunPipelineWithLocalSourcePublishesToRegisteredDestinationBackends(t *testing.T) {
+	sourceRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+	s3Destination := fake.New()
+	sshDestination := fake.New()
+	cfg := config.Config{
+		Pipelines: []config.Pipeline{{
+			ID: "reports",
+			Source: config.Backend{
+				Backend: config.BackendHTTPUpload,
+				Upload:  config.HTTPUpload{TokenEnv: "UPLOAD_TOKEN"},
+			},
+			Destinations: []config.Destination{
+				{
+					ID:       "object-archive",
+					Backend:  config.BackendS3,
+					Endpoint: "http://s3.test",
+					Bucket:   "destination-bucket",
+				},
+				{
+					ID:      "ssh-archive",
+					Backend: config.BackendSSH,
+					Host:    "ssh.test",
+					Path:    "/destination",
+				},
+			},
+		}},
+	}
+	config.ApplyDefaults(&cfg)
+	provider := fakeBackendFactoryProvider(t, map[string]storage.Backend{
+		"s3:destination-bucket": s3Destination,
+		"ssh:/destination":      sshDestination,
+	})
+
+	report, err := runPipelineConfigWithLocalSourceAndBackendFactory(context.Background(), cfg, RunPipelineWithLocalSourceOptions{
+		PipelineID: "reports",
+		SourceRoot: sourceRoot,
+	}, provider)
+	if err != nil {
+		t.Fatalf("runPipelineConfigWithLocalSourceAndBackendFactory() error = %v", err)
+	}
+
+	if got, want := len(report.Actions), 2; got != want {
+		t.Fatalf("action count = %d, want %d", got, want)
+	}
+	testutil.AssertFakeFile(t, s3Destination, "report.md", "# Report\nSunny.\n")
+	testutil.AssertFakeFile(t, sshDestination, "summary.txt", "Summary\n")
+}
+
 func TestRunExplicitPreserveRelativePathMappingMatchesDefault(t *testing.T) {
 	sourceRoot := t.TempDir()
 	destinationRoot := t.TempDir()
@@ -1516,6 +1627,21 @@ func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestina
 	return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination)
 }
 
+func writeUploadPipelineConfig(t *testing.T, destinationRoot string) string {
+	t.Helper()
+	return writeConfigFile(t, `
+pipelines:
+  - id: reports
+    source:
+      backend: http_upload
+      token_env: UPLOAD_TOKEN
+    destinations:
+      - id: archive
+        backend: local
+        path: `+destinationRoot+`
+`)
+}
+
 func writeTwoPipelineConfig(t *testing.T, firstSource, firstDestination, secondSource, secondDestination string) string {
 	t.Helper()
 	return writeConfigFile(t, `
@@ -1553,6 +1679,20 @@ func writeDestinationState(t *testing.T, root, relative string, manifest bundle.
 	testutil.WriteDestinationState(t, root, relative, manifest, testutil.DestinationStateOptions{})
 }
 
+func writeJSONManifest(t *testing.T, root string, manifest bundle.Manifest) {
+	t.Helper()
+	if err := os.MkdirAll(root, 0o755); err != nil {
+		t.Fatalf("mkdir manifest root: %v", err)
+	}
+	data, err := json.MarshalIndent(manifest, "", "  ")
+	if err != nil {
+		t.Fatalf("marshal manifest: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(root, bundle.ManifestName), append(data, '\n'), 0o600); err != nil {
+		t.Fatalf("write manifest: %v", err)
+	}
+}
+
 func readStateFile(t *testing.T, path string) state.DistributorState {
 	t.Helper()
 	return testutil.ReadDestinationState(t, path)