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,169 +1,91 @@
-# HTTP API Stabilization Roadmap
+# HTTP API Boundary Roadmap
 
-## Summary
+## Purpose
 
-Prepare `distributor` for a later HTTP API without implementing the HTTP
+This roadmap records the accepted boundary for a future HTTP API. The current
-server in this roadmap. The goal is to make the existing run workflow callable
+application exposes CLI commands and internal app-layer run contracts; it does
-as a single-pipeline, structured, concurrency-safe application use case.
+not implement an HTTP server, HTTP routes, a `serve` command, app-level
+authentication, or in-app TLS.
 
-Chosen defaults:
+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.
 
-- Scope: stabilization only; HTTP routes and `serve` command are deferred to a
+## Accepted Direction
-  later roadmap.
-- Future HTTP trigger mode: asynchronous start with run status.
-- Security boundary: private bind or reverse proxy/mTLS outside the app; no
-  app-level auth or in-app TLS in v1.
-- Trigger input: pipeline ID only.
 
-## Implementation Rules
+The future HTTP API should trigger configured distributor pipelines through the
+existing app-layer single-pipeline run path and in-memory coordinator.
 
-- Implement stages in order; each stage should be one prompt or commit unless
+The HTTP API is intentionally narrow:
-  trivially small.
-- Before each stage, read `docs/policy/architecture.md`,
-  `docs/policy/development.md`, and `docs/policy/documentation.md`.
-- Preserve existing CLI behavior, text output, JSON output, config semantics,
-  backend behavior, manifest schema, and destination state schema.
-- Do not introduce external dependencies, a CLI framework, a generic workflow
-  engine, a plugin system, durable run storage, or broad adapter abstractions.
-- Do not document the HTTP API as implemented outside `docs/roadmap/`.
 
-## Stages
+- 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.
 
-### Stage 1: Structured Run Core
+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.
 
-Refactor `internal/app` so the run workflow first produces a structured run
+## Error Mapping
-report, then projects that report to text or JSON output.
 
-Required behavior:
+Future transport code should map app-layer errors without changing app-layer
+error ownership:
 
-- Keep `app.Run` as the CLI-facing entrypoint.
+- Unknown pipeline IDs map to `404 Not Found`.
-- Move stdout writes out of the core planning/execution loop where practical.
+- Duplicate in-flight runs for the same pipeline ID map to `409 Conflict`.
-- Preserve the existing `run --format text` and `run --format json` output
+- Validation, config, source, destination, publish, transform, and notification
-  byte-for-byte except where tests already allow map ordering.
+  errors map to transport errors according to their app-layer context.
-- Preserve partial-result behavior: destination-scoped failures produce a
-  structured report plus an aggregated error.
 
-Suggested internal shape:
+Duplicate runs must not be queued. Run state remains in memory unless a later
+roadmap explicitly adds durable run storage.
 
-- Introduce an app-owned `RunReport` model containing dry-run state, pipeline
+## Context And Lifetime
-  summaries, action records, summary counters, warnings, and output errors.
-- Keep JSON tags compatible with the current `run` JSON output.
-- Keep failure aggregation in `internal/app`; do not move it into `publish`,
-  `config`, or storage adapters.
 
-Tests:
+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 or update `internal/app` tests proving structured reports include
+This split allows a client disconnect or request timeout to stop waiting for
-  warnings, actions, outputs, summary counters, and partial failures.
+admission without canceling a run that has already been accepted.
-- Add CLI regression coverage for existing text and JSON output.
-- Run `go test ./internal/app ./internal/cli`.
 
-### Stage 2: Single-Pipeline Run Entry Point
+## Security Boundary
 
-Add an app-layer entrypoint for running exactly one configured pipeline by ID.
+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.
 
-Required behavior:
+The first HTTP implementation should not include:
 
-- Add an exported internal app function such as
+- bearer-token authentication;
-  `RunPipeline(ctx, RunPipelineOptions) (RunReport, error)`.
+- in-app TLS configuration;
-- `RunPipelineOptions` must include config path, pipeline ID, dry-run, force,
+- public-network exposure defaults.
-  and optional notifier.
-- Invalid pipeline IDs must return a typed or predicate-detectable error
-  suitable for later HTTP `404` mapping.
-- `RunPipeline` must reuse the same backend factory, secret loading, transform
-  registry, warning generation, destination planning, publish execution,
-  notification behavior, and failure aggregation as `Run`.
-- Do not add a public CLI `run --pipeline` flag in this stage.
 
