Refresh app and HTTP boundary documentation

docs/internal/app.md

@@ -2,66 +2,141 @@
 
 ## Purpose
 
-`internal/app` owns top-level use cases for `run`, `validate`, and `inspect`. It wires configuration, storage backends, transforms, publish planning, execution, summaries, and notification handoff.
+`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.
 
-## Inputs and outputs
+The package is the boundary between callers and lower-level domain packages. It
+does not own manifest validation rules, destination state comparison, storage
+path rules, output planning, transform rendering, or backend-specific behavior.
 
-`Run` accepts a context, optional config path, dry-run flag, force flag, stdout writer, output format, and optional notifier. It loads YAML config, discovers source bundles for each configured pipeline, plans each destination independently, optionally executes publish plans, writes text or JSON output when stdout is supplied, and returns an aggregated error if any destination fails.
+## Use Cases
 
-`Validate` and `Inspect` accept either a local path or one configured pipeline source. `Validate` discovers and validates bundles. `Inspect` writes bundle metadata and manifest file entries to stdout when provided.
+`Run` is the CLI-facing all-pipeline entrypoint. It accepts a context, optional
+config path, dry-run flag, force flag, stdout writer, output format, and
+optional notifier. It runs every configured pipeline, builds a `RunReport`, and
+projects the report to text or JSON when stdout is supplied.
 
-## Run flow
+`RunPipeline` is the app-layer single-pipeline entrypoint. It accepts a context,
+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.
 
-The runner:
+`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.
+
+## Run Reports
+
+`RunReport` is the structured result model for run workflows. It includes
+dry-run state, pipeline summaries, action records, output metadata, summary
+counters, warnings, and destination-scoped output errors.
+
+Text and JSON run output are projections of `RunReport`. JSON tags on report
+records match the CLI JSON output contract. Text output preserves the CLI
+summary shape while keeping output rendering outside the core planning and
+execution loop.
+
+Destination-scoped failures produce a report plus an aggregated error. Fatal
+setup failures, such as config loading, source open, or source discovery
+failures, return before a complete run report is available.
+
+## Run Flow
+
+The app runner:
 
 1. loads config from the supplied path or `config.DefaultConfigPath`;
-2. opens the configured source backend;
+2. loads configured secret files into a config-owned environment resolver;
-3. discovers validated bundles from the source root;
+3. builds the app-level backend factory and transform registry;
-4. selects source bundles for each destination according to destination path mapping;
+4. opens each selected pipeline source backend;
-5. opens each destination backend independently;
+5. discovers validated source bundles from the source root;
-6. builds publish plans for the selected bundle and destination combinations;
+6. selects source bundles for each destination according to path mapping;
-7. prints plan lines or JSON action records and records summary counters;
+7. opens destination backends independently;
-8. executes publish or replacement plans unless dry-run is enabled;
+8. builds publish plans for selected bundle and destination combinations;
-9. invokes the notifier after successful publish or replacement actions.
+9. records warnings, action records, output metadata, and summary counters;
+10. executes publish or replacement plans unless dry-run is enabled;
+11. invokes the notifier after successful publish or replacement actions;
+12. returns the structured report and any aggregated destination failures.
 
-Destination failures are collected while later destinations continue to run. Source open and source discovery failures stop the run because there are no valid bundles to fan out.
+`RunPipeline` follows the same flow after selecting a single configured
+pipeline. It uses the same backend factory, secret loading, transform registry,
+warning generation, destination planning, publish execution, notification
+behavior, and failure aggregation as `Run`.
 
-## Run implementation
+## Coordination
 
-`run.go` contains the public `Run` entrypoint and the main configuration orchestration path. Package-local run helpers are grouped by responsibility:
+`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.
 
-- `run_selection.go`: destination bundle selection, path mapping decisions, and fixed-path warnings;
+Coordinator records contain a run ID, pipeline ID, status, timestamps, completed
-- `run_warnings.go`: secret and SSH warning data;
+report, and error text when applicable. Active state is memory-only and is
-- `run_output.go`: text plan lines, JSON action records, and output projections;
+cleared after success, failure, unknown pipeline ID, or context cancellation.
-- `run_summary.go`: summary counters and JSON summary records;
+
-- `run_failures.go`: destination failure aggregation and partial-result detection;
+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.
+
+## Errors
+
+`Run` returns immediately for config loading errors, context cancellation before
+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`.
+
+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
+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.
+
+## Package Layout
+
+Run helpers are grouped by responsibility:
+
+- `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.
+- `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.
+- `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`.
 
-These helpers remain in `internal/app` because command output, warning collection, destination failure aggregation, notifier handoff, and backend construction are app-owned orchestration concerns.
+## Backend And Transform Wiring
 
-## Backend and transform wiring
+The app-level backend factory registers local, SSH, and S3 backends for runtime
+execution. Source and destination backend config is converted through a shared
+app-local open spec before adapter construction.
 
-The app-level backend factory registers local, SSH, and S3 backends for execution. Source and destination backend config is converted through a shared app-local open spec before adapter construction. S3 explicit credential references are resolved through the config environment resolver.
+Credential references are resolved through the config environment resolver.
+Production app code must not read backend credential environment variables
+directly.
 
-The app-level transform registry registers Markdown-to-HTML using `internal/transform/markdown`. Lower-level publish code receives a resolver and does not import concrete transform implementations.
+The app-level transform registry registers Markdown-to-HTML through
+`internal/transform/markdown`. Lower-level publish code receives a resolver and
+does not import concrete transform implementations.
 
-## Dry-run behavior
+## Dry-Run Behavior
 
