Add implementation roadmap for HTTP API

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/http.md

@@ -0,0 +1,349 @@
+# Roadmap: HTTP Upload API
+
+## Purpose
+
+Add an HTTP upload API that lets producer applications push complete source
+bundles into `distributor`.
+
+The current application already supports local, SSH/SFTP, and S3-compatible
+source and destination backends. Those source backends are pull-oriented:
+`distributor` opens configured storage, discovers `manifest.json`, validates
+bundles, and fans out selected outputs to configured destinations.
+
+`http_upload` is different because it is push-oriented. A producer sends one
+complete bundle to `distributor`, and `distributor` stages that upload before
+running normal validation and fan-out. The HTTP layer should be an ingestion
+layer over the existing app, bundle, storage, publish, transform, link, state,
+and notification behavior.
+
+## Current Implementation Grounding
+
+Implemented behavior already provides the core pieces this feature should reuse:
+
+- source bundle validation through the bundle package;
+- local, SSH/SFTP, and S3-compatible storage backends;
+- destination fan-out through publish planning and execution;
+- single-pipeline app execution through the app layer;
+- structured run reports and JSON-capable CLI output;
+- secrets-directory environment resolution for credential material.
+
+The HTTP implementation should not duplicate bundle validation or publication
+logic. Once an upload is staged, it should proceed through the same validation
+and fan-out behavior as any other source bundle.
+
+## Accepted Direction
+
+Add `http_upload` as a source backend option for configured pipelines.
+
+An `http_upload` source is not a normal durable storage backend. It represents
+an HTTP ingestion endpoint that receives an uploaded bundle, writes it into
+pipeline-local staging storage, validates it, and then dispatches the existing
+pipeline fan-out flow.
+
+Accepted behavior:
+
+- producer applications upload a compliant source bundle manifest and all
+  referenced files;
+- uploads are asynchronous;
+- each accepted upload receives a generated run id and an initial `accepted`
+  status;
+- clients can query run status later by run id;
+- authentication uses a static token associated with the selected
+  `http_upload` pipeline;
+- upload requests send that token with `Authorization: Bearer <token>`;
+- tokens may be supplied through `secrets.directory` using the existing
+  internal environment resolver;
+- each `http_upload` pipeline has a configurable upload staging directory;
+- default staging root is `/var/spool/distributor`;
+- default pipeline staging directory is `/var/spool/distributor/<pipeline_id>`;
+- default maximum upload size is 20 MB;
+- upload archives may be uncompressed tar or gzip-compressed tar;
+- the server must prevent multiple simultaneous active runs of the same
+  pipeline;
+- the server uses an internal bounded queue and a global `max_concurrency`
+  setting for accepted uploads;
+- completed status records and staged run directories use time-based retention;
+- every accepted upload is a new run with a `distributor`-generated run id.
+
+## Configuration Shape
+
+Add `http_upload` as a source backend only. It should not be valid as a
+destination backend.
+
+Add top-level HTTP server configuration for cross-pipeline server behavior:
+
+```yaml
+server:
+  http:
+    bind: 127.0.0.1:8080
+    staging_root: /var/spool/distributor
+    max_upload_size: 20MB
+    queue_size: 16
+    max_concurrency: 1
+    retention: 24h
+```
+
+Defaults:
+
+- `bind`: `127.0.0.1:8080`;
+- `staging_root`: `/var/spool/distributor`;
+- `max_upload_size`: `20MB`;
+- `queue_size`: `16`;
+- `max_concurrency`: `1`;
+- `retention`: `24h`.
+
+Pipeline shape:
+
+```yaml
+pipelines:
+  - id: weather-daily
+    source:
+      backend: http_upload
+      token_env: WEATHER_DAILY_UPLOAD_TOKEN
+      staging_path: /var/spool/distributor/weather-daily
+      max_upload_size: 20MB
+    destinations:
+      - id: archive
+        backend: s3
+        ...
+```
+
+`token_env` is required for the first implementation. Literal token values in
+YAML are not supported. Real environment variables or `secrets.directory` files
+provide the token without putting secret values in config.
+
+If `staging_path` is omitted, default it to:
+
+```text
+/var/spool/distributor/<pipeline_id>
+```
+
+If source-level `max_upload_size` is omitted, use the top-level
+`server.http.max_upload_size` value.
+
+Server-level queue configuration is separate from per-pipeline source
+configuration. Per-pipeline fields may override staging path and upload size;
+bind address, queue size, concurrency, and retention are server-level settings.
+
+## Upload Model
+
+The uploaded request should contain one complete source bundle.
+
+Use archive upload rather than multipart fields for each file. `distributor`
+should extract the archive into a per-run staging directory under the pipeline
+staging path, then validate the extracted bundle.
+
+The first implementation supports:
+
+- uncompressed tar;
+- gzip-compressed tar.
+
+The server should accept uncompressed tar as `application/x-tar`. It should
+accept gzip-compressed tar as `application/gzip` or `application/x-gzip`.
+
+Archive extraction must be conservative:
+
+- reject absolute paths;
+- reject path traversal and backslash paths;
+- reject symlinks, hardlinks, devices, sockets, and other special entries;
+- require exactly one root-level `manifest.json`;
+- require all manifest-listed files to be present as regular files;
+- reject extra nested manifests;
+- enforce upload size limits before extraction;
+- enforce extracted size and file-count limits after extraction;
+- clean up failed extraction directories.
+
+After extraction, validation should use the existing source bundle validation
+contract. A bundle that fails manifest, path, size, digest, or file validation
+must not be published.
+
+## Async Run Flow
+
+HTTP upload processing should be asynchronous:
+
+1. Authenticate the request token and map it to exactly one `http_upload`
+   pipeline.
+2. Admit or reject the request according to queue capacity and per-pipeline
+   active-run rules.
+3. Create a run id and per-run staging directory.
+4. Record initial in-memory run status as `accepted`.
+5. Return an acceptance response without waiting for fan-out to complete.
+6. In a worker, extract the archive, validate the staged bundle, and run the
+   pipeline fan-out using the staged bundle as the effective source.
+7. Record final success or failure status.
+
+Initial acceptance response shape:
+
+```json
+{
+  "run_id": "weather-daily.20260603T120000Z.ab12cd34",
+  "status": "accepted"
+}
+```
+
+Run ids should be generated by `distributor`. Use the pipeline id, a
+filesystem-safe UTC timestamp, and a short random suffix to avoid collisions.
+
+Suggested status values:
+
+- `accepted`;
+- `queued`;
+- `running`;
+- `succeeded`;
+- `failed`;
+- `expired`.
+
+Status records should include run id, pipeline id, status, timestamps, and the
+completed run report or error details when available. Status records are kept
+in memory for the first implementation and expire after the configured
+retention period.
+
+Every accepted upload is treated as a new run. The first implementation does
+not accept producer-supplied idempotency keys.
+
+## Queue And Concurrency
+
+The server must not run more than one upload-triggered execution for the same
+pipeline at the same time.
+
+Use two controls:
+
+- a global worker limit, configured as `max_concurrency`;
+- a bounded admission queue, configured as `queue_size`;
+- a per-pipeline single-active-run guard.
+
+If a run for the same pipeline is already active, later accepted uploads for
+that pipeline should wait in queue rather than starting concurrently.
+
+If the queue is full, the upload request should fail before consuming and
+staging the request body. The server must not allow unbounded memory or disk
+growth.
+
+## Authentication
+
+Each `http_upload` pipeline has one static upload token.
+
+Upload requests send the token in an HTTP bearer header:
+
+```http
+Authorization: Bearer <token>
+```
+
+Authentication maps the incoming bearer token to exactly one configured
+pipeline. If no pipeline matches, the request fails. If more than one pipeline
+resolves to the same token, config validation should fail before the server
+starts.
+
+Secret values must never be logged, returned in responses, or included in run
+status.
+
+## HTTP Server Boundary
+
+Add a server mode rather than trying to make `run` poll an HTTP source.
+
+The likely CLI shape is:
+
+```sh
+distributor serve --config <path>
+```
+
+The HTTP API should default to a private bind address. Operators that need
+public access, TLS termination, or mTLS should place `distributor` behind a
+reverse proxy or private network boundary unless a later roadmap explicitly
+adds in-app TLS.
+
+The server should expose:
+
+- `POST /upload`;
+- `GET /runs/<run_id>`;
+- `GET /healthz`.
+
+`GET /healthz` should return success after the server has loaded and validated
+configuration and is ready to accept requests.
+
+## Relationship To Existing Architecture
+
+`http_upload` should reuse existing code paths after staging:
+
+- upload staging should produce a local staged bundle tree;
+- staged bundle validation should use the existing bundle validation contract;
+- fan-out should use the app-layer single-pipeline run behavior where possible;
+- destination handling should remain backend-agnostic;
+- publish, transform, link, state, and notification behavior should not know
+  that the source arrived over HTTP.
+
+If the current app-layer single-pipeline runner assumes it can open and walk the
+configured source backend, the HTTP implementation should add a narrow app-layer
+entry point for "run this pipeline using this already-staged source backend"
+rather than bending `http_upload` into a fake durable storage backend.
+
+## Non-Goals
+
+The first HTTP upload implementation should not add:
+
+- destination-side HTTP upload;
+- browser UI;
+- producer execution;
+- source manifest schema changes;
+- warning-only digest mismatch behavior;
+- durable database-backed queueing;
+- producer-supplied idempotency keys;
+- zstd archive support;
+- URL-token authentication;
+- public network exposure defaults;
+- in-app TLS;
+- general-purpose workflow orchestration.
+
+## Testing Expectations
+
+Suggested coverage:
+
+- config validation accepts `http_upload` sources and rejects `http_upload`
+  destinations;
+- default staging path becomes `/var/spool/distributor/<pipeline_id>`;
+- token env references resolve through real environment and `secrets.directory`;
+- duplicate token values across pipelines fail validation;
+- upload size limit defaults to 20 MB and is enforced;
+- archive extraction rejects unsafe paths, symlinks, hardlinks, devices,
+  missing manifest, missing manifest-listed files, and nested manifests;
+- valid uploaded bundles validate through the existing bundle contract;
+- upload admission returns `accepted` and a generated run id;
+- status lookup reports queued, running, succeeded, failed, and expired states;
+- per-pipeline runs do not execute concurrently;
+- global `max_concurrency` is honored;
+- full queues reject uploads before request-body staging;
+- `GET /runs/<run_id>` returns in-memory status records until retention expiry;
+- failed extraction and failed runs clean up or retain staging data according to
+  the configured retention policy;
+- secret values never appear in logs, responses, or status records;
+- normal local, SSH/SFTP, and S3 source behavior remains unchanged.
+
+## Documentation Updates After Implementation
+
+- Update `docs/config.md` with `http_upload` source fields and defaults.
+- Update `docs/cli.md` with `serve` syntax and HTTP behavior.
+- Update `docs/operations.md` with upload, queue, status, and staging
+  workflows.
+- Update `docs/troubleshooting.md` for authentication, archive extraction,
+  validation, queue, and fan-out failures.
+- Add examples only if they are secret-free and safe to run locally.
+- Update internal docs for any new app, HTTP, queue, or ingestion packages.
+
+Keep this roadmap under `docs/roadmap/` until implemented.
+
+## Future Work
+
+The first implementation intentionally defers:
+
+- URL-token authentication;
+- zstd-compressed archive support;
+- durable status persistence across process restarts;
+- database-backed queueing;
+- producer-supplied idempotency keys;
+- run listing, cancellation, or retry endpoints;
+- in-app TLS;
+- public network exposure defaults;
+- browser UI.
+
+These items should remain out of current-behavior docs until a later roadmap
+selects and specifies them.

