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;
-3. discovers validated bundles from the source root;
-4. selects source bundles for each destination according to destination path mapping;
-5. opens each destination backend independently;
-6. builds publish plans for the selected bundle and destination combinations;
-7. prints plan lines or JSON action records and records summary counters;
-8. executes publish or replacement plans unless dry-run is enabled;
-9. invokes the notifier after successful publish or replacement actions.
+2. loads configured secret files into a config-owned environment resolver;
+3. builds the app-level backend factory and transform registry;
+4. opens each selected pipeline source backend;
+5. discovers validated source bundles from the source root;
+6. selects source bundles for each destination according to path mapping;
+7. opens destination backends independently;
+8. builds publish plans for selected bundle and destination combinations;
+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;
-- `run_warnings.go`: secret and SSH warning data;
-- `run_output.go`: text plan lines, JSON action records, and output projections;
-- `run_summary.go`: summary counters and JSON summary records;
-- `run_failures.go`: destination failure aggregation and partial-result detection;
+Coordinator records contain a run ID, pipeline ID, status, timestamps, completed
+report, and error text when applicable. Active state is memory-only and is
+cleared after success, failure, unknown pipeline ID, or context cancellation.
+
+The admission context is checked before a run is accepted. Once accepted, the
+run uses the coordinator lifetime context, so caller cancellation can stop
+waiting for admission without owning the actual run lifetime.
+
+The coordinator does not queue duplicate runs, persist run records, or define
+transport endpoints.
+
+## 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.
-
-## Failure behavior
-
-`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.
+Dry-run 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 invoke notifications.
 
 ## 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
-`distributor`. Implemented behavior is documented in the current user,
-operator, internal, policy, integration, and example documentation:
+## Purpose
 
-- `README.md`
-- `docs/cli.md`
-- `docs/config.md`
-- `docs/operations.md`
-- `docs/troubleshooting.md`
-- `docs/internal/`
-- `docs/integrations/markdown.md`
-- `docs/policy/`
-- `examples/`
+This roadmap records the accepted boundary for a future HTTP API. The current
+application exposes CLI commands and internal app-layer run contracts; it does
+not implement an HTTP server, HTTP routes, a `serve` command, app-level
+authentication, or in-app TLS.
 
-`distributor` currently supports local, SSH/SFTP, and S3-compatible source and
-destination backends; producer bundle creation through `pkg/bundle` and
-`distributor manifest create`; configured source validation and inspection;
-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.
+Implemented internal run contracts are documented in `docs/internal/app.md`.
+This file is the canonical home for future HTTP boundary decisions until an
+HTTP implementation roadmap replaces it.
 
-## Future Work
+## Accepted Direction
 
-These items are not implemented. They should not be documented as current
-behavior outside `docs/roadmap/` unless a future implementation adds them.
+The future HTTP API should trigger configured distributor pipelines through the
+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
-  around shared root options.
-- Add output formats beyond `text` and `json` only if a concrete consumer
-  requires them.
-- Add a versioned JSON schema reference after the first JSON-capable release.
-- Add destination-state inspection behind an explicit flag such as
-  `--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.
+- A trigger request accepts only a pipeline ID as application input.
+- A trigger request starts work asynchronously and returns a run ID after
+  admission.
+- Run status is read through a separate status endpoint keyed by run ID.
+- Status records expose run ID, pipeline ID, current status, timestamps, and
+  completed report or error details when available.
 
-### 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
-  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.
+## Error Mapping
 
-### 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
-  multi-page aggregation.
-- Add richer transform metadata only if future state consumers need more than
-  the transform name and output path.
-- 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.
+- Unknown pipeline IDs map to `404 Not Found`.
+- Duplicate in-flight runs for the same pipeline ID map to `409 Conflict`.
+- Validation, config, source, destination, publish, transform, and notification
+  errors map to transport errors according to their app-layer context.
 
-### 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
-  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.
+## Context And Lifetime
 
-### 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
-  credential paths only when a concrete backend workflow requires them.
-- 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.
+This split allows a client disconnect or request timeout to stop waiting for
+admission without canceling a run that has already been accepted.
 