-Dry-run still loads config, opens backends, discovers bundles, inspects destinations, resolves transforms, and builds publish plans. It does not write destination outputs, write `.distributor.json`, delete managed outputs, perform forced prefix deletion, or notify.
+Dry-run loads config, opens backends, discovers bundles, inspects destinations,
-
+resolves transforms, and builds publish plans. It does not write destination
-## Failure behavior
+outputs, write `.distributor.json`, delete managed outputs, perform forced
-
+prefix deletion, or invoke notifications.
-`Run` returns immediately for config loading errors, context cancellation before work starts, source open errors, and source discovery errors. Per-destination backend, planning, execution, and notification errors are aggregated into one run error after remaining destinations have been attempted.
-
-Run diagnostics include pipeline id, destination id, destination backend, and bundle path for destination-scoped failures. Source open and discovery failures include the source backend.
-
-Stdout write errors are returned immediately because the caller's requested output stream can no longer be trusted.
-
-## Boundaries
-
-`internal/app` coordinates packages but does not own manifest validation rules, destination state comparison, storage path rules, output planning, transform rendering, or backend-specific filesystem behavior.
-
-Configured-source `Validate` and `Inspect` share source backend construction with `Run` and do not open destinations.
 
 ## Tests
 
@@ -71,10 +146,18 @@ Before changing app orchestration, inspect tests under:
 - `internal/cli`
 - `internal/publish`
 
+Use focused app tests for report structure, single-pipeline execution,
+coordinator admission, warning generation, notification behavior, and
+partial-result aggregation.
+
 ## Invariants
 
 - One source fans out to each destination independently.
 - Destination failures do not prevent later destinations from being planned.
+- 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/roadmap/implementation.md

@@ -1,102 +1,91 @@
-# Roadmap
+# HTTP API Boundary Roadmap
 
-This directory contains only future, deferred, or aspirational work for
+## Purpose
-`distributor`. Implemented behavior is documented in the current user,
-operator, internal, policy, integration, and example documentation:
 
-- `README.md`
+This roadmap records the accepted boundary for a future HTTP API. The current
-- `docs/cli.md`
+application exposes CLI commands and internal app-layer run contracts; it does
-- `docs/config.md`
+not implement an HTTP server, HTTP routes, a `serve` command, app-level
-- `docs/operations.md`
+authentication, or in-app TLS.
-- `docs/troubleshooting.md`
-- `docs/internal/`
-- `docs/integrations/markdown.md`
-- `docs/policy/`
-- `examples/`
 
-`distributor` currently supports local, SSH/SFTP, and S3-compatible source and
+Implemented internal run contracts are documented in `docs/internal/app.md`.
-destination backends; producer bundle creation through `pkg/bundle` and
+This file is the canonical home for future HTTP boundary decisions until an
-`distributor manifest create`; configured source validation and inspection;
+HTTP implementation roadmap replaces it.
-Markdown sidecar and `index.html` publication; archive and fixed destination
-path mapping; destination link metadata; shared text/JSON CLI output; and
-managed destination replacement behavior.
 
-## Future Work
+## Accepted Direction
 
-These items are not implemented. They should not be documented as current
+The future HTTP API should trigger configured distributor pipelines through the
-behavior outside `docs/roadmap/` unless a future implementation adds them.
+existing app-layer single-pipeline run path and in-memory coordinator.
 
-### CLI And Status Output
+The HTTP API is intentionally narrow:
 
-- Add a root-global output flag only if the command parser is later refactored
+- A trigger request accepts only a pipeline ID as application input.
-  around shared root options.
+- A trigger request starts work asynchronously and returns a run ID after
-- Add output formats beyond `text` and `json` only if a concrete consumer
+  admission.
-  requires them.
+- Run status is read through a separate status endpoint keyed by run ID.
-- Add a versioned JSON schema reference after the first JSON-capable release.
+- Status records expose run ID, pipeline ID, current status, timestamps, and
-- Add destination-state inspection behind an explicit flag such as
+  completed report or error details when available.
-  `--with-destinations` if operators need fan-out status diagnostics from
-  `inspect`.
-- Add additional status or inspection presentation for destination primary
-  links beyond the current `run --format json` result model.
 
-### Producer Workflows
+The application remains a bundle distribution tool. The HTTP API must not turn
+`distributor` into a workflow engine, CMS, report generator, or public web
+authoring service.
 
-- Add a no-write manifest creation mode, such as writing manifest JSON to
+## Error Mapping
-  stdout, if producer pipelines need to capture manifests directly.
-- Add broader producer workflow helpers, such as richer ignore rules or
-  template scaffolding, if real producer use cases require them.
-- Add remote or storage-backed producer writers only if producer applications
-  need to assemble bundles outside the local filesystem.
 
-### Publication And Transform Behavior
+Future transport code should map app-layer errors without changing app-layer
+error ownership:
 
-- Add a separate collection or site-index transform if distributor needs
+- Unknown pipeline IDs map to `404 Not Found`.
-  multi-page aggregation.
+- Duplicate in-flight runs for the same pipeline ID map to `409 Conflict`.
-- Add richer transform metadata only if future state consumers need more than
+- Validation, config, source, destination, publish, transform, and notification
-  the transform name and output path.
+  errors map to transport errors according to their app-layer context.
-- Add custom HTML index output names only if fixed `index.html` is too limiting
-  for real deployments.
-- Add richer fixed-destination source selection policies if deployments need
-  something other than newest-by-`created`.
-- Add stricter handling for equal latest timestamps if timestamp ties become
-  common in producer workflows.
-- Add higher-level status or approval workflows for fixed-root replacements if
-  dry-run output is not enough operational protection.
-- Add richer link policies only if `auto`, `html`, and `source` prove
-  insufficient.
 
-### State And Compatibility
+Duplicate runs must not be queued. Run state remains in memory unless a later
+roadmap explicitly adds durable run storage.
 
-- Define a post-release destination state schema bump policy before introducing
+## Context And Lifetime
-  materially incompatible state changes.
-- Add warning-only digest mismatch handling only if an operator workflow needs
-  publication to continue after validation failures.
-- Add compatibility parsing for legacy SSH URI config only if migration support
-  is required.
 
-### Backends, Security, And Deployment
+The request context guards admission. Once a run is admitted, execution is tied
+to the server or coordinator lifetime context rather than to the client request
+lifetime.
 