-Tests:
+A later roadmap must explicitly change this security decision before any of
+those features are added.
 
-- Run only the requested pipeline from a multi-pipeline config.
+## Non-Goals
-- Return the chosen not-found error for an unknown pipeline ID.
-- Preserve existing `app.Run` all-pipelines behavior.
-- Run `go test ./internal/app ./internal/cli`.
 
-### Stage 3: In-Memory Pipeline Run Coordinator
+The future HTTP API should not add:
 
-Add a narrow concurrency coordinator around the single-pipeline app seam.
+- 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.
 
-Required behavior:
+## Verification Expectations
 
-- Keep the coordinator in `internal/app` unless a later HTTP implementation
+Any later HTTP implementation should preserve existing CLI behavior and keep
-  introduces a transport package.
+the app-layer run contracts tested. At minimum, it should cover:
-- Allow different pipeline IDs to run concurrently.
-- Reject a second in-flight run for the same pipeline ID with a typed or
-  predicate-detectable duplicate-run error.
-- Track active runs in memory only.
-- Clear active state after success, failure, or context cancellation.
-- Do not queue duplicate runs and do not persist run state to disk or a
-  database.
 
-Async-ready constraints:
+- trigger requests with known and unknown pipeline IDs;
-
+- duplicate in-flight trigger requests;
-- Model run IDs and status records so a later HTTP API can expose asynchronous
+- asynchronous acceptance and status lookup;
-  status.
+- private bind defaults;
-- A future HTTP request context should guard admission; the actual run should
+- request-context admission behavior;
-  be tied to a server/coordinator lifetime context.
+- coordinator-lifetime run execution.
-
-Tests:
-
-- Concurrent same-pipeline requests produce exactly one accepted run and one
-  duplicate-run error.
-- Concurrent different-pipeline requests both start.
-- Active state is cleared after success and after failure.
-- Unknown pipeline IDs do not leave active state behind.
-- Run `go test ./internal/app`.
-
-### Stage 4: Future HTTP Boundary Contract
-
-Record the target HTTP boundary in this roadmap only; do not implement routes
-yet.
-
-Required target contract:
-
-- Future trigger endpoint should accept only a pipeline ID.
-- Future trigger behavior should be asynchronous: accept the run, return a run
-  ID, and expose status through a later status endpoint.
-- Future duplicate in-flight pipeline runs should map to `409 Conflict`.
-- Future unknown pipeline IDs should map to `404 Not Found`.
-- Future server should default to private binding, such as `127.0.0.1`, and
-  rely on a reverse proxy, private network, or external mTLS for transport
-  security.
-- Do not add bearer-token auth, in-app TLS config, or public-network exposure in
-  v1 unless a later roadmap explicitly changes this decision.
-
-Tests:
-
-- Documentation-only stage; no tests required beyond any repository doc checks
-  that exist.
-
-### Stage 5: Stabilization Sweep
-
-Clean up only code made obsolete by stages 1-3.
-
-Required behavior:
-
-- Remove duplicate result projection, warning collection, and
-  pipeline-selection helpers only when clearly replaced.
-- Keep test helpers local unless they are broadly reusable.
-- Update `docs/internal/app.md` only for implemented internal app contracts.
-- Do not update `docs/cli.md`, `docs/config.md`, `docs/operations.md`, or
-  README for HTTP behavior because no HTTP API exists yet.
-
-Tests:
-
-- Run `go test ./internal/app ./internal/cli`.
-- Run `go test ./...` before considering the roadmap complete.
-
-## Acceptance Criteria
-
-The roadmap is complete when:
-
-- Existing `distributor run` behavior is unchanged.
-- `internal/app` exposes a tested single-pipeline run path.
-- Run results are available as structured data without scraping stdout.
-- The coordinator prevents concurrent same-pipeline runs while allowing
-  different pipelines to run.
-- Future HTTP status, duplicate, not-found, and private-bind decisions are
-  recorded under `docs/roadmap/`.
-- The full test suite passes with `go test ./...`.

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())
 	}