docs/roadmap/implementation.md

@@ -1,102 +1,328 @@
-# Roadmap
+# HTTP Upload API Implementation 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 is the canonical staged implementation plan for
+`docs/roadmap/http.md`.
 
-`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.
+The current application supports CLI-driven distribution using configured
+local, SSH/SFTP, and S3-compatible source and destination backends. It does not
+yet implement an HTTP server, a `serve` command, `http_upload` source
+configuration, upload authentication, archive ingestion, async upload status,
+or HTTP routes.
 
-## Future Work
+Future, planned, or aspirational behavior belongs under `docs/roadmap/` until
+it is implemented. Current-behavior docs must be updated only after the
+corresponding stage is complete.
 
-These items are not implemented. They should not be documented as current
-behavior outside `docs/roadmap/` unless a future implementation adds them.
+## Implementation Principles
 
-### CLI And Status Output
+- Treat `docs/roadmap/http.md` as the source of truth for accepted HTTP upload
+  policy.
+- Prefer standard-library HTTP, tar, gzip, and sync primitives.
+- Do not add external dependencies for the first HTTP upload implementation.
+- Do not register `http_upload` as a durable storage backend.
+- Reuse existing bundle validation, destination fan-out, transform, link,
+  state, notification, and run-report behavior after upload staging.
+- Keep deferred items out of current-behavior docs: URL-token auth, zstd,
+  durable status, database queues, producer idempotency keys, run
+  cancellation/listing, in-app TLS, public exposure defaults, and browser UI.
 
-- 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.
+## Stage 1: HTTP Upload Configuration
 
-### Producer Workflows
+Goal: add config support for HTTP upload sources and server settings without
+adding HTTP runtime behavior.
 
-- 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.
+Implementation scope:
 
-### Publication And Transform Behavior
+- add top-level `server.http` config with defaults:
+  - `bind: 127.0.0.1:8080`;
+  - `staging_root: /var/spool/distributor`;
+  - `max_upload_size: 20MB`;
+  - `queue_size: 16`;
+  - `max_concurrency: 1`;
+  - `retention: 24h`;
+- add source-only `backend: http_upload`;
+- add `http_upload` source fields:
+  - required `token_env`;
+  - optional `staging_path`;
+  - optional `max_upload_size`;
+- default omitted `staging_path` to
+  `/var/spool/distributor/<pipeline_id>`;
+- default omitted source `max_upload_size` to
+  `server.http.max_upload_size`;
+- reject `http_upload` as a destination backend;
+- parse sizes with `B`, `KB`, `MB`, and `GB` suffixes using 1024 multipliers;
+- parse `retention` with `time.ParseDuration`;
+- keep literal upload tokens out of YAML.
 
-- 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.
+Documentation updates after implementation:
 
-### State And Compatibility
+- update `docs/config.md` for implemented config fields and defaults;
+- update `docs/internal/config.md` for config ownership and validation rules;
+- do not document HTTP routes yet.
 
-- 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.
+Tests:
 
-### Backends, Security, And Deployment
+- config loading accepts valid `server.http` and `http_upload` source config;
+- defaults apply for bind, staging root, staging path, upload size, queue size,
+  concurrency, and retention;
+- known-field decoding rejects unknown fields;
+- invalid sizes, invalid durations, missing token env, missing pipeline ids,
+  duplicate pipeline ids, and `http_upload` destinations fail validation;
+- existing local, SSH/SFTP, and S3 config behavior remains unchanged.
 
-- 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.
+Completion criteria: `go test ./internal/config` passes and no runtime code
+attempts to execute `http_upload`.
 
-## Roadmap Maintenance
+## Stage 2: Upload Archive Staging
 
-When adding future roadmap work:
+Goal: stage uploaded tar or tar.gz archives into a validated local source
+bundle tree.
 
-- 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.
+Implementation scope:
+
+- add an internal ingestion package for upload archive staging;
+- support uncompressed tar and gzip-compressed tar only;
+- accept content types:
+  - `application/x-tar`;
+  - `application/gzip`;
+  - `application/x-gzip`;
+- stream request bodies to a per-run archive or staging path while enforcing
+  max upload size;
+- extract into a per-run staging directory under the pipeline staging path;
+- reject absolute paths, path traversal, backslash paths, symlinks, hardlinks,
+  devices, sockets, and other special entries;
+- require exactly one root-level `manifest.json`;
+- reject nested manifests;
+- require all manifest-listed files to exist as regular files;
+- enforce extracted size and file-count limits;
+- clean up failed extraction directories;
+- validate staged bundles through the existing source bundle contract.
+
+Documentation updates after implementation:
+
+- update relevant `docs/internal/` docs for the new ingestion package;
+- keep user-facing HTTP docs out of current-behavior docs until the server
+  stage is implemented.
+
+Tests:
+
+- valid tar and tar.gz uploads stage successfully;
+- unsupported content types fail;
+- max upload size is enforced while streaming;
+- unsafe archive entries are rejected;
+- missing manifest, nested manifest, missing listed file, digest mismatch, and
+  non-regular manifest-listed files fail validation;
+- failed extraction cleans up staging data according to the package contract.
+
+Completion criteria: ingestion can produce a validated staged local bundle and
+does not publish anything.
+
+## Stage 3: Staged Source Pipeline Execution
+
+Goal: run one configured pipeline using an already-staged local source bundle
+root.
+
+Implementation scope:
+
+- add a narrow app-layer entry point for executing one pipeline with a staged
+  local source backend/root;
+- bypass normal source backend opening only for this staged-source entry point;
+- do not register `http_upload` as a normal `storage.Backend`;
+- reuse existing destination fan-out, transform, link, state, notification, and
+  run-report behavior;
+- ensure publish, transform, link, and state code do not know the source
+  arrived over HTTP.
+
+Documentation updates after implementation:
+
+- update `docs/internal/app.md` for the staged-source app entry point;
+- update `docs/internal/bundle.md` only if source validation behavior changes.
+
+Tests:
+
+- staged valid bundles publish through configured local destinations;
+- invalid staged bundles fail before destination writes;
+- configured destination behavior for local, SSH/SFTP, and S3 remains
+  backend-agnostic;
+- run reports match existing app report semantics.
+
+Completion criteria: app-level tests prove staged local bundles can run through
+normal fan-out without HTTP server code.
+
+## Stage 4: Async Upload Queue And Status
+
+Goal: add in-memory async upload coordination, queueing, and status tracking.
+
+Implementation scope:
+
+- add an in-memory upload coordinator;
+- generate run ids shaped like
+  `<pipeline_id>.<utc_timestamp>.<random_suffix>`;
+- use statuses:
+  - `accepted`;
+  - `queued`;
+  - `running`;
+  - `succeeded`;
+  - `failed`;
+  - `expired`;
+- enforce global `max_concurrency`;
+- enforce bounded `queue_size`;
+- enforce one active running upload per pipeline;
+- queue later accepted uploads for the same pipeline instead of running them
+  concurrently;
+- reject uploads before consuming or staging the request body when the queue is
+  full;
+- keep status in memory;
+- apply time-based retention to completed status and staged run directories;
+- treat every accepted upload as a new run;
+- do not support producer-supplied idempotency keys.
+
+Documentation updates after implementation:
+
+- update internal docs for the upload coordinator;
+- do not add user-facing HTTP docs until the server stage is implemented.
+
+Tests:
+
+- run id format includes pipeline id, filesystem-safe UTC timestamp, and random
+  suffix;
+- queue size is bounded;
+- full queues reject admission before staging;
+- same-pipeline uploads serialize;
+- different pipelines run concurrently up to `max_concurrency`;
+- status transitions cover accepted, queued, running, succeeded, failed, and
+  expired;
+- final status retains run report or error details until retention expiry;
+- staging directories are retained or cleaned according to retention policy.
+
+Completion criteria: coordinator tests pass without starting an HTTP server.
+
+## Stage 5: HTTP Server And `serve` CLI
+
+Goal: expose the upload coordinator through HTTP and add the server CLI command.
+
+Implementation scope:
+
+- add `distributor serve --config <path>`;
+- load config and `secrets.directory` before starting the server;
+- resolve each `http_upload` `token_env` through the config-owned environment
+  resolver;
+- fail startup if any configured token is missing or duplicated;
+- bind to `server.http.bind`, defaulting to `127.0.0.1:8080`;
+- implement `POST /upload`;
+- implement `GET /runs/<run_id>`;
+- implement `GET /healthz`;
+- authenticate uploads with `Authorization: Bearer <token>`;
+- map each token to exactly one configured `http_upload` pipeline;
+- do not require or accept a producer-submitted pipeline id;
+- return `202 Accepted` with:
+
+  ```json
+  {"run_id":"<id>","status":"accepted"}
+  ```
+
+- return `401` for missing or invalid bearer token;
+- return `413` for oversized upload;
+- return `415` for unsupported archive content type;
+- return `503` for full queue;
+- return `404` for unknown run status;
+- never log or return secret token values.
+
+Documentation updates after implementation:
+
+- update `docs/cli.md` with `serve` syntax;
+- update `docs/config.md` with HTTP upload source and server config;
+- update `docs/troubleshooting.md` for startup, auth, upload, and status
+  failures;
+- update relevant internal docs for HTTP server package boundaries.
+
+Tests:
+
+- CLI parsing recognizes `serve --config <path>`;
+- private bind default is applied;
+- startup fails for missing or duplicate tokens;
+- auth accepts valid bearer tokens and rejects missing or invalid tokens;
+- routes return the expected status codes and JSON response shapes;
+- health succeeds after configuration is loaded and the server is ready;
+- responses, logs, and status records do not expose secret token values.
+
+Completion criteria: `httptest` route tests and CLI tests pass, and no current
+local/SSH/S3 CLI behavior regresses.
+
+## Stage 6: End-To-End HTTP Upload Flow
+
+Goal: prove the complete async HTTP upload path publishes valid bundles and
+rejects invalid ones safely.
+
+Implementation scope:
+
+- add integration-style tests using `httptest`;
+- submit valid tar and tar.gz bundles;
+- poll `GET /runs/<run_id>` until completion;
+- verify fan-out reaches configured local destinations;
+- verify invalid archives fail without publishing;
+- verify same-pipeline uploads serialize;
+- verify different pipelines can run up to `max_concurrency`;
+- verify status includes completed run report or error details.
+
+Documentation updates after implementation:
+
+- update `docs/operations.md` with an HTTP upload workflow;
+- add safe local examples only if they are secret-free and testable.
+
+Tests:
+
+- valid upload returns `202 Accepted`, then `succeeded`;
+- invalid archive returns an accepted run only when admission succeeds, then
+  transitions to `failed`;
+- successful fan-out writes expected destination outputs and state;
+- failed upload does not write destination outputs;
+- same-pipeline and cross-pipeline concurrency follow configured policy;
+- run focused package tests and `go test ./...`.
+
+Completion criteria: end-to-end HTTP upload tests pass and full test suite
+passes.
+
+## Stage 7: Documentation And Roadmap Closeout
+
+Goal: document implemented HTTP upload behavior and remove completed roadmap
+drift.
+
+Implementation scope:
+
+- update current-behavior docs for implemented HTTP upload behavior:
+  - `docs/config.md`;
+  - `docs/cli.md`;
+  - `docs/operations.md`;
+  - `docs/troubleshooting.md`;
+  - relevant `docs/internal/` docs;
+- add only secret-free, safe local examples;
+- keep deferred items out of current docs;
+- remove or rewrite completed roadmap material once behavior is fully
+  documented.
+
+Deferred items:
+
+- URL-token authentication;
+- zstd archive support;
+- durable status persistence;
+- database-backed queues;
+- producer idempotency keys;
+- run cancellation or listing;
+- in-app TLS;
+- public exposure defaults;
+- browser UI.
+
+Tests:
+
+- documentation consistency checks show implemented HTTP behavior is no longer
+  described only as future work;
+- current-behavior docs do not describe deferred behavior as available;
+- examples are valid and secret-free;
+- run focused tests for docs/examples backed by tests and `go test ./...` if
+  examples or behavior docs changed with code.
+
+Completion criteria: current docs describe implemented behavior accurately, and
+`docs/roadmap/` contains only future or deferred HTTP work.

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