-- Add authentication mechanisms beyond the implemented SSH agent/key and S3
+This split allows a client disconnect or request timeout to stop waiting for
-  credential paths only when a concrete backend workflow requires them.
+admission without canceling a run that has already been accepted.
-- Add broad recursive destination deletion outside managed bundle paths only if
-  a future design can preserve the current safety boundary.
-- Add concurrent fan-out publishing only if runtime profiling shows it is
-  needed.
-- Add streaming, resumable, or multipart S3 uploads only if object sizes make
-  the current write path insufficient.
-- Add cloud-provider-specific IAM integration docs only when the repository
-  includes tested provider-specific behavior.
-- Add repository-managed packaging, release, and deployment automation when the
-  release process is ready to be standardized.
 
-## Roadmap Maintenance
+## Security Boundary
 
-When adding future roadmap work:
+The first HTTP server should default to private binding, such as `127.0.0.1`.
+Operators should expose it through a reverse proxy, private network, or external
+mTLS when transport security or remote access is required.
 
-- describe user-visible behavior and safety boundaries;
+The first HTTP implementation should not include:
-- define which current docs must change after implementation;
+
-- keep examples secret-free and runnable or clearly environment-gated;
+- bearer-token authentication;
-- keep workflow labels out of production code, tests, config fields, and
+- in-app TLS configuration;
-  user-facing documentation;
+- public-network exposure defaults.
-- run focused tests for the changed behavior and `go test ./...` for
+
-  cross-package changes.
+A later roadmap must explicitly change this security decision before any of
+those features are added.
+
+## Non-Goals
+
+The future HTTP API should not add:
+
+- public CLI flags for selecting one pipeline during `distributor run`;
+- a CLI framework;
+- a generic workflow engine;
+- plugin execution;
+- durable run storage;
+- app-level authentication;
+- in-app TLS.
+
+## Verification Expectations
+
+Any later HTTP implementation should preserve existing CLI behavior and keep
+the app-layer run contracts tested. At minimum, it should cover:
+
+- trigger requests with known and unknown pipeline IDs;
+- duplicate in-flight trigger requests;
+- asynchronous acceptance and status lookup;
+- private bind defaults;
+- request-context admission behavior;
+- coordinator-lifetime run execution.

internal/app/run.go

@@ -21,6 +21,14 @@ type RunOptions struct {
 	Notifier     notify.Notifier
 }
 
+type RunPipelineOptions struct {
+	ConfigPath string
+	PipelineID 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
@@ -40,90 +48,108 @@ func Run(ctx context.Context, options RunOptions) error {
 	return runConfig(ctx, cfg, options)
 }
 
+func RunPipeline(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
+	if err := ctx.Err(); err != nil {
+		return RunReport{}, err
+	}
+
+	configPath := options.ConfigPath
+	if configPath == "" {
+		configPath = config.DefaultConfigPath
+	}
+	cfg, err := config.LoadFile(configPath)
+	if err != nil {
+		return RunReport{}, err
+	}
+	return runPipelineConfig(ctx, cfg, options)
+}
+
 func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error {
 	return runConfigWithBackendFactory(ctx, cfg, 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)
+}
+
+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{
+		Secrets:   cfg.Secrets,
+		Pipelines: []config.Pipeline{pipeline},
+	}, RunOptions{
+		DryRun:   options.DryRun,
+		Force:    options.Force,
+		Notifier: options.Notifier,
+	}, provider)
+}
+
 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) {
+		return err
+	}
+	if outputErr := WriteRunReport(options.Stdout, options.OutputFormat, report); outputErr != nil {
+		return outputErr
+	}
+	return err
+}
+
+func buildRunReportWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) (RunReport, error) {
 	notifier := options.Notifier
 	if notifier == nil {
 		notifier = notify.Noop{}
 	}
-	jsonOutput := IsJSONOutput(options.OutputFormat)
 	summary := runSummary{dryRun: options.DryRun}
-	result := runResult{
+	report := RunReport{
 		DryRun:    options.DryRun,
-		Pipelines: []runPipelineResult{},
+		Pipelines: []RunPipelineSummary{},
-		Actions:   []runActionResult{},
+		Actions:   []RunActionRecord{},
 	}
-	var warnings []OutputWarning
 	var failures runFailures
 	secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil)
 	if err != nil {
-		return err
+		return report, err
 	}
 	secretWarnings := secretConflictWarnings(secretLoad.Conflicts)
-	if jsonOutput {
+	report.PreambleWarnings = append(report.PreambleWarnings, secretWarnings...)
-		warnings = append(warnings, secretWarnings...)
+	report.addWarnings(secretWarnings)
-	} else if options.Stdout != nil {
-		if err := writeWarnings(options.Stdout, secretWarnings); err != nil {
-			return err
-		}
-	}
 	backends := provider(secretLoad.Environment)
 	backends.readOnlyKnownHosts = options.DryRun
 	transforms := newTransformRegistry()
-	if options.Stdout != nil && !jsonOutput {
-		if _, err := fmt.Fprintf(options.Stdout, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil {
-			return err
-		}
-	}
 	for _, pipeline := range cfg.Pipelines {
 		pipelineWarnings := sshWarnings(pipeline)
-		if jsonOutput {
+		report.addWarnings(pipelineWarnings)
-			warnings = append(warnings, pipelineWarnings...)
-		} else if options.Stdout != nil {
-			if err := writeWarnings(options.Stdout, pipelineWarnings); err != nil {
-				return err
-			}
-		}
 		sourceBackend, err := backends.openSource(ctx, pipeline.Source)
 		if err != nil {
-			return fmt.Errorf("pipeline %s source backend %s: %w", pipeline.ID, pipeline.Source.Backend, err)
+			return report, 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 fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err)
+			return report, fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err)
 		}
-		result.Pipelines = append(result.Pipelines, runPipelineResult{
+		report.Pipelines = append(report.Pipelines, RunPipelineSummary{
 			ID:            pipeline.ID,
 			SourceBackend: pipeline.Source.Backend,
 			BundleCount:   len(bundles),
 			Destinations:  destinationIDs(pipeline.Destinations),
+			Warnings:      pipelineWarnings,
 		})
-		if options.Stdout != nil && !jsonOutput {
+		pipelineIndex := len(report.Pipelines) - 1
-			if _, err := fmt.Fprintf(options.Stdout, "- pipeline=%s source=%s bundles=%d destinations=%s\n", pipeline.ID, pipeline.Source.Backend, len(bundles), destinationSummary(pipeline.Destinations)); err != nil {
-				closeBackend(sourceBackend)
-				return err
-			}
-		}
 		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))
-					if jsonOutput {
+					report.addWarning(warning)
-						warnings = append(warnings, warning)
+					report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, warningEvent(warning))
-					} else if options.Stdout != nil {
-						if err := writeWarnings(options.Stdout, []OutputWarning{warning}); err != nil {
-							closeBackend(sourceBackend)
-							return err
-						}
-					}
 				}
 			}
 			if len(selections) == 0 {
@@ -134,11 +160,8 @@ func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options
 				for _, selection := range selections {
 					failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(selection.SourceBundle.RootRelativePath), err)
 					summary.recordFailure()
-					if jsonOutput {
+					report.Actions = append(report.Actions, errorAction(pipeline.ID, destination.ID, destination.Backend, selection.SourceBundle.RootRelativePath, err))
-						result.Actions = append(result.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))
-					} else if options.Stdout != nil {
-						writeErrorLine(options.Stdout, selection.SourceBundle.RootRelativePath, destination.ID, destination.Backend, err)
-					}
 				}
 				continue
 			}