-## 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;
-- define which current docs must change after implementation;
-- keep examples secret-free and runnable or clearly environment-gated;
-- keep workflow labels out of production code, tests, config fields, and
-  user-facing documentation;
-- run focused tests for the changed behavior and `go test ./...` for
-  cross-package changes.
+The first HTTP implementation should not include:
+
+- bearer-token authentication;
+- in-app TLS configuration;
+- public-network exposure defaults.
+
+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{},
-		Actions:   []runActionResult{},
+		Pipelines: []RunPipelineSummary{},
+		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 {
-		warnings = append(warnings, secretWarnings...)
-	} else if options.Stdout != nil {
-		if err := writeWarnings(options.Stdout, secretWarnings); err != nil {
-			return err
-		}
-	}
+	report.PreambleWarnings = append(report.PreambleWarnings, secretWarnings...)
+	report.addWarnings(secretWarnings)
 	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 {
-			warnings = append(warnings, pipelineWarnings...)
-		} else if options.Stdout != nil {
-			if err := writeWarnings(options.Stdout, pipelineWarnings); err != nil {
-				return err
-			}
-		}
+		report.addWarnings(pipelineWarnings)
 		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 {
-			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
-			}
-		}
+		pipelineIndex := len(report.Pipelines) - 1
 		for _, destination := range pipeline.Destinations {
 			selections := selectDestinationBundles(destination, bundles)
 			if isFixedPathDestination(destination) {
 				summary.recordFixedPath()
 				if options.DryRun {
 					warning := fixedPathSelectionWarning(pipeline.ID, destination.ID, selections, len(bundles))
-					if jsonOutput {
-						warnings = append(warnings, warning)
-					} else if options.Stdout != nil {
-						if err := writeWarnings(options.Stdout, []OutputWarning{warning}); err != nil {
-							closeBackend(sourceBackend)
-							return err
-						}
-					}
+					report.addWarning(warning)
+					report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, warningEvent(warning))
 				}
 			}
 			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 {
-						result.Actions = append(result.Actions, errorAction(pipeline.ID, destination.ID, destination.Backend, selection.SourceBundle.RootRelativePath, err))
-					} else if options.Stdout != nil {
-						writeErrorLine(options.Stdout, selection.SourceBundle.RootRelativePath, destination.ID, destination.Backend, err)
-					}
+					report.Actions = append(report.Actions, errorAction(pipeline.ID, destination.ID, destination.Backend, selection.SourceBundle.RootRelativePath, err))
+					report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, actionEvent(len(report.Actions)-1))
 				}
 				continue
 			}
@@ -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 {
-							warnings = append(warnings, warning)
-						} else if options.Stdout != nil {
-							if err := writeWarnings(options.Stdout, []OutputWarning{warning}); err != nil {
-								deferCloseDestination()
-								closeBackend(sourceBackend)
-								return err
-							}
-						}
+						report.addWarning(warning)
+						report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, warningEvent(warning))
 					}
 				}
-				if jsonOutput {
-					result.Actions = append(result.Actions, runActionFromPlan(destination.Backend, plan, err))
-				} else if options.Stdout != nil {
-					writePlanLine(options.Stdout, destination.Backend, plan, err)
-				}
+				report.Actions = append(report.Actions, runActionFromPlan(destination.Backend, plan, err))
+				report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, actionEvent(len(report.Actions)-1))
 				if err != nil {
 					failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
 					summary.recordFailure()
@@ -230,20 +243,12 @@ func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options
 		}
 		closeBackend(sourceBackend)
 	}
-	result.Summary = summary.Result()
-	if jsonOutput {
-		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
-		}
-	}
+	report.Summary = summary.Result()
+	report.OutputErrors = failures.outputErrors()
 	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) {
-	if w == nil {
-		return
+func WriteRunReport(w io.Writer, format OutputFormat, report RunReport) error {
+	if IsJSONOutput(format) {
+		return WriteJSONEnvelope(w, "run", len(report.OutputErrors) == 0, report.Warnings, report, report.OutputErrors)
 	}
-	if planErr != nil {
-		destinationID := plan.DestinationID
+	return writeRunReportText(w, report)
+}
+
+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 {
-	if plan.PathMapping != config.PathMappingFixed {
+func pathMappingRecordSummary(action RunActionRecord) string {
+	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) {
-	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 {
+func outputRecordSummary(outputs []RunOutputRecord) 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 {
-	DryRun    bool                `json:"dry_run"`
-	Pipelines []runPipelineResult `json:"pipelines"`
-	Actions   []runActionResult   `json:"actions"`
-	Summary   runSummaryResult    `json:"summary"`
+func destinationIDSummary(ids []string) string {
+	if len(ids) == 0 {
+		return "none"
+	}
+	return strings.Join(ids, ",")
 }
 
-type runPipelineResult struct {
-	ID            string   `json:"id"`
-	SourceBackend string   `json:"source_backend"`
-	BundleCount   int      `json:"bundle_count"`
-	Destinations  []string `json:"destinations"`
+type RunReport struct {
+	DryRun           bool                 `json:"dry_run"`
+	Pipelines        []RunPipelineSummary `json:"pipelines"`
+	Actions          []RunActionRecord    `json:"actions"`
+	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 {
-	return runActionResult{
+func errorAction(pipelineID, destinationID, backend, bundlePath string, err error) RunActionRecord {
+	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 {
-	results := make([]runOutputResult, 0, len(outputs))
+func runOutputsFromPlan(outputs []publish.Output) []RunOutputRecord {
+	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 {
-	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 {
+type RunSummaryCounters 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") {
-		t.Fatalf("stdout = %q, want config summary", stdout.String())
+	wantStdout := "Configured pipelines: 1\n" +
+		"- 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())
 	}