@@ -189,22 +212,12 @@ func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options
 					plan.PathMapping = config.PathMappingFixed
 					if options.DryRun && isDestructiveFixedPathAction(plan.Action) {
 						warning := fixedPathReplacementWarning(plan)
-						if jsonOutput {
+						report.addWarning(warning)
-							warnings = append(warnings, warning)
+						report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, warningEvent(warning))
-						} else if options.Stdout != nil {
-							if err := writeWarnings(options.Stdout, []OutputWarning{warning}); err != nil {
-								deferCloseDestination()
-								closeBackend(sourceBackend)
-								return err
-							}
-						}
 					}
 				}
-				if jsonOutput {
+				report.Actions = append(report.Actions, runActionFromPlan(destination.Backend, plan, err))
-					result.Actions = append(result.Actions, runActionFromPlan(destination.Backend, plan, err))
+				report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, actionEvent(len(report.Actions)-1))
-				} else if options.Stdout != nil {
-					writePlanLine(options.Stdout, destination.Backend, plan, err)
-				}
 				if err != nil {
 					failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
 					summary.recordFailure()
@@ -230,20 +243,12 @@ func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options
 		}
 		closeBackend(sourceBackend)
 	}
-	result.Summary = summary.Result()
+	report.Summary = summary.Result()
-	if jsonOutput {
+	report.OutputErrors = failures.outputErrors()
-		if err := WriteJSONEnvelope(options.Stdout, "run", len(failures.items) == 0, warnings, result, failures.outputErrors()); err != nil {
-			return err
-		}
-	} else if options.Stdout != nil {
-		if _, err := fmt.Fprintln(options.Stdout, summary.Line()); err != nil {
-			return err
-		}
-	}
 	if len(failures.items) > 0 {
-		return failures
+		return report, failures
 	}
-	return nil
+	return report, nil
 }
 
 type closeableBackend interface {

internal/app/run_coordinator.go

@@ -0,0 +1,131 @@
+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

@@ -0,0 +1,223 @@
+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_output.go

@@ -10,61 +10,125 @@ import (
 	"gitea.maximumdirect.net/eric/distributor/internal/storage"
 )
 
-func writePlanLine(w io.Writer, backend string, plan publish.Plan, planErr error) {
+func WriteRunReport(w io.Writer, format OutputFormat, report RunReport) error {
-	if w == nil {
+	if IsJSONOutput(format) {
-		return
+		return WriteJSONEnvelope(w, "run", len(report.OutputErrors) == 0, report.Warnings, report, report.OutputErrors)
 	}
-	if planErr != nil {
+	return writeRunReportText(w, report)
-		destinationID := plan.DestinationID
+}
+
+func writeRunReportText(w io.Writer, report RunReport) error {
+	if w == nil {
+		return nil
+	}
+	if err := writeWarnings(w, report.PreambleWarnings); err != nil {
+		return err
+	}
+	if _, err := fmt.Fprintf(w, "Configured pipelines: %d\n", len(report.Pipelines)); err != nil {
+		return err
+	}
+	for _, pipeline := range report.Pipelines {
+		if err := writeWarnings(w, pipeline.Warnings); err != nil {
+			return err
+		}
+		if _, err := fmt.Fprintf(w, "- pipeline=%s source=%s bundles=%d destinations=%s\n", pipeline.ID, pipeline.SourceBackend, pipeline.BundleCount, destinationIDSummary(pipeline.Destinations)); err != nil {
+			return err
+		}
+		for _, event := range pipeline.events {
+			if event.warning != nil {
+				if err := writeWarnings(w, []OutputWarning{*event.warning}); err != nil {
+					return err
+				}
+				continue
+			}
+			if event.actionIndex < 0 || event.actionIndex >= len(report.Actions) {
+				continue
+			}
+			writeRunActionLine(w, report.Actions[event.actionIndex])
+		}
+	}
+	_, err := fmt.Fprintln(w, report.Summary.Line())
+	return err
+}
+
+func writeRunActionLine(w io.Writer, action RunActionRecord) {
+	if action.Action == "error" {
+		destinationID := action.DestinationID
 		if destinationID == "" {
 			destinationID = "unknown"
 		}
-		fmt.Fprintf(w, "  - bundle=%s destination=%s backend=%s%s action=error reason=%q\n", storage.DisplayPath(plan.BundlePath), destinationID, backend, pathMappingSummary(plan), planErr.Error())
+		fmt.Fprintf(w, "  - bundle=%s destination=%s backend=%s%s action=error reason=%q\n", action.BundlePath, destinationID, action.Backend, pathMappingRecordSummary(action), action.Reason)
 		return
 	}
-	fmt.Fprintf(w, "  - bundle=%s destination=%s backend=%s%s action=%s outputs=%s reason=%q\n", storage.DisplayPath(plan.BundlePath), plan.DestinationID, backend, pathMappingSummary(plan), plan.Action, outputSummary(plan.Outputs), plan.Reason)
+	fmt.Fprintf(w, "  - bundle=%s destination=%s backend=%s%s action=%s outputs=%s reason=%q\n", action.BundlePath, action.DestinationID, action.Backend, pathMappingRecordSummary(action), action.Action, outputRecordSummary(action.Outputs), action.Reason)
 }
 
-func pathMappingSummary(plan publish.Plan) string {
+func pathMappingRecordSummary(action RunActionRecord) string {
-	if plan.PathMapping != config.PathMappingFixed {
+	if action.PathMapping != config.PathMappingFixed {
 		return ""
 	}
-	return fmt.Sprintf(" path_mapping=fixed target=%s", storage.DisplayPath(plan.DestinationBundlePath))
+	return fmt.Sprintf(" path_mapping=fixed target=%s", action.DestinationPath)
 }
 
-func writeErrorLine(w io.Writer, bundlePath, destinationID, backend string, err error) {
+func outputRecordSummary(outputs []RunOutputRecord) string {
-	if w == nil {
-		return
-	}
-	fmt.Fprintf(w, "  - bundle=%s destination=%s backend=%s action=error reason=%q\n", storage.DisplayPath(bundlePath), destinationID, backend, err.Error())
-}
-
-func outputSummary(outputs []publish.Output) string {
 	if len(outputs) == 0 {
 		return "none"
 	}
 	paths := make([]string, 0, len(outputs))
 	for _, output := range outputs {
-		paths = append(paths, output.DestinationPath)
+		paths = append(paths, output.Path)
 	}
 	return strings.Join(paths, ",")
 }
 
-type runResult struct {
+func destinationIDSummary(ids []string) string {
-	DryRun    bool                `json:"dry_run"`
+	if len(ids) == 0 {
-	Pipelines []runPipelineResult `json:"pipelines"`
+		return "none"
-	Actions   []runActionResult   `json:"actions"`
+	}
-	Summary   runSummaryResult    `json:"summary"`
+	return strings.Join(ids, ",")
 }
 
-type runPipelineResult struct {
+type RunReport struct {
-	ID            string   `json:"id"`
+	DryRun           bool                 `json:"dry_run"`
-	SourceBackend string   `json:"source_backend"`
+	Pipelines        []RunPipelineSummary `json:"pipelines"`
-	BundleCount   int      `json:"bundle_count"`
+	Actions          []RunActionRecord    `json:"actions"`
-	Destinations  []string `json:"destinations"`
+	Summary          RunSummaryCounters   `json:"summary"`
+	Warnings         []OutputWarning      `json:"-"`
+	OutputErrors     []OutputError        `json:"-"`
+	PreambleWarnings []OutputWarning      `json:"-"`
 }
 
-type runActionResult struct {
+func (r *RunReport) addWarning(warning OutputWarning) {
+	r.Warnings = append(r.Warnings, warning)
+}
+
+func (r *RunReport) addWarnings(warnings []OutputWarning) {
+	r.Warnings = append(r.Warnings, warnings...)
+}
+
+type RunPipelineSummary struct {
+	ID            string          `json:"id"`
+	SourceBackend string          `json:"source_backend"`
+	BundleCount   int             `json:"bundle_count"`
+	Destinations  []string        `json:"destinations"`
+	Warnings      []OutputWarning `json:"-"`
+	events        []runPipelineEvent
+}
+
+type runPipelineEvent struct {
+	warning     *OutputWarning
+	actionIndex int
+}
+
+func warningEvent(warning OutputWarning) runPipelineEvent {
+	return runPipelineEvent{warning: &warning, actionIndex: -1}
+}
+
+func actionEvent(actionIndex int) runPipelineEvent {
+	return runPipelineEvent{actionIndex: actionIndex}
+}
+
+type RunActionRecord struct {
 	PipelineID      string            `json:"pipeline_id,omitempty"`
 	DestinationID   string            `json:"destination_id"`
 	Backend         string            `json:"backend"`
@@ -75,10 +139,10 @@ type runActionResult struct {
 	Action          string            `json:"action"`
 	PrimaryURL      string            `json:"primary_url,omitempty"`
 	Reason          string            `json:"reason,omitempty"`
-	Outputs         []runOutputResult `json:"outputs"`
+	Outputs         []RunOutputRecord `json:"outputs"`
 }
 
-type runOutputResult struct {
+type RunOutputRecord struct {
 	Path       string `json:"path"`
 	Kind       string `json:"kind"`
 	SourcePath string `json:"source_path,omitempty"`
@@ -88,13 +152,13 @@ type runOutputResult struct {
 	Size       int64  `json:"size"`
 }
 
-func runActionFromPlan(backend string, plan publish.Plan, planErr error) runActionResult {
+func runActionFromPlan(backend string, plan publish.Plan, planErr error) RunActionRecord {
 	if planErr != nil {
 		destinationID := plan.DestinationID
 		if destinationID == "" {
 			destinationID = "unknown"
 		}
-		return runActionResult{
+		return RunActionRecord{
 			PipelineID:      plan.PipelineID,
 			DestinationID:   destinationID,
 			Backend:         backend,
@@ -105,10 +169,10 @@ func runActionFromPlan(backend string, plan publish.Plan, planErr error) runActi
 			Action:          "error",
 			PrimaryURL:      plan.PrimaryURL,
 			Reason:          planErr.Error(),
-			Outputs:         []runOutputResult{},
+			Outputs:         []RunOutputRecord{},
 		}
 	}
-	return runActionResult{
+	return RunActionRecord{
 		PipelineID:      plan.PipelineID,
 		DestinationID:   plan.DestinationID,
 		Backend:         backend,
@@ -123,8 +187,8 @@ func runActionFromPlan(backend string, plan publish.Plan, planErr error) runActi
 	}
 }
 
-func errorAction(pipelineID, destinationID, backend, bundlePath string, err error) runActionResult {
+func errorAction(pipelineID, destinationID, backend, bundlePath string, err error) RunActionRecord {
-	return runActionResult{
+	return RunActionRecord{
 		PipelineID:      pipelineID,
 		DestinationID:   destinationID,
 		Backend:         backend,
@@ -132,15 +196,15 @@ func errorAction(pipelineID, destinationID, backend, bundlePath string, err erro
 		DestinationPath: storage.DisplayPath(bundlePath),
 		Action:          "error",
 		Reason:          err.Error(),
-		Outputs:         []runOutputResult{},
+		Outputs:         []RunOutputRecord{},
 	}
 }
 
-func runOutputsFromPlan(outputs []publish.Output) []runOutputResult {
+func runOutputsFromPlan(outputs []publish.Output) []RunOutputRecord {
-	results := make([]runOutputResult, 0, len(outputs))
+	results := make([]RunOutputRecord, 0, len(outputs))
 	for _, output := range outputs {
 		stateOutput := output.StateOutputFile()
-		results = append(results, runOutputResult{
+		results = append(results, RunOutputRecord{
 			Path:       stateOutput.Path,
 			Kind:       stateOutput.Kind,
 			SourcePath: stateOutput.SourcePath,

internal/app/run_selection.go

@@ -3,7 +3,6 @@ package app
 import (
 	"fmt"
 	"sort"
-	"strings"
 
 	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
 	"gitea.maximumdirect.net/eric/distributor/internal/config"
@@ -78,14 +77,3 @@ func destinationIDs(destinations []config.Destination) []string {
 	}
 	return ids
 }
-
-func destinationSummary(destinations []config.Destination) string {
-	if len(destinations) == 0 {
-		return "none"
-	}
-	ids := make([]string, 0, len(destinations))
-	for _, destination := range destinations {
-		ids = append(ids, destination.ID)
-	}
-	return strings.Join(ids, ",")
-}

internal/app/run_summary.go

@@ -39,15 +39,7 @@ func (s *runSummary) recordFixedPath() {
 	s.fixedPath++
 }
 
-func (s runSummary) Line() string {
+type RunSummaryCounters struct {
-	status := "ok"
-	if s.failures > 0 {
-		status = "failed"
-	}
-	return fmt.Sprintf("Final status: %s planned=%d publish_new=%d replace_older=%d force_replace=%d skipped=%d failed=%d dry_run=%t fixed_path=%d", status, s.planned, s.publishNew, s.replaceOlder, s.forceReplace, s.skipped, s.failures, s.dryRun, s.fixedPath)
-}
-
-type runSummaryResult struct {
 	Status       string `json:"status"`
 	Planned      int    `json:"planned"`
 	PublishNew   int    `json:"publish_new"`
@@ -59,12 +51,16 @@ type runSummaryResult struct {
 	FixedPath    int    `json:"fixed_path"`
 }
 
-func (s runSummary) Result() runSummaryResult {
+func (s RunSummaryCounters) Line() string {
+	return fmt.Sprintf("Final status: %s planned=%d publish_new=%d replace_older=%d force_replace=%d skipped=%d failed=%d dry_run=%t fixed_path=%d", s.Status, s.Planned, s.PublishNew, s.ReplaceOlder, s.ForceReplace, s.Skipped, s.Failed, s.DryRun, s.FixedPath)
+}
+
+func (s runSummary) Result() RunSummaryCounters {
 	status := "ok"
 	if s.failures > 0 {
 		status = "failed"
 	}
-	return runSummaryResult{
+	return RunSummaryCounters{
 		Status:       status,
 		Planned:      s.planned,
 		PublishNew:   s.publishNew,

internal/app/run_test.go

@@ -718,6 +718,176 @@ func TestRunJSONIncludesGeneratedOutputMetadata(t *testing.T) {
 	}
 }
 
+func TestBuildRunReportIncludesStructuredDryRunResults(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+	configPath := testutil.WriteLocalConfigWithLinks(t, sourceRoot, destinationRoot, config.PathMappingFixed, "https://reports.example.com/latest", config.LinkPrimaryAuto, false, true, config.TransformModeIndex)
+	cfg, err := config.LoadFile(configPath)
+	if err != nil {
+		t.Fatalf("load config: %v", err)
+	}
+
+	report, err := buildRunReportWithBackendFactory(context.Background(), cfg, RunOptions{DryRun: true}, newBackendFactoryWithEnvironment)
+	if err != nil {
+		t.Fatalf("buildRunReportWithBackendFactory() error = %v", err)
+	}
+	if !report.DryRun || report.Summary.Status != "ok" || !report.Summary.DryRun {
+		t.Fatalf("report dry-run/status = dry_run:%t summary:%#v, want ok dry-run", report.DryRun, report.Summary)
+	}
+	if got, want := len(report.Pipelines), 1; got != want {
+		t.Fatalf("pipeline count = %d, want %d", got, want)
+	}
+	pipeline := report.Pipelines[0]
+	if pipeline.ID != "reports" || pipeline.SourceBackend != config.BackendLocal || pipeline.BundleCount != 1 || strings.Join(pipeline.Destinations, ",") != "archive" {
+		t.Fatalf("pipeline summary = %#v, want reports/local bundle summary", pipeline)
+	}
+	if got, want := len(report.Warnings), 1; got != want {
+		t.Fatalf("warning count = %d, want %d", got, want)
+	}
+	if !strings.Contains(report.Warnings[0].Message, "path_mapping=fixed candidates=1 selected_bundle=.") {
+		t.Fatalf("warning = %#v, want fixed path selection", report.Warnings[0])
+	}
+	if got, want := len(report.Actions), 1; got != want {
+		t.Fatalf("action count = %d, want %d", got, want)
+	}
+	action := report.Actions[0]
+	if action.PipelineID != "reports" || action.DestinationID != "archive" || action.Action != "publish_new" || action.PrimaryURL != "https://reports.example.com/latest/" {
+		t.Fatalf("action = %#v, want publish_new with primary URL", action)
+	}
+	if action.PathMapping != config.PathMappingFixed || action.DestinationPath != "." {
+		t.Fatalf("action path mapping = %q destination path = %q, want fixed root", action.PathMapping, action.DestinationPath)
+	}
+	if got, want := len(action.Outputs), 1; got != want {
+		t.Fatalf("output count = %d, want %d", got, want)
+	}
+	output := action.Outputs[0]
+	if output.Path != "index.html" || output.Kind != state.OutputKindGenerated || output.SourcePath != "report.md" || output.Transform != "markdown_to_html" || output.URL != "https://reports.example.com/latest/" {
+		t.Fatalf("output = %#v, want generated index metadata", output)
+	}
+	if report.Summary.Planned != 1 || report.Summary.PublishNew != 1 || report.Summary.FixedPath != 1 || report.Summary.Failed != 0 {
+		t.Fatalf("summary = %#v, want publish_new fixed path counters", report.Summary)
+	}
+	if len(report.OutputErrors) != 0 {
+		t.Fatalf("output errors = %#v, want none", report.OutputErrors)
+	}
+}
+
+func TestBuildRunReportIncludesPartialFailures(t *testing.T) {
+	sourceRoot := t.TempDir()
+	firstDestination := t.TempDir()
+	secondDestination := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+	if err := os.WriteFile(filepath.Join(firstDestination, "unmanaged.txt"), []byte("data"), 0o600); err != nil {
+		t.Fatalf("write unmanaged file: %v", err)
+	}
+	cfg, err := config.LoadFile(writeFanoutConfig(t, sourceRoot, firstDestination, secondDestination))
+	if err != nil {
+		t.Fatalf("load config: %v", err)
+	}
+
+	report, err := buildRunReportWithBackendFactory(context.Background(), cfg, RunOptions{}, newBackendFactoryWithEnvironment)
+	if err == nil || !IsPartialResultError(err) {
+		t.Fatalf("buildRunReportWithBackendFactory() error = %v, want partial result error", err)
+	}
+	if report.Summary.Status != "failed" || report.Summary.Planned != 1 || report.Summary.PublishNew != 1 || report.Summary.Failed != 1 {
+		t.Fatalf("summary = %#v, want one planned publish and one failure", report.Summary)
+	}
+	if got, want := len(report.Actions), 2; got != want {
+		t.Fatalf("action count = %d, want %d", got, want)
+	}
+	if report.Actions[0].DestinationID != "archive-one" || report.Actions[0].Action != "error" || !strings.Contains(report.Actions[0].Reason, "fail_unmanaged") {
+		t.Fatalf("first action = %#v, want archive-one error", report.Actions[0])
+	}
+	if report.Actions[1].DestinationID != "archive-two" || report.Actions[1].Action != "publish_new" {
+		t.Fatalf("second action = %#v, want archive-two publish_new", report.Actions[1])
+	}
+	if got, want := len(report.OutputErrors), 1; got != want {
+		t.Fatalf("output error count = %d, want %d", got, want)
+	}
+	outputError := report.OutputErrors[0]
+	if outputError.PipelineID != "reports" || outputError.DestinationID != "archive-one" || outputError.Backend != config.BackendLocal || outputError.BundlePath != "." || !strings.Contains(outputError.Message, "fail_unmanaged") {
+		t.Fatalf("output error = %#v, want archive-one unmanaged failure", outputError)
+	}
+}
+
+func TestRunPipelineRunsOnlyRequestedPipeline(t *testing.T) {
+	firstSource := t.TempDir()
+	secondSource := t.TempDir()
+	firstDestination := t.TempDir()
+	secondDestination := t.TempDir()
+	writeSourceBundle(t, firstSource, "", testBundleOptions{ID: "reports.one"})
+	writeSourceBundle(t, secondSource, "", testBundleOptions{ID: "reports.two"})
+	configPath := writeTwoPipelineConfig(t, firstSource, firstDestination, secondSource, secondDestination)
+	notifier := &recordingNotifier{}
+
+	report, err := RunPipeline(context.Background(), RunPipelineOptions{
+		ConfigPath: configPath,
+		PipelineID: "reports-one",
+		Notifier:   notifier,
+	})
+	if err != nil {
+		t.Fatalf("RunPipeline() error = %v", err)
+	}
+	if got, want := len(report.Pipelines), 1; got != want {
+		t.Fatalf("pipeline count = %d, want %d", got, want)
+	}
+	if report.Pipelines[0].ID != "reports-one" {
+		t.Fatalf("pipeline id = %q, want reports-one", report.Pipelines[0].ID)
+	}
+	if got, want := len(report.Actions), 1; got != want {
+		t.Fatalf("action count = %d, want %d", got, want)
+	}
+	if report.Actions[0].PipelineID != "reports-one" || report.Actions[0].Action != "publish_new" {
+		t.Fatalf("action = %#v, want reports-one publish_new", report.Actions[0])
+	}
+	if got, want := len(notifier.events), 1; got != want {
+		t.Fatalf("notification count = %d, want %d", got, want)
+	}
+	if notifier.events[0].PipelineID != "reports-one" {
+		t.Fatalf("notification pipeline = %q, want reports-one", notifier.events[0].PipelineID)
+	}
+	testutil.AssertFile(t, filepath.Join(firstDestination, "report.md"), "# Report\nSunny.\n")
+	if entries, err := os.ReadDir(secondDestination); err != nil || len(entries) != 0 {
+		t.Fatalf("second destination entries = %v err=%v, want empty", entries, err)
+	}
+}
+
+func TestRunPipelineUnknownIDReturnsNotFound(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+
+	_, err := RunPipeline(context.Background(), RunPipelineOptions{
+		ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot),
+		PipelineID: "missing",
+	})
+	if err == nil || !IsPipelineNotFound(err) {
+		t.Fatalf("RunPipeline() error = %v, want pipeline not found", err)
+	}
+	if !strings.Contains(err.Error(), `pipeline "missing" not found`) {
+		t.Fatalf("RunPipeline() error = %v, want pipeline id in message", err)
+	}
+}
+
+func TestRunStillRunsAllConfiguredPipelines(t *testing.T) {
+	firstSource := t.TempDir()
+	secondSource := t.TempDir()
+	firstDestination := t.TempDir()
+	secondDestination := t.TempDir()
+	writeSourceBundle(t, firstSource, "", testBundleOptions{ID: "reports.one"})
+	writeSourceBundle(t, secondSource, "", testBundleOptions{ID: "reports.two"})
+
+	err := Run(context.Background(), RunOptions{
+		ConfigPath: writeTwoPipelineConfig(t, firstSource, firstDestination, secondSource, secondDestination),
+	})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	testutil.AssertFile(t, filepath.Join(firstDestination, "report.md"), "# Report\nSunny.\n")
+	testutil.AssertFile(t, filepath.Join(secondDestination, "report.md"), "# Report\nSunny.\n")
+}
+
 func TestRunDoesNotNotifyForSkippedDestination(t *testing.T) {
 	sourceRoot := t.TempDir()
 	destinationRoot := t.TempDir()
@@ -1346,6 +1516,29 @@ func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestina
 	return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination)
 }
 
+func writeTwoPipelineConfig(t *testing.T, firstSource, firstDestination, secondSource, secondDestination string) string {
+	t.Helper()
+	return writeConfigFile(t, `
+pipelines:
+  - id: reports-one
+    source:
+      backend: local
+      path: `+firstSource+`
+    destinations:
+      - id: archive
+        backend: local
+        path: `+firstDestination+`
+  - id: reports-two
+    source:
+      backend: local
+      path: `+secondSource+`
+    destinations:
+      - id: archive
+        backend: local
+        path: `+secondDestination+`
+`)
+}
+
 func writeConfigFile(t *testing.T, body string) string {
 	t.Helper()
 	path := filepath.Join(t.TempDir(), "config.yml")

internal/app/source_select.go

@@ -2,6 +2,7 @@ package app
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
@@ -9,6 +10,19 @@ import (
 	"gitea.maximumdirect.net/eric/distributor/internal/storage"
 )
 
+type PipelineNotFoundError struct {
+	ID string
+}
+
+func (e PipelineNotFoundError) Error() string {
+	return fmt.Sprintf("pipeline %q not found", e.ID)
+}
+
+func IsPipelineNotFound(err error) bool {
+	var notFound PipelineNotFoundError
+	return errors.As(err, &notFound)
+}
+
 type sourceCommandOptions struct {
 	CommandName string
 	Path        string
@@ -70,7 +84,7 @@ func selectSourceBundlesFromConfig(ctx context.Context, cfg config.Config, optio
 	}
 	pipeline, ok := findPipeline(cfg, options.PipelineID)
 	if !ok {
-		return sourceSelection{}, fmt.Errorf("pipeline %q not found", options.PipelineID)
+		return sourceSelection{}, PipelineNotFoundError{ID: options.PipelineID}
 	}
 	backends := provider(secretLoad.Environment)
 	sourceBackend, err := backends.openSource(ctx, pipeline.Source)

internal/cli/root_test.go

@@ -603,8 +603,12 @@ func TestExecuteRunDryRun(t *testing.T) {
 	if code != exitOK {
 		t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
 	}
-	if !strings.Contains(stdout.String(), "action=publish_new") {
+	wantStdout := "Configured pipelines: 1\n" +
-		t.Fatalf("stdout = %q, want config summary", stdout.String())
+		"- pipeline=reports source=local bundles=1 destinations=archive\n" +
+		"  - bundle=. destination=archive backend=local action=publish_new outputs=report.md,summary.txt reason=\"destination state is absent\"\n" +
+		"Final status: ok planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true fixed_path=0\n"
+	if got := stdout.String(); got != wantStdout {
+		t.Fatalf("stdout = %q, want %q", got, wantStdout)
 	}
 	if stderr.Len() != 0 {
 		t.Fatalf("stderr = %q, want empty", stderr.String())
@@ -631,6 +635,14 @@ func TestExecuteRunJSONDryRun(t *testing.T) {
 	if result["dry_run"] != true {
 		t.Fatalf("result = %#v, want dry_run true", result)
 	}
+	pipelines, ok := result["pipelines"].([]any)
+	if !ok || len(pipelines) != 1 {
+		t.Fatalf("pipelines = %#v, want one pipeline", result["pipelines"])
+	}
+	pipeline, ok := pipelines[0].(map[string]any)
+	if !ok || pipeline["id"] != "reports" || pipeline["source_backend"] != "local" || pipeline["bundle_count"] != float64(1) {
+		t.Fatalf("pipeline = %#v, want reports/local summary", pipelines[0])
+	}
 	actions, ok := result["actions"].([]any)
 	if !ok || len(actions) != 1 {
 		t.Fatalf("actions = %#v, want one action", result["actions"])
@@ -639,6 +651,14 @@ func TestExecuteRunJSONDryRun(t *testing.T) {
 	if !ok || action["action"] != "publish_new" {
 		t.Fatalf("action = %#v, want publish_new", actions[0])
 	}
+	outputs, ok := action["outputs"].([]any)
+	if !ok || len(outputs) != 2 {
+		t.Fatalf("outputs = %#v, want source outputs", action["outputs"])
+	}
+	summary, ok := result["summary"].(map[string]any)
+	if !ok || summary["status"] != "ok" || summary["planned"] != float64(1) || summary["publish_new"] != float64(1) || summary["dry_run"] != true {
+		t.Fatalf("summary = %#v, want ok dry-run publish counters", result["summary"])
+	}
 	if stderr.Len() != 0 {
 		t.Fatalf("stderr = %q, want empty", stderr.String())
 	}