Clean up upload API documentation roadmap

docs/config.md

@@ -86,11 +86,15 @@ server:
     queue_size: 16
     max_concurrency: 1
     retention: 24h
+upload_tokens:
+  - id: weather-reporter
+    token_env: WEATHER_UPLOAD_TOKEN
+    allow_pipelines:
+      - weather-daily
 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:
@@ -99,9 +103,9 @@ pipelines:
         path: /srv/reports/archive
 ```
 
-`token_env` is required for `http_upload` sources. `staging_path` defaults to `<server.http.staging_root>/<pipeline id>`. `max_upload_size` defaults to `server.http.max_upload_size`.
+`upload_tokens` is required when any pipeline source uses `http_upload`. Each token record resolves its bearer token value from the process environment or `secrets.directory`. `allow_pipelines` lists configured upload pipeline ids that the token may submit to. One token may authorize multiple upload pipelines, and multiple tokens may authorize the same upload pipeline.
 
-`serve` maps each resolved bearer token to exactly one `http_upload` pipeline. Startup fails when a token is missing, empty, or duplicates another upload pipeline token.
+For `http_upload` sources, `staging_path` defaults to `<server.http.staging_root>/<pipeline id>`. `max_upload_size` defaults to `server.http.max_upload_size`.
 
 ## Top-Level Fields
 
@@ -124,6 +128,18 @@ Numeric server values and durations must be greater than zero after defaults are
 
 See [Secrets](#secrets) for resolution rules.
 
+### `upload_tokens`
+
+`upload_tokens` configures bearer tokens for `distributor serve`. It is required when any pipeline source backend is `http_upload` and is invalid when no upload pipelines are configured.
+
+Each token has:
+
+- `id`: required unique slug-like identifier for the token record. It must start with a letter or number and may contain letters, numbers, `.`, `_`, and `-`.
+- `token_env`: required environment variable or secret-file name containing the bearer token value.
+- `allow_pipelines`: required non-empty list of configured pipeline ids whose source backend is `http_upload`.
+
+Token values must resolve to non-empty strings and must be unique across token records. Every configured upload pipeline must be allowed by at least one token.
+
 ### `pipelines`
 
 `pipelines` is required and must contain at least one pipeline.
@@ -216,13 +232,11 @@ HTTP upload backends are valid only as pipeline sources and are served by `distr
 
 ```yaml
 backend: http_upload
-token_env: WEATHER_DAILY_UPLOAD_TOKEN
 staging_path: /var/spool/distributor/weather-daily
 max_upload_size: 20MB
 ```
 
 - `backend`: required value `http_upload`.
-- `token_env`: required environment variable or secret-file name containing the bearer token.
 - `staging_path`: optional staging path. Default: `<server.http.staging_root>/<pipeline id>`.
 - `max_upload_size`: optional per-source upload limit. Default: `server.http.max_upload_size`.
 
@@ -416,7 +430,7 @@ Fields resolved through this resolver:
 
 - `credentials.access_key_id_env`
 - `credentials.secret_access_key_env`
-- `source.token_env` for `http_upload` sources
+- `upload_tokens[].token_env`
 
 ## Maintained Examples
 

docs/consumers/api.md

@@ -9,14 +9,15 @@ This document is the copyable implementation guide for submitting producer outpu
 The upstream application needs these values from deployment or operator configuration:
 
 - distributor endpoint: the HTTP server base URL, such as `https://distributor.example.com`;
-- upload token: bearer token for exactly one configured `http_upload` pipeline;
+- upload token: bearer token that authenticates the producer;
+- pipeline id: configured `http_upload` pipeline that should process this upload;
 - generated files: regular local files to include in the source bundle;
 - bundle id: stable identifier for the logical report stream or artifact;
 - idempotency key: unique key for one producer run, reused only when retrying that same run.
 
 Do not put destination routing, public URLs, transform settings, or credentials in the source manifest. Those belong in the `distributor` pipeline configuration.
 
-The bundle id and idempotency key have different jobs. The bundle id tells `distributor` whether a new upload is a newer version of the same source; keep it stable across runs that should replace the same managed destination artifact. The idempotency key tells `distributor` whether an upload request is a retry; change it for each distinct producer run so new content is enqueued.
+The token, pipeline id, bundle id, and idempotency key have different jobs. The token authenticates the producer. The pipeline id selects the configured distributor workflow, including destinations and publishing policy. The bundle id tells `distributor` whether a new upload is a newer version of the same source; keep it stable across runs that should replace the same managed destination artifact. The idempotency key tells `distributor` whether an upload request is a retry; change it for each distinct producer run so new content is enqueued.
 
 ## Recommended Workflow
 
@@ -55,6 +56,7 @@ func SubmitReport(reportPath, summaryPath string) error {
 		return fmt.Errorf("distributor endpoint and token are required")
 	}
 
+	pipelineID := "weather-hourly"
 	reportID := "weather.hourly.brentwood"
 	runID := time.Now().UTC().Format("20060102T150405.000000000Z")
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
@@ -69,6 +71,7 @@ func SubmitReport(reportPath, summaryPath string) error {
 	}
 
 	result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{
+		PipelineID:     pipelineID,
 		ID:             reportID,
 		IdempotencyKey: reportID + "." + runID,
 		Files: []bundle.BundleFile{
@@ -92,6 +95,7 @@ func SubmitReport(reportPath, summaryPath string) error {
 ## Producer Responsibilities
 
 - Use a stable bundle id for the logical producer output that should replace the same destination artifact, such as `weather.hourly.brentwood`.
+- Set `PipelineID` to the configured upload pipeline that should process the bundle.
 - Do not include per-run timestamps, random values, or job ids in the bundle id unless each run should be treated as a different source.
 - Use an idempotency key that changes for every distinct producer run, such as `<bundle-id>.<run-id>`.
 - Reuse the same idempotency key only when retrying the exact same producer run with the same source manifest.
@@ -106,7 +110,7 @@ Valid bundle paths are relative slash paths. They must not be empty, absolute, c
 
 `pkg/upload` sends `Idempotency-Key` on every upload. If the caller omits one, the package generates a random key for that call and reuses it for in-process retries. That is enough for transient network retry within one process, but it does not give cross-process retry identity.
 
-For producer jobs that may retry after process restart, supply a key derived from the producer run, such as `<bundle-id>.<run-id>`. Reusing the same key with the same normalized source manifest returns the original accepted run. Reusing the same key with different source content returns a conflict. Reusing one key across multiple distinct report generations prevents those generations from being treated as new uploads.
+For producer jobs that may retry after process restart, supply a key derived from the producer run, such as `<bundle-id>.<run-id>`. Reusing the same key with the same token, pipeline id, and normalized source manifest returns the original accepted run. Reusing the same key with different source content in that scope returns a conflict. Reusing one key across multiple distinct report generations prevents those generations from being treated as new uploads.
 
 `Status` polls `/runs/<run-id>` while the distributor server retains the in-memory status record. Status values are `accepted`, `queued`, `running`, `succeeded`, and `failed`. Completed records expire according to the server's `server.http.retention` setting, and server restart clears status and idempotency records.
 

docs/consumers/pkg-upload.md

@@ -8,7 +8,7 @@ Import path:
 import "gitea.maximumdirect.net/eric/distributor/pkg/upload"
 ```
 
-`pkg/upload` is the producer-facing HTTP upload client. It builds on `pkg/bundle`, packages valid source bundles as gzip-compressed tar archives, sends bearer authentication, includes idempotency keys, and exposes a status polling helper.
+`pkg/upload` is the producer-facing HTTP upload client. It builds on `pkg/bundle`, packages valid source bundles as gzip-compressed tar archives, sends bearer authentication, routes uploads to a configured pipeline, includes idempotency keys, and exposes a status polling helper.
 
 `UploadFiles` examples also use:
 
@@ -30,7 +30,7 @@ if err != nil {
 }
 ```
 
-`Endpoint` is the distributor server base URL. The client derives `/upload` and `/runs/<run-id>`. `Token` is required and is sent as `Authorization: Bearer <token>`. Token values are redacted from client errors.
+`Endpoint` is the distributor server base URL. The client derives `/v1/pipelines/<pipeline-id>/upload` and `/runs/<run-id>`. `Token` is required and is sent as `Authorization: Bearer <token>`. Token values are redacted from client errors.
 
 `HTTPClient` and `Retry` are optional. Defaults use a 30 second HTTP timeout and safe retry settings.
 
@@ -40,6 +40,7 @@ Use `UploadFiles` when the producer has generated output files but has not assem
 
 ```go
 result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{
+	PipelineID:     "weather-hourly",
 	ID:             "weather.hourly.brentwood",
 	IdempotencyKey: "weather.hourly.brentwood.20260607T150000Z",
 	Files: []bundle.BundleFile{
@@ -53,7 +54,7 @@ if err != nil {
 _ = result.RunID
 ```
 
-`UploadFiles` creates a temporary bundle, writes and validates a manifest, uploads the archive, and removes temporary files when the call returns. It does not write into producer source directories.
+`PipelineID` is required and selects the configured distributor workflow for this upload. `ID` is the source manifest id and identifies the logical artifact inside that workflow. `UploadFiles` creates a temporary bundle, writes and validates a manifest, uploads the archive, and removes temporary files when the call returns. It does not write into producer source directories.
 
 ## Upload An Existing Bundle
 
@@ -61,6 +62,7 @@ Use `UploadBundle` when the producer already has a complete local bundle root co
 
 ```go
 result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{
+	PipelineID:     "weather-hourly",
 	Root:           "/var/spool/weather/hourly-2026-06-07T15",
 	IdempotencyKey: "weather.hourly.brentwood.20260607T150000Z",
 })
@@ -70,7 +72,7 @@ if err != nil {
 _ = result.RunID
 ```
 
-`UploadBundle` validates the local bundle by default and uploads only `manifest.json` plus manifest-listed files. Unlisted files are not uploaded.
+`PipelineID` is required for existing bundles too. `UploadBundle` validates the local bundle by default and uploads only `manifest.json` plus manifest-listed files. Unlisted files are not uploaded.
 
 ## Result And Status
 
@@ -96,7 +98,7 @@ Every upload request includes `Idempotency-Key`.
 
 If `IdempotencyKey` is omitted, the client generates a random 128-bit lowercase hexadecimal key for that upload operation and reuses it for retries within the same call. For cross-process retry safety, producers should pass a key derived from the producer run, such as `<bundle-id>.<run-id>`.
 
-Do not reuse the same idempotency key for multiple distinct report generations. Reuse it only when retrying the exact same run with the same source manifest. A repeated key with the same manifest returns the original accepted run instead of enqueueing another run; a repeated key with different content returns an idempotency conflict.
+Do not reuse the same idempotency key for multiple distinct report generations. Reuse it only when retrying the exact same run with the same token, pipeline id, and source manifest. A repeated key with the same manifest in that scope returns the original accepted run instead of enqueueing another run; a repeated key with different content returns an idempotency conflict.
 
 The client retries only safe cases:
 
@@ -104,7 +106,7 @@ The client retries only safe cases:
 - temporary network errors;
 - ambiguous mid-upload failures.
 
-It does not retry after `202 Accepted` and does not retry `400`, `401`, `409`, `413`, or `415`.
+It does not retry after `202 Accepted` and does not retry `400`, `401`, `403`, `404`, `409`, `413`, or `415`.
 
 Detect conflicting key reuse with `errors.As`:
 

docs/integrations/http-upload.md

@@ -2,7 +2,7 @@
 
 Audience: producers, operators, and maintainers integrating with `distributor serve`.
 
-`distributor serve` exposes a local HTTP upload API for pipelines whose source backend is `http_upload`. Each bearer token maps to exactly one configured pipeline.
+`distributor serve` exposes a local HTTP upload API for pipelines whose source backend is `http_upload`. Bearer tokens authenticate producers, and the upload path selects the configured pipeline. The selected token must be allowed for the requested pipeline.
 
 ## Authentication
 
@@ -12,9 +12,9 @@ Uploads authenticate with:
 Authorization: Bearer <token>
 ```
 
-Token values are resolved from the configured `source.token_env` through the process environment or `secrets.directory`. Tokens are not configured as YAML literal values.
+Token values are resolved from top-level `upload_tokens` records through the process environment or `secrets.directory`. Tokens are not configured as YAML literal values.
 
-Requests that include `pipeline` or `pipeline_id` query parameters are rejected. The bearer token selects the pipeline.
+Requests that include `pipeline` or `pipeline_id` query parameters are rejected. Use the pipeline id in the upload path.
 
 ## Endpoints
 
@@ -26,7 +26,7 @@ Returns `200 OK` when the server is running:
 {"status":"ok"}
 ```
 
-### `POST /upload`
+### `POST /v1/pipelines/{pipeline_id}/upload`
 
 Accepts one source bundle archive and returns after the archive is staged and validated.
 
@@ -36,7 +36,7 @@ Producers may include:
 Idempotency-Key: <key>
 ```
 
-Idempotency keys are scoped to the authenticated pipeline selected by the bearer token. Valid keys are non-empty ASCII strings up to 128 bytes using letters, digits, `.`, `_`, `-`, and `:`. Invalid keys return `400`.
+`pipeline_id` must name a configured pipeline whose source backend is `http_upload`, and the authenticated token must allow that pipeline. Idempotency keys are scoped to token id, pipeline id, and key. Valid keys are non-empty ASCII strings up to 128 bytes using letters, digits, `.`, `_`, `-`, and `:`. Invalid keys return `400`.
 
 Accepted content types:
 
@@ -54,6 +54,8 @@ Common error responses:
 
 - `400`: pipeline query supplied, invalid idempotency key, archive rejected, malformed archive, or invalid staged source bundle.
 - `401`: missing, empty, or unknown bearer token.
+- `403`: bearer token is valid but is not allowed for the requested pipeline.
+- `404`: upload path is unknown or the requested upload pipeline is not configured.
 - `409`: repeated idempotency key conflicts with another source manifest, or the same key is already being staged.
 - `413`: upload body exceeds the selected pipeline size limit.
 - `415`: unsupported content type.
@@ -71,7 +73,7 @@ Retryable idempotency conflicts include:
 {"error":"upload idempotency key is already being processed","retryable":true}
 ```
 
-When `Idempotency-Key` is omitted, upload admission preserves the raw HTTP behavior: every valid accepted upload receives its own run id. When a key is supplied, the server records the accepted run after archive staging and source bundle validation succeed. Reusing the same key for the same authenticated pipeline and the same normalized source manifest returns the original `202 Accepted` response and does not enqueue another run. Reusing the same key for a different normalized source manifest returns `409 Conflict`. Producers should use a fresh key for each distinct producer run and reuse a key only for retries of that same run.
+When `Idempotency-Key` is omitted, upload admission preserves the raw HTTP behavior: every valid accepted upload receives its own run id. When a key is supplied, the server records the accepted run after archive staging and source bundle validation succeed. Reusing the same key for the same token id, pipeline id, and normalized source manifest returns the original `202 Accepted` response and does not enqueue another run. Reusing the same key for a different normalized source manifest within that scope returns `409 Conflict`. Producers should use a fresh key for each distinct producer run and reuse a key only for retries of that same run.
 
 ### `GET /runs/<run-id>`
 
@@ -119,14 +121,15 @@ if err != nil {
 	return err
 }
 result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{
+	PipelineID:     "reports",
 	Root:           "examples/source-bundle",
 	IdempotencyKey: "reports.example.20260604T120000Z",
 })
 ```
 
-`Endpoint` is the server base URL; the package derives `/upload` and `/runs/<run-id>`. `UploadBundle` validates a local bundle by default and uploads only `manifest.json` plus manifest-listed files. `UploadFiles` creates a temporary bundle from explicit `bundle.BundleFile` values before uploading. When `IdempotencyKey` is omitted, the package generates one random 128-bit lowercase hex key for the upload operation and reuses it across retries.
+`Endpoint` is the server base URL; the package derives `/v1/pipelines/<pipeline-id>/upload` and `/runs/<run-id>`. `PipelineID` is required and selects the configured distributor workflow. `UploadBundle` validates a local bundle by default and uploads only `manifest.json` plus manifest-listed files. `UploadFiles` creates a temporary bundle from explicit `bundle.BundleFile` values before uploading. When `IdempotencyKey` is omitted, the package generates one random 128-bit lowercase hex key for the upload operation and reuses it across retries.
 
-The helper retries only safe cases: `503 Service Unavailable`, temporary network errors, and ambiguous mid-upload failures. It does not retry after `202 Accepted` and does not retry `400`, `401`, `409`, `413`, or `415`. Bearer token values are redacted from returned errors.
+The helper retries only safe cases: `503 Service Unavailable`, temporary network errors, and ambiguous mid-upload failures. It does not retry after `202 Accepted` and does not retry `400`, `401`, `403`, `404`, `409`, `413`, or `415`. Bearer token values are redacted from returned errors.
 
 ## Queue And Retention
 
@@ -138,7 +141,7 @@ Idempotency records are memory-only, expire with completed upload status records
 
 ## Boundaries
 
-The HTTP API does not expose pipeline selection by request parameter, TLS, public routing policy, or durable status storage. Put public access controls, TLS termination, and rate limiting in deployment infrastructure.
+The HTTP API does not expose pipeline selection by query parameter, TLS, public routing policy, or durable status storage. Put public access controls, TLS termination, and rate limiting in deployment infrastructure.
 
 ## Tests
 

docs/internal/app.md

@@ -36,7 +36,7 @@ Run workflows discover and validate source bundles through `internal/bundle`. De
 
 HTTP uploads stage and validate archives before enqueueing a pipeline run with a local staged source root. Go producers can use the public `pkg/upload` package to create client-side gzip tar uploads for this server contract; `internal/app` remains the server-side orchestration boundary and does not import that producer package.
 
-Upload idempotency is owned by the upload coordinator. Optional `Idempotency-Key` values are scoped to the authenticated pipeline. The coordinator reserves a key while staging is in progress, records the accepted run id with the validated source manifest identity after staging succeeds, returns the original accepted record for the same key and same manifest, and rejects the same key with a different manifest as a conflict.
+Upload idempotency is owned by the upload coordinator. Optional `Idempotency-Key` values are scoped to token id, pipeline id, and key. The coordinator reserves a key while staging is in progress, records the accepted run id with the validated source manifest identity after staging succeeds, returns the original accepted record for the same scoped key and same manifest, and rejects the same scoped key with a different manifest as a conflict.
 
 ## Skip And Resume Behavior
 

docs/operations.md

@@ -114,7 +114,7 @@ Forced replacement deletes the current destination bundle path before writing ou
 
 ## HTTP Upload Operation
 
-The [HTTP Upload API Contract](integrations/http-upload.md) defines request and response details. `distributor serve` runs the HTTP upload API for pipelines whose source backend is `http_upload`. Each bearer token maps to exactly one configured upload pipeline. Token values come from the process environment or `secrets.directory`, not from YAML literal values.
+The [HTTP Upload API Contract](integrations/http-upload.md) defines request and response details. `distributor serve` runs the HTTP upload API for pipelines whose source backend is `http_upload`. Top-level `upload_tokens` authenticate producers and allow one or more upload pipelines. Token values come from the process environment or `secrets.directory`, not from YAML literal values.
 
 Start the maintained local example:
 
@@ -132,7 +132,7 @@ curl http://127.0.0.1:8080/healthz
 Upload one tar or tar.gz source bundle archive:
 
 ```sh
-curl -X POST http://127.0.0.1:8080/upload \
+curl -X POST http://127.0.0.1:8080/v1/pipelines/example-http-upload/upload \
   -H "Authorization: Bearer $DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN" \
   -H "Content-Type: application/gzip" \
   --data-binary @bundle.tar.gz
@@ -141,7 +141,7 @@ curl -X POST http://127.0.0.1:8080/upload \
 For safe producer retries, include an idempotency key that is stable for the same producer run and different for each distinct run:
 
 ```sh
-curl -X POST http://127.0.0.1:8080/upload \
+curl -X POST http://127.0.0.1:8080/v1/pipelines/example-http-upload/upload \
   -H "Authorization: Bearer $DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN" \
   -H "Content-Type: application/gzip" \
   -H "Idempotency-Key: producer.run.20260604T120000Z" \
@@ -150,7 +150,7 @@ curl -X POST http://127.0.0.1:8080/upload \
 
 Go producer applications can use `pkg/upload` instead of constructing archives and HTTP requests directly. See [Upstream Producer Integration](consumers/api.md) for the copyable producer implementation guide.
 
-The maintained example client uses the local upload server and reads the token from `DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN`. It generates an idempotency key by default; set `DISTRIBUTOR_EXAMPLE_UPLOAD_IDEMPOTENCY_KEY` when retrying the same producer run across separate process runs.
+The maintained example client uses the local upload server, reads the token from `DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN`, and defaults the pipeline id to `example-http-upload`. Set `DISTRIBUTOR_EXAMPLE_UPLOAD_PIPELINE_ID` or pass a second argument to use another configured upload pipeline. It generates an idempotency key by default; set `DISTRIBUTOR_EXAMPLE_UPLOAD_IDEMPOTENCY_KEY` when retrying the same producer run across separate process runs.
 
 ```sh
 go run ./examples/upload-client
@@ -172,7 +172,7 @@ Status values are `accepted`, `queued`, `running`, `succeeded`, and `failed`. Co
 
 Upload admission is bounded by `server.http.queue_size`. Publication concurrency is bounded by `server.http.max_concurrency`, and the coordinator does not run two uploads for the same pipeline at the same time.
 
-`Idempotency-Key` is optional for raw HTTP clients. When present, it is scoped to the authenticated pipeline. Reusing the same key with the same normalized source manifest returns the original accepted run response and does not enqueue another run. Reusing the key with a different source manifest returns `409 Conflict`. If another request with the same key is still being staged before its manifest is known, the server returns a retryable `409 Conflict`. Idempotency records are memory-only and expire with completed upload status records.
+`Idempotency-Key` is optional for raw HTTP clients. When present, it is scoped to the token id, pipeline id, and key. Reusing the same key with the same normalized source manifest in that scope returns the original accepted run response and does not enqueue another run. Reusing the key with a different source manifest returns `409 Conflict`. If another request with the same key is still being staged before its manifest is known, the server returns a retryable `409 Conflict`. Idempotency records are memory-only and expire with completed upload status records.
 
 The upload server accepts `application/x-tar`, `application/gzip`, and `application/x-gzip`. Archives are extracted into a temporary staging directory, must contain exactly one root-level `manifest.json`, and must validate as one complete source bundle before a run id is issued. Per-source `max_upload_size` bounds both uploaded archive size and extracted bundle size. The implementation also caps extracted file count.
 

docs/roadmap/api.md

@@ -1,172 +1,25 @@
 # API Roadmap
 
-This document records planned API work that is not part of the current
+This document records API work that is not part of the current implementation. Current HTTP upload behavior is documented in `docs/integrations/http-upload.md`, current configuration behavior is documented in `docs/config.md`, and current producer package usage is documented under `docs/consumers/`.
-implementation. Current HTTP upload behavior is documented in
-`docs/integrations/http-upload.md` and current producer package usage is
-documented under `docs/consumers/`.
 
-## Pipeline-Scoped HTTP Upload API
+## Deferred Upload API Work
 
-Current `http_upload` behavior uses one bearer token to both authenticate a
+- Durable upload status storage.
-producer and select exactly one pipeline. That is simple, but it does not scale
+- Durable idempotency records across server restarts.
-well for producer applications that generate multiple report types on different
+- Run listing, retry, and cancellation endpoints.
-schedules.
+- Long-polling or wait-for-completion workflows.
-
+- Multipart, resumable, or streaming upload protocols.
-Planned work:
+- Additional archive content negotiation beyond tar and gzip-compressed tar.
-
+- URL-token authentication for constrained clients.
-- Separate upload authentication from pipeline routing.
+- Upload token lifecycle tooling.
-- Add token records that can authorize one producer/client for one or more
+- Mutual TLS or other in-app identity mechanisms.
-  upload pipelines.
+- In-app TLS termination.
-- Add a pipeline-scoped upload endpoint:
+- In-app public exposure policy.
-
+- In-app upload rate limiting.
-```text
-POST /v1/pipelines/{pipeline_id}/upload
-```
-
-- Keep the source manifest free of routing, destination, transform, and
-  credential data.
-- Keep `http_upload` source-only. A selected pipeline still owns destination
-  configuration, transforms, links, transfer policy, and publication behavior.
-
-## Proposed Configuration Shape
-
-Move bearer token configuration out of individual pipeline sources and into a
-top-level upload token list:
-
-```yaml
-upload_tokens:
-  - id: weatherreporter-prod
-    token_env: WEATHERREPORTER_UPLOAD_TOKEN
-    allow_pipelines:
-      - weather.morning
-      - weather.weekend
-      - weather.next_6_hours
-      - weather.storm
-      - weather.event
-```
-
-Pipeline sources would continue to use `http_upload`, but would no longer need
-one unique token per pipeline:
-
-```yaml
-pipelines:
-  - id: weather.morning
-    source:
-      backend: http_upload
-    destinations:
-      - id: archive
-        backend: s3
-        bucket: reports
-        prefix: weather/morning/archive
-      - id: latest
-        backend: s3
-        bucket: reports
-        prefix: weather/morning/latest
-        path_mapping:
-          mode: fixed
-```
-
-Per-pipeline upload settings such as `staging_path` and `max_upload_size` should
-remain on the `http_upload` source.
-
-## Authorization Semantics
-
-- Missing, malformed, or unknown bearer tokens should return `401 Unauthorized`.
-- Valid tokens that are not allowed for the requested pipeline should return
-  `403 Forbidden`.
-- Requested pipeline ids must name configured pipelines whose source backend is
-  `http_upload`.
-- Multiple upload tokens may authorize the same pipeline.
-- One upload token may authorize multiple pipelines.
-- Token values must continue to resolve through the process environment or
-  `secrets.directory`, not YAML literal values.
-
-Idempotency records should be scoped by token id, pipeline id, and idempotency
-key. This avoids collisions when multiple authorized producers submit to the
-same pipeline.
-
-## Producer Package Changes
-
-Add `PipelineID` to producer upload options:
-
-```go
-result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{
-	PipelineID:     "weather.morning",
-	ID:             "weather.morning.brentwood",
-	IdempotencyKey: "weather.morning.brentwood.20260607T050000Z",
-	Files: []bundle.BundleFile{
-		{SourcePath: reportPath, Path: "report.md"},
-		{SourcePath: dataPath, Path: "data.json"},
-	},
-})
-```
-
-`pkg/upload` should derive `/v1/pipelines/{pipeline_id}/upload` when
-`PipelineID` is set. Status lookup can continue to use run ids returned by the
-server.
-
-The producer contract should remain:
-
-- token identifies and authenticates the producer/client;
-- `PipelineID` selects the configured distributor workflow;
-- source manifest `id` identifies the logical artifact within that workflow;
-- idempotency key identifies one producer run and retry group.
-
-## Compatibility Plan
-
-Prefer a transition period:
-
-- Keep current `POST /upload` behavior for legacy configs where one token maps
-  to exactly one `http_upload` pipeline.
-- Reject legacy `/upload` routing when a token is authorized for multiple
-  pipelines, because routing would be ambiguous.
-- Keep rejecting `pipeline` and `pipeline_id` query parameters.
-- Document `/v1/pipelines/{pipeline_id}/upload` as the preferred endpoint for
-  new clients.
-
-After the transition period, consider deprecating or removing legacy `/upload`
-if the compatibility burden is no longer useful.
-
-## Implementation Work
-
-- Add `upload_tokens` config structs, defaults, validation, and secret
-  resolution.
-- Update upload token resolution to produce token identities and pipeline
-  allowlists instead of a token-to-single-pipeline map.
-- Add the `/v1/pipelines/{pipeline_id}/upload` HTTP route and pipeline id path
-  validation.
-- Preserve `/healthz` and `/runs/<run-id>` behavior.
-- Pass token identity into upload admission so idempotency can be scoped by
-  token id, pipeline id, and key.
-- Add `PipelineID` to `pkg/upload` upload option structs and endpoint
-  construction.
-- Update configuration, operation, integration, consumer, and troubleshooting
-  docs for implemented behavior.
-
-## Tests
-
-Important tests:
-
-- Config loading and validation for `upload_tokens`.
-- Startup failure for missing, empty, duplicated, or invalid upload token
-  records.
-- `401` for missing or unknown bearer token.
-- `403` for valid token not allowed for requested pipeline.
-- Successful upload to two different pipelines with one token.
-- Successful upload to one pipeline from two different authorized tokens.
-- Idempotency isolation across token ids and pipeline ids.
-- Legacy `/upload` compatibility for one-token-one-pipeline routing.
-- Legacy `/upload` rejection when routing is ambiguous.
-- `pkg/upload` endpoint construction with `PipelineID`.
-- Producer package tests for missing or invalid `PipelineID`.
 
 ## Boundaries
 
-- Do not add destination selection to producer manifests.
+- Producers do not choose destination ids, destination paths, transforms, links, publish policy, transfer policy, or storage backends through upload requests.
-- Do not let producers specify destination ids, transforms, links, publish
+- Source manifests remain free of routing, destination, transform, and credential data.
-  policy, or transfer policy through the upload API.
+- `http_upload` remains source-only unless a future design changes that contract.
-- Do not add durable upload status or durable idempotency as part of this work;
+- Public access policy, TLS termination, and rate limiting belong in deployment infrastructure unless a future design changes that boundary.
-  those remain separate roadmap items.
-- Do not add in-app public exposure policy, TLS, or rate limiting as part of
-  this work; those remain deployment-layer concerns unless a future
-  implementation changes that boundary.

docs/roadmap/implementation.md

@@ -1,275 +1,22 @@
-# Pipeline-Scoped Upload API Implementation Plan
+# Upload API Implementation Notes
 
-This roadmap is for an LLM coding agent implementing the planned API work in
+This file has no active implementation tasks for the pipeline-scoped upload API.
-`docs/roadmap/api.md`. It describes future work only. Do not update
-current-behavior docs outside `docs/roadmap/` until the corresponding stage is
-implemented.
 
-Before implementation, read:
+Current behavior is documented in:
 
-- `docs/policy/architecture.md`
+- `docs/config.md`
-- `docs/policy/development.md`
+- `docs/integrations/http-upload.md`
-- `docs/policy/documentation.md`
+- `docs/consumers/api.md`
-- `docs/roadmap/api.md`
+- `docs/consumers/pkg-upload.md`
+- `docs/operations.md`
+- `docs/troubleshooting.md`
 
-## Target Behavior
+Deferred API work is tracked in `docs/roadmap/api.md` and broader deferred work is tracked in `docs/roadmap/future.md`.
 
-The final implementation is a breaking v1 HTTP upload API change:
+## Boundaries
 
-- Upload authentication is configured with top-level `upload_tokens`.
+- Keep producer routing, destination selection, transform policy, publish policy, transfer policy, and backend credentials out of source manifests.
-- Pipeline routing is selected by `POST /v1/pipelines/{pipeline_id}/upload`.
+- Keep `pkg/upload` focused on producer-side bundle submission and status polling.
-- Producers must set `PipelineID` in `pkg/upload` upload options.
+- Keep server-side upload authentication, authorization, queueing, status, and publish orchestration in `internal/app`.
-- Legacy per-source `source.token_env` is removed.
+- Keep archive extraction and staged bundle validation in `internal/ingest`.
-- Legacy `POST /upload` no longer accepts uploads.
+- Keep durable status, durable idempotency, retry endpoints, cancellation endpoints, and wait helpers out of the current implementation until a new roadmap item defines them.
-- Idempotency is scoped by token id, pipeline id, and idempotency key.
-- `/healthz` and `/runs/<run-id>` continue to work.
-
-The producer contract is:
-
-- token authenticates the producer/client;
-- `PipelineID` selects the configured distributor workflow;
-- source manifest `id` identifies the logical artifact within that workflow;
-- idempotency key identifies one producer run and retry group.
-
-## Stage 1: Config Model And Validation
-
-Goal: make the YAML schema express upload authentication separately from
-pipeline source configuration.
-
-Implementation:
-
-- Add `UploadTokens []UploadToken` to `internal/config.Config` with YAML key
-  `upload_tokens`.
-- Add `UploadToken` with fields:
-  - `ID string` as `id`;
-  - `TokenEnv string` as `token_env`;
-  - `AllowPipelines []string` as `allow_pipelines`.
-- Remove `TokenEnv` from `HTTPUpload`; keep `StagingPath` and
-  `MaxUploadSize`.
-- Keep `http_upload` source defaults for `staging_path` and `max_upload_size`.
-- Update validation:
-  - `upload_tokens` is required when any pipeline uses
-    `source.backend: http_upload`.
-  - `upload_tokens` is invalid when it references no configured upload
-    pipelines.
-  - token `id` is required, slug-like, and unique.
-  - token `token_env` is required.
-  - token `allow_pipelines` is required.
-  - each token's `allow_pipelines` entries are unique.
-  - each allowed pipeline id exists and names a pipeline whose source backend
-    is `http_upload`.
-  - every configured `http_upload` pipeline is allowed by at least one token.
-  - `http_upload` sources no longer require `token_env`.
-- Remove tests and fixtures that expect `source.token_env`.
-- Add config load/validation tests for valid multi-pipeline tokens, multiple
-  tokens for one pipeline, missing token list, duplicate token ids, duplicate
-  allowlist entries, unknown allowed pipeline id, non-upload allowed pipeline
-  id, and upload pipeline not allowed by any token.
-
-Acceptance:
-
-- `go test ./internal/config` passes.
-- YAML containing `source.token_env` fails as an unknown field.
-- YAML using top-level `upload_tokens` and `http_upload` sources without
-  `source.token_env` loads and validates.
-
-## Stage 2: Upload Token Resolution And HTTP Routing
-
-Goal: authenticate by bearer token, authorize by token allowlist, and route by
-URL pipeline id.
-
-Implementation:
-
-- Replace the current token-to-single-pipeline map with resolved upload token
-  records containing:
-  - token id;
-  - resolved token value;
-  - allowed pipeline id set.
-- Resolve token values through the existing config environment resolver so
-  process environment and `secrets.directory` behavior remains consistent.
-- Startup must fail when a token env is missing, empty, or resolves to the same
-  token value as another upload token. Error messages may name token ids and
-  env var names, but must not print token values.
-- Change `uploadHTTPHandler` routes:
-  - keep `GET /healthz`;
-  - keep `GET /runs/<run-id>`;
-  - add `POST /v1/pipelines/{pipeline_id}/upload`;
-  - remove legacy `POST /upload`.
-- Path parsing rules:
-  - match exactly `/v1/pipelines/<pipeline-id>/upload`;
-  - reject missing pipeline id, extra path segments, and query-based
-    `pipeline` or `pipeline_id` routing;
-  - validate requested pipeline id using the same slug-like id policy used for
-    configured pipeline ids.
-- Request handling rules:
-  - missing, malformed, or unknown bearer token returns `401`;
-  - valid token not allowed for requested pipeline returns `403`;
-  - requested pipeline must be configured with `source.backend: http_upload`;
-  - content type and idempotency key validation remain unchanged;
-  - successful requests submit the requested pipeline id and token id to the
-    upload coordinator.
-- Update HTTP handler tests for accepted v1 upload, unauthorized upload,
-  forbidden pipeline, invalid pipeline path, removed `/upload`, invalid
-  content type, invalid idempotency key, and no token leakage.
-
-Acceptance:
-
-- `go test ./internal/app` passes for handler tests touched in this stage.
-- `POST /upload` returns not found or another non-accepting error and does not
-  call `Submit`.
-- `POST /v1/pipelines/{pipeline_id}/upload` routes only when token
-  authorization allows that pipeline.
-
-## Stage 3: Coordinator Idempotency Scope
-
-Goal: prevent idempotency collisions between distinct authorized clients and
-between pipelines.
-
-Implementation:
-
-- Add `TokenID string` to `UploadRequest`.
-- Update idempotency scope to include token id, pipeline id, and key.
-- Preserve existing behavior inside one idempotency scope:
-  - same key and same normalized manifest returns the original accepted run;
-  - same key and different normalized manifest returns conflict;
-  - same key while staging returns retryable conflict.
-- Keep requests without idempotency keys unscoped and always admitted according
-  to queue capacity.
-- Update coordinator tests:
-  - same token, same pipeline, same key returns original run;
-  - same token, same pipeline, same key with changed manifest conflicts;
-  - different token ids can use the same key for the same pipeline without
-    collision;
-  - same token id can use the same key for different pipelines without
-    collision;
-  - pending-key retryable conflict still applies only within the same token and
-    pipeline scope.
-
-Acceptance:
-
-- `go test ./internal/app` passes.
-- Existing queueing, retention, status, and same-pipeline serialization behavior
-  remains unchanged.
-
-## Stage 4: Public `pkg/upload` API
-
-Goal: make producer clients route uploads through the v1 pipeline-scoped API.
-
-Implementation:
-
-- Add `PipelineID string` to `UploadFilesOptions`.
-- Add `PipelineID string` to `UploadBundleOptions`.
-- Require non-empty valid `PipelineID` in both upload methods before local
-  bundle staging, validation, archiving, or HTTP requests.
-- Use the same accepted pipeline id syntax as server config ids: starts with an
-  ASCII letter or digit and then contains ASCII letters, digits, `.`, `_`, or
-  `-`.
-- Build upload URLs as
-  `/v1/pipelines/{pipeline_id}/upload`.
-- Keep `Status(ctx, runID)` unchanged.
-- Update package tests:
-  - missing `PipelineID` fails before local file work or HTTP request;
-  - invalid `PipelineID` fails before HTTP request;
-  - `UploadFiles` and `UploadBundle` post to the v1 route;
-  - retry behavior reuses the same idempotency key and v1 route;
-  - token redaction still works.
-- Update `pkg/upload` package documentation to explain the four-part producer
-  contract: token, `PipelineID`, manifest `ID`, idempotency key.
-
-Acceptance:
-
-- `go test ./pkg/upload ./pkg/bundle` passes.
-- Existing `Status` tests pass without endpoint changes.
-
-## Stage 5: Examples And End-To-End App Coverage
-
-Goal: prove the new server and client API work together across realistic upload
-pipelines.
-
-Implementation:
-
-- Update `examples/http-upload-local.yml` to use top-level `upload_tokens`.
-- Update `examples/upload-client` to require or default a pipeline id and pass
-  it to `pkg/upload`.
-- Update app integration tests:
-  - one token authorized for two upload pipelines can upload to both;
-  - two tokens authorized for one upload pipeline can upload to that pipeline;
-  - a valid token rejected for a disallowed pipeline returns `403`;
-  - removed `/upload` endpoint does not enqueue work;
-  - full upload publishes through the selected pipeline and records the
-    selected pipeline id in status/report output.
-- Update helper config builders in tests to use `upload_tokens`.
-
-Acceptance:
-
-- `go test ./internal/app` passes.
-- Example config loading tests pass.
-- `go run ./examples/upload-client` remains documented as an example that
-  targets a configured pipeline.
-
-## Stage 6: Implemented-Behavior Documentation
-
-Goal: move the feature from roadmap-only language into current-behavior docs
-after the code is implemented.
-
-Implementation:
-
-- Update `docs/config.md`:
-  - document top-level `upload_tokens`;
-  - remove `source.token_env`;
-  - document `http_upload` source fields `staging_path` and
-    `max_upload_size`;
-  - document token allowlist semantics.
-- Update `docs/integrations/http-upload.md`:
-  - replace `POST /upload` with
-    `POST /v1/pipelines/{pipeline_id}/upload`;
-  - document `401` versus `403`;
-  - document idempotency scope by token id and pipeline id.
-- Update `docs/consumers/api.md` and `docs/consumers/pkg-upload.md`:
-  - show `PipelineID` in `UploadFiles` and `UploadBundle`;
-  - explain token authentication versus pipeline routing.
-- Update `docs/operations.md` and `docs/troubleshooting.md` for the new curl
-  path, upload token config, and forbidden-pipeline diagnosis.
-- Update `README.md` only if its producer integration links or summary become
-  stale.
-- Revise `docs/roadmap/api.md` so it no longer presents implemented behavior as
-  future work. Either mark the pipeline-scoped upload API as implemented and
-  leave only deferred ideas, or remove implemented sections.
-
-Acceptance:
-
-- `rg -n "POST /upload|source.token_env|token_env.*http_upload|maps each resolved bearer token to exactly one" README.md docs examples`
-  returns no stale current-behavior references outside historical roadmap
-  context.
-- Documentation outside `docs/roadmap/` describes only implemented behavior.
-
-## Final Verification
-
-Run these commands after all stages are implemented:
-
-```sh
-go test ./internal/config
-go test ./internal/app
-go test ./pkg/bundle ./pkg/upload
-go test ./...
-```
-
-Also run:
-
-```sh
-rg -n "POST /upload|source.token_env|/v1/pipelines|upload_tokens|PipelineID" README.md docs examples internal pkg
-```
-
-Review the output for stale references, missing docs, and tests that still
-expect legacy upload routing.
-
-## Non-Goals
-
-- Do not let producers specify destination ids, destination paths, transforms,
-  links, publish policy, transfer policy, or storage backends through upload
-  requests.
-- Do not add durable upload status, durable idempotency, retry endpoints,
-  cancellation endpoints, long polling, or `UploadAndWait`.
-- Do not add in-app TLS, public exposure policy, or rate limiting.
-- Do not introduce new public package families beyond existing `pkg/bundle` and
-  `pkg/upload`.

docs/troubleshooting.md

@@ -400,7 +400,7 @@ Reference: [Configuration](config.md#serverhttp).
 
 Symptom: `upload token environment variable ... is not set`, `... is empty`, or `upload token environment variables ... resolve to the same value`.
 
-Likely cause: an `http_upload` source references a missing/empty `token_env`, or two upload pipelines resolve to the same bearer token.
+Likely cause: a top-level upload token record references a missing or empty `token_env`, or two token records resolve to the same bearer token.
 
 Diagnostic:
 
@@ -410,20 +410,20 @@ env | cut -d= -f1 | rg '^<token-variable>$'
 ls -l <secrets-directory>/<token-variable>
 ```
 
-Safe fix: provide one distinct non-empty token value per upload pipeline through the process environment or `secrets.directory`. Do not put literal tokens in YAML.
+Safe fix: provide one distinct non-empty token value per upload token record through the process environment or `secrets.directory`. Do not put literal tokens in YAML.
 
-Reference: [Configuration](config.md#http-upload-source-backend).
+Reference: [Configuration](config.md#upload_tokens).
 
 ## Upload Request Is Unauthorized
 
-Symptom: `POST /upload` returns `401`.
+Symptom: `POST /v1/pipelines/<pipeline-id>/upload` returns `401`.
 
-Likely cause: the request lacks `Authorization: Bearer <token>`, has an empty token, or uses a token that does not match any configured upload pipeline.
+Likely cause: the request lacks `Authorization: Bearer <token>`, has an empty token, or uses a token that does not match any configured upload token record.
 
 Diagnostic:
 
 ```sh
-curl -i -X POST http://127.0.0.1:8080/upload \
+curl -i -X POST http://127.0.0.1:8080/v1/pipelines/<pipeline-id>/upload \
   -H "Authorization: Bearer $DISTRIBUTOR_UPLOAD_TOKEN" \
   -H "Content-Type: application/x-tar" \
   --data-binary @bundle.tar
@@ -433,11 +433,27 @@ Safe fix: use the token value resolved by the configured `token_env`. Do not inc
 
 Reference: [Operations](operations.md#http-upload-operation).
 
+## Upload Request Is Forbidden
+
+Symptom: `POST /v1/pipelines/<pipeline-id>/upload` returns `403`.
+
+Likely cause: the bearer token is valid, but its configured `allow_pipelines` list does not include the requested upload pipeline.
+
+Diagnostic:
+
+```sh
+rg -n 'upload_tokens:|allow_pipelines:|id:' <config-path>
+```
+
+Safe fix: request the intended pipeline id, or update the token allowlist to include the configured `http_upload` pipeline that this producer may submit to.
+
+Reference: [Configuration](config.md#upload_tokens).
+
 ## Upload Request Is Rejected Before A Run ID
 
-Symptom: `POST /upload` returns `400`, `413`, `415`, or `503`.
+Symptom: `POST /v1/pipelines/<pipeline-id>/upload` returns `400`, `413`, `415`, or `503`.
 
-Likely cause: the request included a `pipeline` or `pipeline_id` query, archive content is malformed, the body exceeds size limits, content type is unsupported, or the in-memory upload queue is full.
+Likely cause: the request path has an invalid pipeline id, included a `pipeline` or `pipeline_id` query, archive content is malformed, the body exceeds size limits, content type is unsupported, or the in-memory upload queue is full.
 
 Diagnostic:
 
@@ -447,20 +463,20 @@ tar -tzf bundle.tar.gz
 rg -n 'max_upload_size|queue_size|max_concurrency' <config-path>
 ```
 
-Safe fix: send one valid tar or tar.gz source bundle archive with `Content-Type: application/x-tar`, `application/gzip`, or `application/x-gzip`; remove pipeline query parameters; reduce archive size or raise the configured limit; retry after queue pressure drops.
+Safe fix: send one valid tar or tar.gz source bundle archive to `/v1/pipelines/<pipeline-id>/upload` with `Content-Type: application/x-tar`, `application/gzip`, or `application/x-gzip`; remove pipeline query parameters; reduce archive size or raise the configured limit; retry after queue pressure drops.
 
 Reference: [Operations](operations.md#http-upload-operation).
 
 ## Upload Idempotency Conflict
 
-Symptom: `POST /upload` returns `409`.
+Symptom: `POST /v1/pipelines/<pipeline-id>/upload` returns `409`.
 
-Likely cause: the request reused an `Idempotency-Key` for the same authenticated pipeline with a different source manifest, or another request with the same key is still being staged before its manifest is known.
+Likely cause: the request reused an `Idempotency-Key` for the same token id and pipeline id with a different source manifest, or another request with the same key is still being staged before its manifest is known.
 
 Diagnostic:
 
 ```sh
-curl -i -X POST http://127.0.0.1:8080/upload \
+curl -i -X POST http://127.0.0.1:8080/v1/pipelines/<pipeline-id>/upload \
   -H "Authorization: Bearer $DISTRIBUTOR_UPLOAD_TOKEN" \
   -H "Content-Type: application/gzip" \
   -H "Idempotency-Key: <key>" \
@@ -469,7 +485,7 @@ curl -i -X POST http://127.0.0.1:8080/upload \
 
 Safe fix: if the response includes `"retryable":true`, retry the same upload later with the same key. Otherwise, inspect the producer operation and use the same key only for the same source bundle.
 
-Reference: [HTTP Upload API Contract](integrations/http-upload.md#post-upload).
+Reference: [HTTP Upload API Contract](integrations/http-upload.md).
 
 ## Upload Status Is Missing
 

examples/http-upload-local.yml

@@ -9,11 +9,15 @@ server:
     queue_size: 16
     max_concurrency: 1
     retention: 24h
+upload_tokens:
+  - id: example-uploader
+    token_env: DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN
+    allow_pipelines:
+      - example-http-upload
 pipelines:
   - id: example-http-upload
     source:
       backend: http_upload
-      token_env: DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN
     destinations:
       - id: local-archive
         backend: local
@@ -21,4 +25,3 @@ pipelines:
         publish:
           source: true
           html: false
-

examples/upload-client/main.go

@@ -23,6 +23,13 @@ func main() {
 	if len(os.Args) > 1 {
 		bundleRoot = os.Args[1]
 	}
+	pipelineID := os.Getenv("DISTRIBUTOR_EXAMPLE_UPLOAD_PIPELINE_ID")
+	if pipelineID == "" {
+		pipelineID = "example-http-upload"
+	}
+	if len(os.Args) > 2 {
+		pipelineID = os.Args[2]
+	}
 	idempotencyKey := os.Getenv("DISTRIBUTOR_EXAMPLE_UPLOAD_IDEMPOTENCY_KEY")
 
 	client, err := upload.NewClient(upload.ClientOptions{
@@ -35,7 +42,10 @@ func main() {
 
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
-	opts := upload.UploadBundleOptions{Root: bundleRoot}
+	opts := upload.UploadBundleOptions{
+		PipelineID: pipelineID,
+		Root:       bundleRoot,
+	}
 	if idempotencyKey != "" {
 		opts.IdempotencyKey = idempotencyKey
 	}

internal/app/run_test.go

@@ -292,7 +292,6 @@ func TestRunPipelineWithLocalSourcePublishesToRegisteredDestinationBackends(t *t
 			ID: "reports",
 			Source: config.Backend{
 				Backend: config.BackendHTTPUpload,
-				Upload:  config.HTTPUpload{TokenEnv: "UPLOAD_TOKEN"},
 			},
 			Destinations: []config.Destination{
 				{
@@ -309,6 +308,11 @@ func TestRunPipelineWithLocalSourcePublishesToRegisteredDestinationBackends(t *t
 				},
 			},
 		}},
+		UploadTokens: []config.UploadToken{{
+			ID:             "reporter",
+			TokenEnv:       "UPLOAD_TOKEN",
+			AllowPipelines: []string{"reports"},
+		}},
 	}
 	config.ApplyDefaults(&cfg)
 	provider := fakeBackendFactoryProvider(t, map[string]storage.Backend{
@@ -1674,11 +1678,15 @@ func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestina
 func writeUploadPipelineConfig(t *testing.T, destinationRoot string) string {
 	t.Helper()
 	return writeConfigFile(t, `
+upload_tokens:
+  - id: reporter
+    token_env: UPLOAD_TOKEN
+    allow_pipelines:
+      - reports
 pipelines:
   - id: reports
     source:
       backend: http_upload
-      token_env: UPLOAD_TOKEN
     destinations:
       - id: archive
         backend: local

internal/app/serve_test.go

@@ -70,14 +70,24 @@ func writeServeUploadConfig(t *testing.T, tokenEnvs []string) string {
 server:
   http:
     bind: 127.0.0.1:0
-pipelines:
+upload_tokens:
 `
 	for index, tokenEnv := range tokenEnvs {
 		body += `
+  - id: reporter-` + string(rune('a'+index)) + `
+    token_env: ` + tokenEnv + `
+    allow_pipelines:
+      - reports-` + string(rune('a'+index)) + `
+`
+	}
+	body += `
+pipelines:
+`
+	for index := range tokenEnvs {
+		body += `
   - id: reports-` + string(rune('a'+index)) + `
     source:
       backend: http_upload
-      token_env: ` + tokenEnv + `
     destinations:
       - id: archive
         backend: local

internal/app/upload_coordinator.go

@@ -44,6 +44,7 @@ type UploadRunRecord struct {
 }
 
 type UploadRequest struct {
+	TokenID        string
 	PipelineID     string
 	ContentType    string
 	Body           io.Reader
@@ -115,6 +116,7 @@ type uploadJob struct {
 }
 
 type uploadIdempotencyScope struct {
+	TokenID    string
 	PipelineID string
 	Key        string
 }
@@ -200,7 +202,7 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
 	if err := ingest.ValidateContentType(request.ContentType); err != nil {
 		return UploadRunRecord{}, err
 	}
-	scope, hasKey := uploadRequestIdempotencyScope(pipeline.ID, request.IdempotencyKey)
+	scope, hasKey := uploadRequestIdempotencyScope(request.TokenID, pipeline.ID, request.IdempotencyKey)
 
 	coordinator.mu.Lock()
 	coordinator.expireLocked(coordinator.now().UTC())
@@ -286,11 +288,11 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
 	return record, nil
 }
 
-func uploadRequestIdempotencyScope(pipelineID, key string) (uploadIdempotencyScope, bool) {
+func uploadRequestIdempotencyScope(tokenID, pipelineID, key string) (uploadIdempotencyScope, bool) {
 	if key == "" {
 		return uploadIdempotencyScope{}, false
 	}
-	return uploadIdempotencyScope{PipelineID: pipelineID, Key: key}, true
+	return uploadIdempotencyScope{TokenID: tokenID, PipelineID: pipelineID, Key: key}, true
 }
 
 func (coordinator *UploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bool) {

internal/app/upload_coordinator_test.go

@@ -271,6 +271,7 @@ func TestUploadCoordinatorIdempotencyReturnsOriginalRunForSameManifest(t *testin
 	})
 
 	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-a",
 		PipelineID:     "reports",
 		ContentType:    ingest.ContentTypeTar,
 		Body:           strings.NewReader("same"),
@@ -282,6 +283,7 @@ func TestUploadCoordinatorIdempotencyReturnsOriginalRunForSameManifest(t *testin
 	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
 
 	second, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-a",
 		PipelineID:     "reports",
 		ContentType:    ingest.ContentTypeTar,
 		Body:           strings.NewReader("same"),
@@ -310,6 +312,7 @@ func TestUploadCoordinatorIdempotencyConflictsForDifferentManifest(t *testing.T)
 	})
 
 	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-a",
 		PipelineID:     "reports",
 		ContentType:    ingest.ContentTypeTar,
 		Body:           strings.NewReader("one"),
@@ -321,6 +324,7 @@ func TestUploadCoordinatorIdempotencyConflictsForDifferentManifest(t *testing.T)
 	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
 
 	_, err = coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-a",
 		PipelineID:     "reports",
 		ContentType:    ingest.ContentTypeTar,
 		Body:           strings.NewReader("two"),
@@ -331,6 +335,42 @@ func TestUploadCoordinatorIdempotencyConflictsForDifferentManifest(t *testing.T)
 	}
 }
 
+func TestUploadCoordinatorIdempotencyIsScopedByToken(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports"},
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        manifestUploadStage,
+		run:          successfulUploadRun,
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-a",
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("one"),
+		IdempotencyKey: "shared-key",
+	})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	second, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-b",
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("two"),
+		IdempotencyKey: "shared-key",
+	})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	if second.ID == first.ID {
+		t.Fatalf("run ids matched across tokens: %q", second.ID)
+	}
+}
+
 func TestUploadCoordinatorIdempotencyIsScopedByPipeline(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -343,6 +383,7 @@ func TestUploadCoordinatorIdempotencyIsScopedByPipeline(t *testing.T) {
 	})
 
 	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-a",
 		PipelineID:     "reports-one",
 		ContentType:    ingest.ContentTypeTar,
 		Body:           strings.NewReader("one"),
@@ -352,6 +393,7 @@ func TestUploadCoordinatorIdempotencyIsScopedByPipeline(t *testing.T) {
 		t.Fatalf("first Submit() error = %v", err)
 	}
 	second, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-a",
 		PipelineID:     "reports-two",
 		ContentType:    ingest.ContentTypeTar,
 		Body:           strings.NewReader("two"),
@@ -408,6 +450,7 @@ func TestUploadCoordinatorIdempotencyReturnsRetryableConflictWhileStaging(t *tes
 	firstErr := make(chan error, 1)
 	go func() {
 		_, err := coordinator.Submit(context.Background(), UploadRequest{
+			TokenID:        "reporter-a",
 			PipelineID:     "reports",
 			ContentType:    ingest.ContentTypeTar,
 			Body:           strings.NewReader("same"),
@@ -419,6 +462,7 @@ func TestUploadCoordinatorIdempotencyReturnsRetryableConflictWhileStaging(t *tes
 
 	var reads atomic.Int64
 	_, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:     "reporter-a",
 		PipelineID:  "reports",
 		ContentType: ingest.ContentTypeTar,
 		Body: readerFunc(func(data []byte) (int, error) {
@@ -440,6 +484,57 @@ func TestUploadCoordinatorIdempotencyReturnsRetryableConflictWhileStaging(t *tes
 	}
 }
 
+func TestUploadCoordinatorIdempotencyPendingScopeIncludesToken(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var calls atomic.Int64
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports"},
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage: func(ctx context.Context, opts ingest.StageOptions) (ingest.StagedBundle, error) {
+			if calls.Add(1) == 1 {
+				close(entered)
+				<-release
+			}
+			return manifestUploadStage(ctx, opts)
+		},
+		run: successfulUploadRun,
+	})
+	firstErr := make(chan error, 1)
+	go func() {
+		_, err := coordinator.Submit(context.Background(), UploadRequest{
+			TokenID:        "reporter-a",
+			PipelineID:     "reports",
+			ContentType:    ingest.ContentTypeTar,
+			Body:           strings.NewReader("same"),
+			IdempotencyKey: "in-flight",
+		})
+		firstErr <- err
+	}()
+	<-entered
+
+	second, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-b",
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("same"),
+		IdempotencyKey: "in-flight",
+	})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	if second.ID == "" {
+		t.Fatal("second run id is empty, want accepted run")
+	}
+	close(release)
+	if err := <-firstErr; err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+}
+
 func TestUploadCoordinatorIdempotencyExpiresWithCompletedStatus(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -455,6 +550,7 @@ func TestUploadCoordinatorIdempotencyExpiresWithCompletedStatus(t *testing.T) {
 	})
 
 	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-a",
 		PipelineID:     "reports",
 		ContentType:    ingest.ContentTypeTar,
 		Body:           strings.NewReader("same"),
@@ -469,6 +565,7 @@ func TestUploadCoordinatorIdempotencyExpiresWithCompletedStatus(t *testing.T) {
 	coordinator.Expire()
 
 	second, err := coordinator.Submit(context.Background(), UploadRequest{
+		TokenID:        "reporter-a",
 		PipelineID:     "reports",
 		ContentType:    ingest.ContentTypeTar,
 		Body:           strings.NewReader("same"),
@@ -564,11 +661,11 @@ func uploadCoordinatorConfig(t *testing.T, opts uploadCoordinatorConfigOptions)
 		}},
 	}
 	for _, pipelineID := range opts.pipelineIDs {
+		tokenEnv := strings.ToUpper(strings.ReplaceAll(pipelineID, "-", "_")) + "_TOKEN"
 		cfg.Pipelines = append(cfg.Pipelines, config.Pipeline{
 			ID: pipelineID,
 			Source: config.Backend{
 				Backend: config.BackendHTTPUpload,
-				Upload:  config.HTTPUpload{TokenEnv: strings.ToUpper(strings.ReplaceAll(pipelineID, "-", "_")) + "_TOKEN"},
 			},
 			Destinations: []config.Destination{{
 				ID:      "archive",
@@ -576,6 +673,11 @@ func uploadCoordinatorConfig(t *testing.T, opts uploadCoordinatorConfigOptions)
 				Path:    t.TempDir(),
 			}},
 		})
+		cfg.UploadTokens = append(cfg.UploadTokens, config.UploadToken{
+			ID:             pipelineID + "-reporter",
+			TokenEnv:       tokenEnv,
+			AllowPipelines: []string{pipelineID},
+		})
 	}
 	return cfg
 }

internal/app/upload_http.go

@@ -20,7 +20,14 @@ type uploadCoordinator interface {
 
 type uploadHTTPHandler struct {
 	coordinator     uploadCoordinator
-	tokens      map[string]string
+	tokens          map[string]resolvedUploadToken
+	uploadPipelines map[string]struct{}
+}
+
+type resolvedUploadToken struct {
+	ID               string
+	Value            string
+	AllowedPipelines map[string]struct{}
 }
 
 type uploadAcceptedResponse struct {
@@ -44,36 +51,55 @@ func newUploadHTTPHandler(ctx context.Context, cfg config.Config, environment co
 	return uploadHTTPHandler{
 		coordinator:     NewUploadCoordinator(ctx, cfg),
 		tokens:          tokens,
+		uploadPipelines: uploadPipelineSet(cfg),
 	}, nil
 }
 
-func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]string, error) {
+func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]resolvedUploadToken, error) {
-	tokens := make(map[string]string)
+	tokens := make(map[string]resolvedUploadToken)
-	for _, pipeline := range cfg.Pipelines {
+	for _, uploadToken := range cfg.UploadTokens {
-		if pipeline.Source.Backend != config.BackendHTTPUpload {
+		token, ok := environment.Lookup(uploadToken.TokenEnv)
-			continue
-		}
-		tokenName := pipeline.Source.Upload.TokenEnv
-		token, ok := environment.Lookup(tokenName)
 		if !ok {
-			return nil, fmt.Errorf("upload token environment variable %s is not set", tokenName)
+			return nil, fmt.Errorf("upload token %s environment variable %s is not set", uploadToken.ID, uploadToken.TokenEnv)
 		}
 		if token == "" {
-			return nil, fmt.Errorf("upload token environment variable %s is empty", tokenName)
+			return nil, fmt.Errorf("upload token %s environment variable %s is empty", uploadToken.ID, uploadToken.TokenEnv)
 		}
 		if existing, exists := tokens[token]; exists {
-			return nil, fmt.Errorf("upload token environment variables for pipelines %s and %s resolve to the same value", existing, pipeline.ID)
+			return nil, fmt.Errorf("upload token environment variables for tokens %s and %s resolve to the same value", existing.ID, uploadToken.ID)
+		}
+		tokens[token] = resolvedUploadToken{
+			ID:               uploadToken.ID,
+			Value:            token,
+			AllowedPipelines: pipelineIDSet(uploadToken.AllowPipelines),
 		}
-		tokens[token] = pipeline.ID
 	}
 	return tokens, nil
 }
 
+func uploadPipelineSet(cfg config.Config) map[string]struct{} {
+	pipelines := make(map[string]struct{})
+	for _, pipeline := range cfg.Pipelines {
+		if pipeline.Source.Backend == config.BackendHTTPUpload {
+			pipelines[pipeline.ID] = struct{}{}
+		}
+	}
+	return pipelines
+}
+
+func pipelineIDSet(ids []string) map[string]struct{} {
+	set := make(map[string]struct{}, len(ids))
+	for _, id := range ids {
+		set[id] = struct{}{}
+	}
+	return set
+}
+
 func (handler uploadHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	switch {
 	case r.Method == http.MethodGet && r.URL.Path == "/healthz":
 		handler.handleHealth(w)
-	case r.Method == http.MethodPost && r.URL.Path == "/upload":
+	case r.Method == http.MethodPost && strings.HasPrefix(r.URL.Path, "/v1/pipelines/"):
 		handler.handleUpload(w, r)
 	case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/runs/"):
 		handler.handleRunStatus(w, r)
@@ -91,11 +117,28 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req
 		writeHTTPError(w, http.StatusBadRequest, "pipeline id is not accepted")
 		return
 	}
-	pipelineID, ok := handler.authenticate(r.Header.Get("Authorization"))
+	pipelineID, ok := uploadPipelineIDFromPath(r.URL.Path)
+	if !ok {
+		writeHTTPError(w, http.StatusNotFound, "not found")
+		return
+	}
+	if !config.IsSlugLikeID(pipelineID) {
+		writeHTTPError(w, http.StatusBadRequest, "invalid pipeline id")
+		return
+	}
+	token, ok := handler.authenticate(r.Header.Get("Authorization"))
 	if !ok {
 		writeHTTPError(w, http.StatusUnauthorized, "unauthorized")
 		return
 	}
+	if _, ok := handler.uploadPipelines[pipelineID]; !ok {
+		writeHTTPError(w, http.StatusNotFound, "upload pipeline not found")
+		return
+	}
+	if _, ok := token.AllowedPipelines[pipelineID]; !ok {
+		writeHTTPError(w, http.StatusForbidden, "forbidden")
+		return
+	}
 	contentType := r.Header.Get("Content-Type")
 	if err := ingest.ValidateContentType(contentType); err != nil {
 		writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type")
@@ -107,6 +150,7 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req
 		return
 	}
 	record, err := handler.coordinator.Submit(r.Context(), UploadRequest{
+		TokenID:        token.ID,
 		PipelineID:     pipelineID,
 		ContentType:    contentType,
 		Body:           r.Body,
@@ -122,6 +166,19 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req
 	})
 }
 
+func uploadPipelineIDFromPath(path string) (string, bool) {
+	const prefix = "/v1/pipelines/"
+	const suffix = "/upload"
+	if !strings.HasPrefix(path, prefix) || !strings.HasSuffix(path, suffix) {
+		return "", false
+	}
+	pipelineID := strings.TrimSuffix(strings.TrimPrefix(path, prefix), suffix)
+	if pipelineID == "" || strings.Contains(pipelineID, "/") {
+		return "", false
+	}
+	return pipelineID, true
+}
+
 func (handler uploadHTTPHandler) handleRunStatus(w http.ResponseWriter, r *http.Request) {
 	rawRunID := strings.TrimPrefix(r.URL.Path, "/runs/")
 	if rawRunID == "" || strings.Contains(rawRunID, "/") {
@@ -136,17 +193,17 @@ func (handler uploadHTTPHandler) handleRunStatus(w http.ResponseWriter, r *http.
 	writeJSON(w, http.StatusOK, record)
 }
 
-func (handler uploadHTTPHandler) authenticate(header string) (string, bool) {
+func (handler uploadHTTPHandler) authenticate(header string) (resolvedUploadToken, bool) {
 	const prefix = "Bearer "
 	if !strings.HasPrefix(header, prefix) {
-		return "", false
+		return resolvedUploadToken{}, false
 	}
 	token := strings.TrimSpace(strings.TrimPrefix(header, prefix))
 	if token == "" {
-		return "", false
+		return resolvedUploadToken{}, false
 	}
-	pipelineID, ok := handler.tokens[token]
+	resolved, ok := handler.tokens[token]
-	return pipelineID, ok
+	return resolved, ok
 }
 
 func uploadIdempotencyKey(header http.Header) (string, error) {

internal/app/upload_http_integration_test.go

@@ -21,6 +21,7 @@ import (
 	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
 	"gitea.maximumdirect.net/eric/distributor/internal/storage"
 	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
+	clientupload "gitea.maximumdirect.net/eric/distributor/pkg/upload"
 )
 
 func TestHTTPUploadPublishesTarAndGzipFanout(t *testing.T) {
@@ -79,14 +80,15 @@ func TestHTTPUploadInvalidArchiveIsRejectedWithoutRunID(t *testing.T) {
 	}}, 4, 1))
 	handler := uploadHTTPHandler{
 		coordinator:     coordinator,
-		tokens:      map[string]string{"reports-secret": "reports"},
+		tokens:          map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")},
+		uploadPipelines: pipelineIDSet([]string{"reports"}),
 	}
 	server := httptest.NewServer(handler)
 	defer server.Close()
 
 	status, body := postHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("not a tar archive"))
 	if status != http.StatusBadRequest {
-		t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusBadRequest, body)
+		t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusBadRequest, body)
 	}
 	if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") {
 		t.Fatalf("invalid archive response exposed run id or token: %s", body)
@@ -107,7 +109,8 @@ func TestHTTPUploadIdempotencyReturnsOriginalRunForSameBundle(t *testing.T) {
 	}}, 4, 1))
 	handler := uploadHTTPHandler{
 		coordinator:     coordinator,
-		tokens:      map[string]string{"reports-secret": "reports"},
+		tokens:          map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")},
+		uploadPipelines: pipelineIDSet([]string{"reports"}),
 	}
 	server := httptest.NewServer(handler)
 	defer server.Close()
@@ -134,7 +137,8 @@ func TestHTTPUploadIdempotencyReturnsConflictForDifferentBundle(t *testing.T) {
 	}}, 4, 1))
 	handler := uploadHTTPHandler{
 		coordinator:     coordinator,
-		tokens:      map[string]string{"reports-secret": "reports"},
+		tokens:          map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")},
+		uploadPipelines: pipelineIDSet([]string{"reports"}),
 	}
 	server := httptest.NewServer(handler)
 	defer server.Close()
@@ -146,7 +150,7 @@ func TestHTTPUploadIdempotencyReturnsConflictForDifferentBundle(t *testing.T) {
 		ID: "weather.daily.brentwood.2026-05-31",
 	}))
 	if status != http.StatusConflict {
-		t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusConflict, body)
+		t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusConflict, body)
 	}
 	if strings.Contains(body, "reports-secret") {
 		t.Fatalf("conflict response exposed token: %s", body)
@@ -168,14 +172,15 @@ func TestHTTPUploadOversizedArchiveIsRejectedWithoutRunID(t *testing.T) {
 	coordinator := NewUploadCoordinator(context.Background(), cfg)
 	handler := uploadHTTPHandler{
 		coordinator:     coordinator,
-		tokens:      map[string]string{"reports-secret": "reports"},
+		tokens:          map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")},
+		uploadPipelines: pipelineIDSet([]string{"reports"}),
 	}
 	server := httptest.NewServer(handler)
 	defer server.Close()
 
 	status, body := postHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, bundleArchive(t, false, testutil.BundleOptions{}))
 	if status != http.StatusRequestEntityTooLarge {
-		t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusRequestEntityTooLarge, body)
+		t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusRequestEntityTooLarge, body)
 	}
 	if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") {
 		t.Fatalf("oversized response exposed run id or token: %s", body)
@@ -211,7 +216,8 @@ func TestHTTPUploadSamePipelineRequestsSerialize(t *testing.T) {
 	})
 	handler := uploadHTTPHandler{
 		coordinator:     coordinator,
-		tokens:      map[string]string{"reports-secret": "reports"},
+		tokens:          map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")},
+		uploadPipelines: pipelineIDSet([]string{"reports"}),
 	}
 	server := httptest.NewServer(handler)
 	defer server.Close()
@@ -262,16 +268,17 @@ func TestHTTPUploadDifferentPipelinesRunConcurrently(t *testing.T) {
 	})
 	handler := uploadHTTPHandler{
 		coordinator: coordinator,
-		tokens: map[string]string{
+		tokens: map[string]resolvedUploadToken{
-			"one-secret": "reports-one",
+			"one-secret": uploadHTTPTestToken("reports-one-reporter", "one-secret", "reports-one"),
-			"two-secret": "reports-two",
+			"two-secret": uploadHTTPTestToken("reports-two-reporter", "two-secret", "reports-two"),
 		},
+		uploadPipelines: pipelineIDSet([]string{"reports-one", "reports-two"}),
 	}
 	server := httptest.NewServer(handler)
 	defer server.Close()
 
-	firstRunID := submitHTTPUpload(t, server, "one-secret", ingest.ContentTypeTar, []byte("first"))
+	firstRunID := submitHTTPUploadToPipeline(t, server, "reports-one", "one-secret", ingest.ContentTypeTar, []byte("first"))
-	secondRunID := submitHTTPUpload(t, server, "two-secret", ingest.ContentTypeTar, []byte("second"))
+	secondRunID := submitHTTPUploadToPipeline(t, server, "reports-two", "two-secret", ingest.ContentTypeTar, []byte("second"))
 	waitForStartedPipelines(t, started, "reports-one", "reports-two")
 	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusRunning)
 	waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusRunning)
@@ -284,6 +291,199 @@ func TestHTTPUploadDifferentPipelinesRunConcurrently(t *testing.T) {
 	waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusSucceeded)
 }
 
+func TestHTTPUploadOneTokenCanUploadToMultiplePipelines(t *testing.T) {
+	firstDestination := t.TempDir()
+	secondDestination := t.TempDir()
+	cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{
+		{
+			id:           "reports-one",
+			tokenEnv:     "SHARED_UPLOAD_TOKEN",
+			stagingPath:  filepath.Join(t.TempDir(), "reports-one"),
+			destinations: []string{firstDestination},
+		},
+		{
+			id:           "reports-two",
+			tokenEnv:     "SHARED_UPLOAD_TOKEN",
+			stagingPath:  filepath.Join(t.TempDir(), "reports-two"),
+			destinations: []string{secondDestination},
+		},
+	}, 4, 1)
+	cfg.UploadTokens = []config.UploadToken{{
+		ID:             "shared-reporter",
+		TokenEnv:       "SHARED_UPLOAD_TOKEN",
+		AllowPipelines: []string{"reports-one", "reports-two"},
+	}}
+	handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{
+		"SHARED_UPLOAD_TOKEN": "shared-secret",
+	}))
+	if err != nil {
+		t.Fatalf("newUploadHTTPHandler() error = %v", err)
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	firstRunID := submitHTTPUploadToPipeline(t, server, "reports-one", "shared-secret", ingest.ContentTypeTar, bundleArchive(t, false, testutil.BundleOptions{
+		ID: "reports.one.2026-06-08",
+	}))
+	secondRunID := submitHTTPUploadToPipeline(t, server, "reports-two", "shared-secret", ingest.ContentTypeTar, bundleArchive(t, false, testutil.BundleOptions{
+		ID: "reports.two.2026-06-08",
+	}))
+
+	first := waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)
+	second := waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusSucceeded)
+	if first.PipelineID != "reports-one" || second.PipelineID != "reports-two" {
+		t.Fatalf("statuses pipeline = %q/%q, want reports-one/reports-two", first.PipelineID, second.PipelineID)
+	}
+	assertPublishedBundle(t, firstDestination)
+	assertPublishedBundle(t, secondDestination)
+}
+
+func TestHTTPUploadMultipleTokensCanUploadToOnePipeline(t *testing.T) {
+	destination := t.TempDir()
+	cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
+		id:           "reports",
+		tokenEnv:     "FIRST_UPLOAD_TOKEN",
+		stagingPath:  filepath.Join(t.TempDir(), "reports"),
+		destinations: []string{destination},
+	}}, 4, 1)
+	cfg.UploadTokens = []config.UploadToken{
+		{ID: "first-reporter", TokenEnv: "FIRST_UPLOAD_TOKEN", AllowPipelines: []string{"reports"}},
+		{ID: "second-reporter", TokenEnv: "SECOND_UPLOAD_TOKEN", AllowPipelines: []string{"reports"}},
+	}
+	handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{
+		"FIRST_UPLOAD_TOKEN":  "first-secret",
+		"SECOND_UPLOAD_TOKEN": "second-secret",
+	}))
+	if err != nil {
+		t.Fatalf("newUploadHTTPHandler() error = %v", err)
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	firstRunID := submitHTTPUpload(t, server, "first-secret", ingest.ContentTypeTar, bundleArchive(t, false, testutil.BundleOptions{}))
+	secondRunID := submitHTTPUpload(t, server, "second-secret", ingest.ContentTypeTar, bundleArchive(t, false, testutil.BundleOptions{}))
+
+	first := waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)
+	second := waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusSucceeded)
+	if first.PipelineID != "reports" || second.PipelineID != "reports" {
+		t.Fatalf("statuses pipeline = %q/%q, want reports/reports", first.PipelineID, second.PipelineID)
+	}
+	assertPublishedBundle(t, destination)
+}
+
+func TestHTTPUploadRejectsDisallowedPipelineAndLegacyUploadWithoutQueueing(t *testing.T) {
+	coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{
+		{
+			id:           "reports",
+			tokenEnv:     "REPORTS_TOKEN",
+			stagingPath:  filepath.Join(t.TempDir(), "reports"),
+			destinations: []string{t.TempDir()},
+		},
+		{
+			id:           "private",
+			tokenEnv:     "PRIVATE_TOKEN",
+			stagingPath:  filepath.Join(t.TempDir(), "private"),
+			destinations: []string{t.TempDir()},
+		},
+	}, 4, 1))
+	handler := uploadHTTPHandler{
+		coordinator: coordinator,
+		tokens: map[string]resolvedUploadToken{
+			"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports"),
+		},
+		uploadPipelines: pipelineIDSet([]string{"reports", "private"}),
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	status, body := postHTTPUploadToPipeline(t, server, "private", "reports-secret", ingest.ContentTypeTar, []byte("archive"))
+	if status != http.StatusForbidden {
+		t.Fatalf("disallowed upload status = %d, want %d; body = %s", status, http.StatusForbidden, body)
+	}
+	status, body = postLegacyHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("archive"))
+	if status != http.StatusNotFound {
+		t.Fatalf("legacy upload status = %d, want %d; body = %s", status, http.StatusNotFound, body)
+	}
+	if got := coordinator.QueueDepth(); got != 0 {
+		t.Fatalf("queue depth = %d, want 0", got)
+	}
+}
+
+func TestHTTPUploadPublishesThroughSelectedPipeline(t *testing.T) {
+	firstDestination := t.TempDir()
+	secondDestination := t.TempDir()
+	cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{
+		{
+			id:           "reports-one",
+			tokenEnv:     "SHARED_UPLOAD_TOKEN",
+			stagingPath:  filepath.Join(t.TempDir(), "reports-one"),
+			destinations: []string{firstDestination},
+		},
+		{
+			id:           "reports-two",
+			tokenEnv:     "SHARED_UPLOAD_TOKEN",
+			stagingPath:  filepath.Join(t.TempDir(), "reports-two"),
+			destinations: []string{secondDestination},
+		},
+	}, 4, 1)
+	cfg.UploadTokens = []config.UploadToken{{
+		ID:             "shared-reporter",
+		TokenEnv:       "SHARED_UPLOAD_TOKEN",
+		AllowPipelines: []string{"reports-one", "reports-two"},
+	}}
+	handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{
+		"SHARED_UPLOAD_TOKEN": "shared-secret",
+	}))
+	if err != nil {
+		t.Fatalf("newUploadHTTPHandler() error = %v", err)
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	bundleRoot := t.TempDir()
+	testutil.WriteSourceBundle(t, bundleRoot, "", testutil.BundleOptions{
+		ID: "reports.selected.2026-06-08",
+	})
+	client, err := clientupload.NewClient(clientupload.ClientOptions{
+		Endpoint:   server.URL,
+		Token:      "shared-secret",
+		HTTPClient: server.Client(),
+	})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	result, err := client.UploadBundle(context.Background(), clientupload.UploadBundleOptions{
+		PipelineID: "reports-two",
+		Root:       bundleRoot,
+	})
+	if err != nil {
+		t.Fatalf("UploadBundle() error = %v", err)
+	}
+	runID := UploadRunID(result.RunID)
+	record := waitForHTTPUploadStatus(t, server, runID, UploadStatusSucceeded)
+
+	if record.PipelineID != "reports-two" {
+		t.Fatalf("record pipeline = %q, want reports-two", record.PipelineID)
+	}
+	if record.Report == nil {
+		t.Fatal("completed status report = nil, want run report")
+	}
+	if got, want := len(record.Report.Pipelines), 1; got != want {
+		t.Fatalf("report pipeline count = %d, want %d", got, want)
+	}
+	if record.Report.Pipelines[0].ID != "reports-two" {
+		t.Fatalf("report pipeline = %q, want reports-two", record.Report.Pipelines[0].ID)
+	}
+	if got, want := len(record.Report.Actions), 1; got != want {
+		t.Fatalf("report action count = %d, want %d", got, want)
+	}
+	if record.Report.Actions[0].PipelineID != "reports-two" {
+		t.Fatalf("action pipeline = %q, want reports-two", record.Report.Actions[0].PipelineID)
+	}
+	assertDirectoryEmpty(t, firstDestination)
+	assertPublishedBundle(t, secondDestination)
+}
+
 type httpUploadPipelineSpec struct {
 	id           string
 	tokenEnv     string
@@ -311,7 +511,6 @@ func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpe
 			Source: config.Backend{
 				Backend: config.BackendHTTPUpload,
 				Upload: config.HTTPUpload{
-					TokenEnv:      spec.tokenEnv,
 					StagingPath:   spec.stagingPath,
 					MaxUploadSize: &size,
 				},
@@ -326,6 +525,11 @@ func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpe
 			})
 		}
 		cfg.Pipelines = append(cfg.Pipelines, pipeline)
+		cfg.UploadTokens = append(cfg.UploadTokens, config.UploadToken{
+			ID:             spec.id + "-reporter",
+			TokenEnv:       spec.tokenEnv,
+			AllowPipelines: []string{spec.id},
+		})
 	}
 	config.ApplyDefaults(&cfg)
 	return cfg
@@ -333,7 +537,12 @@ func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpe
 
 func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) UploadRunID {
 	t.Helper()
-	status, responseBody := postHTTPUpload(t, server, token, contentType, body)
+	return submitHTTPUploadToPipeline(t, server, "reports", token, contentType, body)
+}
+
+func submitHTTPUploadToPipeline(t *testing.T, server *httptest.Server, pipelineID, token, contentType string, body []byte) UploadRunID {
+	t.Helper()
+	status, responseBody := postHTTPUploadToPipeline(t, server, pipelineID, token, contentType, body)
 	return decodeAcceptedHTTPUpload(t, status, responseBody)
 }
 
@@ -346,7 +555,7 @@ func submitHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, conte
 func decodeAcceptedHTTPUpload(t *testing.T, status int, responseBody string) UploadRunID {
 	t.Helper()
 	if status != http.StatusAccepted {
-		t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusAccepted, responseBody)
+		t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusAccepted, responseBody)
 	}
 	var accepted uploadAcceptedResponse
 	if err := json.Unmarshal([]byte(responseBody), &accepted); err != nil {
@@ -360,12 +569,22 @@ func decodeAcceptedHTTPUpload(t *testing.T, status int, responseBody string) Upl
 
 func postHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) (int, string) {
 	t.Helper()
-	return postHTTPUploadWithKey(t, server, token, contentType, "", body)
+	return postHTTPUploadToPipeline(t, server, "reports", token, contentType, body)
+}
+
+func postHTTPUploadToPipeline(t *testing.T, server *httptest.Server, pipelineID, token, contentType string, body []byte) (int, string) {
+	t.Helper()
+	return postHTTPUploadWithKeyToPipeline(t, server, pipelineID, token, contentType, "", body)
 }
 
 func postHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, contentType, key string, body []byte) (int, string) {
 	t.Helper()
-	request, err := http.NewRequest(http.MethodPost, server.URL+"/upload", bytes.NewReader(body))
+	return postHTTPUploadWithKeyToPipeline(t, server, "reports", token, contentType, key, body)
+}
+
+func postHTTPUploadWithKeyToPipeline(t *testing.T, server *httptest.Server, pipelineID, token, contentType, key string, body []byte) (int, string) {
+	t.Helper()
+	request, err := http.NewRequest(http.MethodPost, server.URL+"/v1/pipelines/"+pipelineID+"/upload", bytes.NewReader(body))
 	if err != nil {
 		t.Fatalf("NewRequest() error = %v", err)
 	}
@@ -376,7 +595,27 @@ func postHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, content
 	}
 	response, err := server.Client().Do(request)
 	if err != nil {
-		t.Fatalf("POST /upload error = %v", err)
+		t.Fatalf("POST upload error = %v", err)
+	}
+	defer response.Body.Close()
+	data, err := io.ReadAll(response.Body)
+	if err != nil {
+		t.Fatalf("read response body: %v", err)
+	}
+	return response.StatusCode, string(data)
+}
+
+func postLegacyHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) (int, string) {
+	t.Helper()
+	request, err := http.NewRequest(http.MethodPost, server.URL+"/upload", bytes.NewReader(body))
+	if err != nil {
+		t.Fatalf("NewRequest() error = %v", err)
+	}
+	request.Header.Set("Authorization", "Bearer "+token)
+	request.Header.Set("Content-Type", contentType)
+	response, err := server.Client().Do(request)
+	if err != nil {
+		t.Fatalf("POST legacy upload error = %v", err)
 	}
 	defer response.Body.Close()
 	data, err := io.ReadAll(response.Body)

internal/app/upload_http_test.go

@@ -53,10 +53,14 @@ func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) {
 		ID: "weekly",
 		Source: config.Backend{
 			Backend: config.BackendHTTPUpload,
-			Upload:  config.HTTPUpload{TokenEnv: "OTHER_UPLOAD_TOKEN"},
 		},
 		Destinations: cfg.Pipelines[0].Destinations,
 	})
+	cfg.UploadTokens = append(cfg.UploadTokens, config.UploadToken{
+		ID:             "weekly-reporter",
+		TokenEnv:       "OTHER_UPLOAD_TOKEN",
+		AllowPipelines: []string{"weekly"},
+	})
 	config.ApplyDefaults(&cfg)
 	secret := "super-secret-token"
 	_, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{
@@ -71,6 +75,39 @@ func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) {
 	}
 }
 
+func TestResolveUploadTokensAllowsMultiplePipelines(t *testing.T) {
+	cfg := uploadHTTPTestConfig()
+	cfg.Pipelines = append(cfg.Pipelines, config.Pipeline{
+		ID: "weekly",
+		Source: config.Backend{
+			Backend: config.BackendHTTPUpload,
+		},
+		Destinations: cfg.Pipelines[0].Destinations,
+	})
+	cfg.UploadTokens[0].AllowPipelines = []string{"reports", "weekly"}
+	config.ApplyDefaults(&cfg)
+
+	tokens, err := resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{
+		"UPLOAD_TOKEN": "secret",
+	}))
+	if err != nil {
+		t.Fatalf("resolveUploadTokens() error = %v", err)
+	}
+	token, ok := tokens["secret"]
+	if !ok {
+		t.Fatal("resolved token missing")
+	}
+	if token.ID != "reporter" || token.Value != "secret" {
+		t.Fatalf("resolved token = %#v, want id and value", token)
+	}
+	if _, ok := token.AllowedPipelines["reports"]; !ok {
+		t.Fatalf("allowed pipelines = %#v, want reports", token.AllowedPipelines)
+	}
+	if _, ok := token.AllowedPipelines["weekly"]; !ok {
+		t.Fatalf("allowed pipelines = %#v, want weekly", token.AllowedPipelines)
+	}
+}
+
 func TestNewUploadHTTPHandlerAcceptsDefaultedConfig(t *testing.T) {
 	cfg := uploadHTTPTestConfig()
 	cfg.Server.HTTP.Bind = ""
@@ -109,10 +146,11 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
 				return UploadRunRecord{ID: "reports.20260603T120000Z.abcdef12", Status: UploadStatusAccepted}, nil
 			},
 		},
-		tokens: map[string]string{"valid-token": "reports"},
+		tokens:          map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")},
+		uploadPipelines: pipelineIDSet([]string{"reports"}),
 	}
 	recorder := httptest.NewRecorder()
-	request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
+	request := httptest.NewRequest(http.MethodPost, "/v1/pipelines/reports/upload", strings.NewReader("archive"))
 	request.Header.Set("Authorization", "Bearer valid-token")
 	request.Header.Set("Content-Type", "application/x-tar")
 	request.Header.Set("Idempotency-Key", "producer.retry:20260603")
@@ -125,6 +163,9 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
 	if submitted.PipelineID != "reports" {
 		t.Fatalf("submitted pipeline = %q, want reports", submitted.PipelineID)
 	}
+	if submitted.TokenID != "reporter" {
+		t.Fatalf("submitted token id = %q, want reporter", submitted.TokenID)
+	}
 	if submitted.IdempotencyKey != "producer.retry:20260603" {
 		t.Fatalf("submitted idempotency key = %q, want producer.retry:20260603", submitted.IdempotencyKey)
 	}
@@ -143,12 +184,13 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
 func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) {
 	handler := uploadHTTPHandler{
 		coordinator:     fakeUploadCoordinator{},
-		tokens:      map[string]string{"valid-token": "reports"},
+		tokens:          map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")},
+		uploadPipelines: pipelineIDSet([]string{"reports"}),
 	}
 
-	for _, authHeader := range []string{"", "Bearer wrong-token"} {
+	for _, authHeader := range []string{"", "Basic valid-token", "Bearer", "Bearer wrong-token"} {
 		recorder := httptest.NewRecorder()
-		request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
+		request := httptest.NewRequest(http.MethodPost, "/v1/pipelines/reports/upload", strings.NewReader("archive"))
 		request.Header.Set("Authorization", authHeader)
 		request.Header.Set("Content-Type", "application/x-tar")
 
@@ -163,7 +205,73 @@ func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) {
 	}
 }
 
-func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t *testing.T) {
+func TestUploadHTTPHandlerRejectsForbiddenPipeline(t *testing.T) {
+	handler := uploadHTTPHandler{
+		coordinator: fakeUploadCoordinator{
+			submit: func(context.Context, UploadRequest) (UploadRunRecord, error) {
+				t.Fatal("Submit should not be called")
+				return UploadRunRecord{}, nil
+			},
+		},
+		tokens:          map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")},
+		uploadPipelines: pipelineIDSet([]string{"reports", "private"}),
+	}
+	recorder := httptest.NewRecorder()
+	request := httptest.NewRequest(http.MethodPost, "/v1/pipelines/private/upload", strings.NewReader("archive"))
+	request.Header.Set("Authorization", "Bearer valid-token")
+	request.Header.Set("Content-Type", "application/x-tar")
+
+	handler.ServeHTTP(recorder, request)
+
+	if recorder.Code != http.StatusForbidden {
+		t.Fatalf("status = %d, want %d; body = %q", recorder.Code, http.StatusForbidden, recorder.Body.String())
+	}
+	if strings.Contains(recorder.Body.String(), "valid-token") {
+		t.Fatalf("forbidden response exposed token: %q", recorder.Body.String())
+	}
+}
+
+func TestUploadHTTPHandlerRejectsInvalidPathAndRemovedLegacyUpload(t *testing.T) {
+	handler := uploadHTTPHandler{
+		coordinator: fakeUploadCoordinator{
+			submit: func(context.Context, UploadRequest) (UploadRunRecord, error) {
+				t.Fatal("Submit should not be called")
+				return UploadRunRecord{}, nil
+			},
+		},
+		tokens:          map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")},
+		uploadPipelines: pipelineIDSet([]string{"reports"}),
+	}
+	tests := []struct {
+		name       string
+		url        string
+		wantStatus int
+	}{
+		{name: "legacy upload", url: "/upload", wantStatus: http.StatusNotFound},
+		{name: "missing pipeline", url: "/v1/pipelines//upload", wantStatus: http.StatusNotFound},
+		{name: "extra segment", url: "/v1/pipelines/reports/upload/extra", wantStatus: http.StatusNotFound},
+		{name: "invalid pipeline id", url: "/v1/pipelines/.reports/upload", wantStatus: http.StatusBadRequest},
+		{name: "pipeline query", url: "/v1/pipelines/reports/upload?pipeline=other", wantStatus: http.StatusBadRequest},
+		{name: "pipeline id query", url: "/v1/pipelines/reports/upload?pipeline_id=other", wantStatus: http.StatusBadRequest},
+		{name: "unknown upload pipeline", url: "/v1/pipelines/missing/upload", wantStatus: http.StatusNotFound},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			recorder := httptest.NewRecorder()
+			request := httptest.NewRequest(http.MethodPost, tt.url, strings.NewReader("archive"))
+			request.Header.Set("Authorization", "Bearer valid-token")
+			request.Header.Set("Content-Type", "application/x-tar")
+
+			handler.ServeHTTP(recorder, request)
+
+			if recorder.Code != tt.wantStatus {
+				t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String())
+			}
+		})
+	}
+}
+
+func TestUploadHTTPHandlerRejectsUnsupportedContentTypeAndInvalidKey(t *testing.T) {
 	tests := []struct {
 		name        string
 		url         string
@@ -174,14 +282,14 @@ func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t
 	}{
 		{
 			name:        "unsupported content type",
-			url:         "/upload",
+			url:         "/v1/pipelines/reports/upload",
 			contentType: "application/zip",
 			body:        strings.NewReader("archive"),
 			wantStatus:  http.StatusUnsupportedMediaType,
 		},
 		{
 			name:        "invalid key syntax",
-			url:         "/upload",
+			url:         "/v1/pipelines/reports/upload",
 			contentType: "application/x-tar",
 			keyValues:   []string{"bad key"},
 			body:        strings.NewReader("archive"),
@@ -189,7 +297,7 @@ func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t
 		},
 		{
 			name:        "empty key",
-			url:         "/upload",
+			url:         "/v1/pipelines/reports/upload",
 			contentType: "application/x-tar",
 			keyValues:   []string{""},
 			body:        strings.NewReader("archive"),
@@ -197,7 +305,7 @@ func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t
 		},
 		{
 			name:        "too long key",
-			url:         "/upload",
+			url:         "/v1/pipelines/reports/upload",
 			contentType: "application/x-tar",
 			keyValues:   []string{strings.Repeat("a", 129)},
 			body:        strings.NewReader("archive"),
@@ -205,19 +313,12 @@ func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t
 		},
 		{
 			name:        "multiple keys",
-			url:         "/upload",
+			url:         "/v1/pipelines/reports/upload",
 			contentType: "application/x-tar",
 			keyValues:   []string{"one", "two"},
 			body:        strings.NewReader("archive"),
 			wantStatus:  http.StatusBadRequest,
 		},
-		{
-			name:        "submitted pipeline id",
-			url:         "/upload?pipeline_id=reports",
-			contentType: "application/x-tar",
-			body:        strings.NewReader("archive"),
-			wantStatus:  http.StatusBadRequest,
-		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -228,7 +329,8 @@ func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t
 						return UploadRunRecord{}, nil
 					},
 				},
-				tokens: map[string]string{"valid-token": "reports"},
+				tokens:          map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")},
+				uploadPipelines: pipelineIDSet([]string{"reports"}),
 			}
 			recorder := httptest.NewRecorder()
 			request := httptest.NewRequest(http.MethodPost, tt.url, tt.body)
@@ -270,10 +372,11 @@ func TestUploadHTTPHandlerMapsSubmitErrors(t *testing.T) {
 						return UploadRunRecord{}, tt.err
 					},
 				},
-				tokens: map[string]string{"valid-token": "reports"},
+				tokens:          map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")},
+				uploadPipelines: pipelineIDSet([]string{"reports"}),
 			}
 			recorder := httptest.NewRecorder()
-			request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
+			request := httptest.NewRequest(http.MethodPost, "/v1/pipelines/reports/upload", strings.NewReader("archive"))
 			request.Header.Set("Authorization", "Bearer valid-token")
 			request.Header.Set("Content-Type", "application/x-tar")
 
@@ -306,7 +409,8 @@ func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) {
 				}, true
 			},
 		},
-		tokens: map[string]string{"valid-token": "reports"},
+		tokens:          map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")},
+		uploadPipelines: pipelineIDSet([]string{"reports"}),
 	}
 
 	recorder := httptest.NewRecorder()
@@ -362,7 +466,6 @@ func uploadHTTPTestConfig() config.Config {
 			Source: config.Backend{
 				Backend: config.BackendHTTPUpload,
 				Upload: config.HTTPUpload{
-					TokenEnv:      "UPLOAD_TOKEN",
 					StagingPath:   "/tmp/distributor-test/reports",
 					MaxUploadSize: &size,
 				},
@@ -374,6 +477,11 @@ func uploadHTTPTestConfig() config.Config {
 				Publish: &config.PublishPolicy{Source: true},
 			}},
 		}},
+		UploadTokens: []config.UploadToken{{
+			ID:             "reporter",
+			TokenEnv:       "UPLOAD_TOKEN",
+			AllowPipelines: []string{"reports"},
+		}},
 	}
 	config.ApplyDefaults(&cfg)
 	return cfg
@@ -384,3 +492,11 @@ func uploadHTTPTestEnvironment(values map[string]string) config.Environment {
 		return "", false
 	})
 }
+
+func uploadHTTPTestToken(id, value string, pipelines ...string) resolvedUploadToken {
+	return resolvedUploadToken{
+		ID:               id,
+		Value:            value,
+		AllowedPipelines: pipelineIDSet(pipelines),
+	}
+}

internal/config/backend_view_test.go

@@ -53,12 +53,15 @@ func TestBackendViewValidationKeepsHTTPUploadSourceOnly(t *testing.T) {
 		ID: "reports",
 		Source: Backend{
 			Backend: BackendHTTPUpload,
-			Upload:  HTTPUpload{TokenEnv: "UPLOAD_TOKEN"},
 		},
 		Destinations: []Destination{{
 			ID:      "archive",
 			Backend: BackendHTTPUpload,
 		}},
+	}}, UploadTokens: []UploadToken{{
+		ID:             "reporter",
+		TokenEnv:       "UPLOAD_TOKEN",
+		AllowPipelines: []string{"reports"},
 	}}}
 	ApplyDefaults(&cfg)
 

internal/config/config.go

@@ -3,6 +3,7 @@ package config
 type Config struct {
 	Server       Server        `yaml:"server"`
 	Secrets      Secrets       `yaml:"secrets"`
+	UploadTokens []UploadToken `yaml:"upload_tokens"`
 	Pipelines    []Pipeline    `yaml:"pipelines"`
 }
 
@@ -23,6 +24,12 @@ type Secrets struct {
 	Directory string `yaml:"directory"`
 }
 
+type UploadToken struct {
+	ID             string   `yaml:"id"`
+	TokenEnv       string   `yaml:"token_env"`
+	AllowPipelines []string `yaml:"allow_pipelines"`
+}
+
 type Pipeline struct {
 	ID           string           `yaml:"id"`
 	Source       Backend          `yaml:"source"`
@@ -68,7 +75,6 @@ type Backend struct {
 }
 
 type HTTPUpload struct {
-	TokenEnv      string    `yaml:"token_env"`
 	StagingPath   string    `yaml:"staging_path"`
 	MaxUploadSize *ByteSize `yaml:"max_upload_size"`
 }

internal/config/load_test.go

@@ -256,13 +256,17 @@ pipelines:
   - id: weather-daily
     source:
       backend: http_upload
-      token_env: WEATHER_DAILY_UPLOAD_TOKEN
       staging_path: /srv/distributor/staging/weather-daily
       max_upload_size: 32MB
     destinations:
       - id: archive
         backend: local
         path: /archive
+upload_tokens:
+  - id: weather-reporter
+    token_env: WEATHER_DAILY_UPLOAD_TOKEN
+    allow_pipelines:
+      - weather-daily
 `)
 
 	server := cfg.Server.HTTP
@@ -289,9 +293,6 @@ pipelines:
 	if got, want := source.Backend, BackendHTTPUpload; got != want {
 		t.Fatalf("source.backend = %q, want %q", got, want)
 	}
-	if got, want := source.Upload.TokenEnv, "WEATHER_DAILY_UPLOAD_TOKEN"; got != want {
-		t.Fatalf("source.token_env = %q, want %q", got, want)
-	}
 	if got, want := source.Upload.StagingPath, "/srv/distributor/staging/weather-daily"; got != want {
 		t.Fatalf("source.staging_path = %q, want %q", got, want)
 	}
@@ -309,11 +310,15 @@ pipelines:
   - id: weather-daily
     source:
       backend: http_upload
-      token_env: WEATHER_DAILY_UPLOAD_TOKEN
     destinations:
       - id: archive
         backend: local
         path: /archive
+upload_tokens:
+  - id: weather-reporter
+    token_env: WEATHER_DAILY_UPLOAD_TOKEN
+    allow_pipelines:
+      - weather-daily
 `)
 
 	source := cfg.Pipelines[0].Source
@@ -325,6 +330,120 @@ pipelines:
 	}
 }
 
+func TestLoadFileAcceptsHTTPUploadTokens(t *testing.T) {
+	tests := map[string]string{
+		"valid multi pipeline token": `
+pipelines:
+  - id: weather-daily
+    source:
+      backend: http_upload
+    destinations:
+      - id: archive
+        backend: local
+        path: /archive/weather
+  - id: calendar-daily
+    source:
+      backend: http_upload
+    destinations:
+      - id: archive
+        backend: local
+        path: /archive/calendar
+upload_tokens:
+  - id: reporter
+    token_env: REPORTER_UPLOAD_TOKEN
+    allow_pipelines:
+      - weather-daily
+      - calendar-daily
+`,
+		"multiple tokens for one pipeline": `
+pipelines:
+  - id: reports
+    source:
+      backend: http_upload
+    destinations:
+      - id: archive
+        backend: local
+        path: /archive
+upload_tokens:
+  - id: reporter-a
+    token_env: REPORTER_A_UPLOAD_TOKEN
+    allow_pipelines:
+      - reports
+  - id: reporter-b
+    token_env: REPORTER_B_UPLOAD_TOKEN
+    allow_pipelines:
+      - reports
+`,
+	}
+	for name, body := range tests {
+		t.Run(name, func(t *testing.T) {
+			loadConfig(t, body)
+		})
+	}
+}
+
+func TestLoadFileRejectsInvalidUploadTokens(t *testing.T) {
+	tests := map[string]struct {
+		body string
+		want string
+	}{
+		"missing token list": {
+			body: `pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+			want: "upload_tokens is required",
+		},
+		"duplicate token ids": {
+			body: `upload_tokens: [{id: reporter, token_env: ONE_UPLOAD_TOKEN, allow_pipelines: [reports]}, {id: reporter, token_env: TWO_UPLOAD_TOKEN, allow_pipelines: [reports]}]
+pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+			want: "upload token id reporter is duplicated",
+		},
+		"duplicate allowlist entries": {
+			body: `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports, reports]}]
+pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+			want: "allow_pipelines contains duplicate pipeline id reports",
+		},
+		"unknown allowed pipeline id": {
+			body: `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [missing]}]
+pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+			want: "references unknown pipeline missing",
+		},
+		"non upload allowed pipeline id": {
+			body: `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}, {id: uploader, token_env: OTHER_UPLOAD_TOKEN, allow_pipelines: [upload]}]
+pipelines: [{id: reports, source: {backend: local, path: /source}, destinations: [{id: archive, backend: local, path: /archive}]}, {id: upload, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive-upload}]}]`,
+			want: "references non-http_upload pipeline reports",
+		},
+		"upload pipeline not allowed": {
+			body: `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}]
+pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}, {id: other, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive-other}]}]`,
+			want: "http_upload pipeline other is not allowed by any upload token",
+		},
+		"missing token id": {
+			body: `upload_tokens: [{token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}]
+pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+			want: "upload_tokens[0].id is required",
+		},
+		"invalid token id": {
+			body: `upload_tokens: [{id: ".reporter", token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}]
+pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+			want: "upload_tokens[0].id must be a slug-like identifier",
+		},
+		"missing token env": {
+			body: `upload_tokens: [{id: reporter, allow_pipelines: [reports]}]
+pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+			want: "upload_tokens[0].token_env is required",
+		},
+		"missing allowlist": {
+			body: `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN}]
+pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+			want: "upload_tokens[0].allow_pipelines is required",
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			assertLoadError(t, tt.body, tt.want)
+		})
+	}
+}
+
 func TestLoadFileValidBackendConfigs(t *testing.T) {
 	tests := map[string]string{
 		"local": `
@@ -516,15 +635,21 @@ func TestLoadFileRejectsInvalidS3Config(t *testing.T) {
 func TestLoadFileRejectsInvalidHTTPUploadConfig(t *testing.T) {
 	tests := map[string]string{
 		"server size": `server: {http: {max_upload_size: 20XB}}`,
-		"source size":             `pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN, max_upload_size: 20XB}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+		"source size": `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}]
-		"zero source size":        `pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN, max_upload_size: 0B}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+pipelines: [{id: reports, source: {backend: http_upload, max_upload_size: 20XB}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+		"zero source size": `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}]
+pipelines: [{id: reports, source: {backend: http_upload, max_upload_size: 0B}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
 		"server duration":         `server: {http: {retention: forever}}`,
 		"zero server duration":    `server: {http: {retention: 0s}}`,
-		"missing token env":       `pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+		"missing upload tokens":   `pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
 		"destination http upload": `pipelines: [{id: reports, source: {backend: local, path: /source}, destinations: [{id: ingest, backend: http_upload}]}]`,
-		"literal token":           `pipelines: [{id: reports, source: {backend: http_upload, token: secret, token_env: UPLOAD_TOKEN}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+		"literal token": `upload_tokens: [{id: reporter, token: secret, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}]
+pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
 		"unknown server field": `server: {http: {surprise: true}}`,
-		"unknown source field":    `pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN, surprise: true}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+		"legacy source token env": `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}]
+pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
+		"unknown source field": `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}]
+pipelines: [{id: reports, source: {backend: http_upload, surprise: true}, destinations: [{id: archive, backend: local, path: /archive}]}]`,
 	}
 	for name, body := range tests {
 		t.Run(name, func(t *testing.T) {

internal/config/validate.go

@@ -10,6 +10,10 @@ import (
 
 var idPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]*$`)
 
+func IsSlugLikeID(value string) bool {
+	return idPattern.MatchString(value)
+}
+
 type ValidationErrors []string
 
 func (e ValidationErrors) Error() string {
@@ -29,11 +33,12 @@ func Validate(cfg Config) error {
 	}
 
 	pipelineIDs := make(map[string]struct{}, len(cfg.Pipelines))
+	uploadPipelineIDs := make(map[string]struct{})
 	for pipelineIndex, pipeline := range cfg.Pipelines {
 		pipelineContext := fmt.Sprintf("pipelines[%d]", pipelineIndex)
 		if pipeline.ID == "" {
 			errs = append(errs, pipelineContext+".id is required")
-		} else if !idPattern.MatchString(pipeline.ID) {
+		} else if !IsSlugLikeID(pipeline.ID) {
 			errs = append(errs, pipelineContext+".id must be a slug-like identifier")
 		} else if _, exists := pipelineIDs[pipeline.ID]; exists {
 			errs = append(errs, "pipeline id "+pipeline.ID+" is duplicated")
@@ -42,6 +47,9 @@ func Validate(cfg Config) error {
 		}
 
 		errs = validateSourceBackend(errs, pipelineContext+".source", pipeline.Source)
+		if pipeline.Source.Backend == BackendHTTPUpload && pipeline.ID != "" {
+			uploadPipelineIDs[pipeline.ID] = struct{}{}
+		}
 		errs = validateValidationPolicy(errs, pipelineContext+".validation", pipeline.Validation)
 		if len(pipeline.Destinations) == 0 {
 			errs = append(errs, pipelineContext+".destinations is required")
@@ -52,7 +60,7 @@ func Validate(cfg Config) error {
 			destinationContext := fmt.Sprintf("%s.destinations[%d]", pipelineContext, destinationIndex)
 			if destination.ID == "" {
 				errs = append(errs, destinationContext+".id is required")
-			} else if !idPattern.MatchString(destination.ID) {
+			} else if !IsSlugLikeID(destination.ID) {
 				errs = append(errs, destinationContext+".id must be a slug-like identifier")
 			} else if _, exists := destinationIDs[destination.ID]; exists {
 				errs = append(errs, "destination id "+destination.ID+" is duplicated in pipeline "+pipeline.ID)
@@ -68,6 +76,8 @@ func Validate(cfg Config) error {
 		}
 	}
 
+	errs = validateUploadTokens(errs, cfg.UploadTokens, pipelineIDs, uploadPipelineIDs)
+
 	if len(errs) > 0 {
 		return errs
 	}
@@ -112,9 +122,6 @@ func validateDestinationBackend(errs ValidationErrors, context string, destinati
 }
 
 func validateHTTPUploadSource(errs ValidationErrors, context string, upload HTTPUpload) ValidationErrors {
-	if upload.TokenEnv == "" {
-		errs = append(errs, context+".token_env is required for http_upload backend")
-	}
 	if upload.StagingPath == "" {
 		errs = append(errs, context+".staging_path is required for http_upload backend")
 	}
@@ -124,6 +131,70 @@ func validateHTTPUploadSource(errs ValidationErrors, context string, upload HTTP
 	return errs
 }
 
+func validateUploadTokens(errs ValidationErrors, tokens []UploadToken, pipelineIDs, uploadPipelineIDs map[string]struct{}) ValidationErrors {
+	if len(uploadPipelineIDs) == 0 {
+		if len(tokens) > 0 {
+			errs = append(errs, "upload_tokens must reference configured http_upload pipelines")
+		}
+		return errs
+	}
+	if len(tokens) == 0 {
+		return append(errs, "upload_tokens is required when any pipeline source backend is http_upload")
+	}
+
+	tokenIDs := make(map[string]struct{}, len(tokens))
+	allowedUploadPipelineIDs := make(map[string]struct{}, len(uploadPipelineIDs))
+	for tokenIndex, token := range tokens {
+		context := fmt.Sprintf("upload_tokens[%d]", tokenIndex)
+		if token.ID == "" {
+			errs = append(errs, context+".id is required")
+		} else if !IsSlugLikeID(token.ID) {
+			errs = append(errs, context+".id must be a slug-like identifier")
+		} else if _, exists := tokenIDs[token.ID]; exists {
+			errs = append(errs, "upload token id "+token.ID+" is duplicated")
+		} else {
+			tokenIDs[token.ID] = struct{}{}
+		}
+
+		if token.TokenEnv == "" {
+			errs = append(errs, context+".token_env is required")
+		}
+		if len(token.AllowPipelines) == 0 {
+			errs = append(errs, context+".allow_pipelines is required")
+		}
+
+		seenAllowed := make(map[string]struct{}, len(token.AllowPipelines))
+		for allowIndex, pipelineID := range token.AllowPipelines {
+			allowContext := fmt.Sprintf("%s.allow_pipelines[%d]", context, allowIndex)
+			if pipelineID == "" {
+				errs = append(errs, allowContext+" is required")
+				continue
+			}
+			if _, exists := seenAllowed[pipelineID]; exists {
+				errs = append(errs, context+".allow_pipelines contains duplicate pipeline id "+pipelineID)
+				continue
+			}
+			seenAllowed[pipelineID] = struct{}{}
+			if _, exists := pipelineIDs[pipelineID]; !exists {
+				errs = append(errs, allowContext+" references unknown pipeline "+pipelineID)
+				continue
+			}
+			if _, exists := uploadPipelineIDs[pipelineID]; !exists {
+				errs = append(errs, allowContext+" references non-http_upload pipeline "+pipelineID)
+				continue
+			}
+			allowedUploadPipelineIDs[pipelineID] = struct{}{}
+		}
+	}
+
+	for pipelineID := range uploadPipelineIDs {
+		if _, exists := allowedUploadPipelineIDs[pipelineID]; !exists {
+			errs = append(errs, "http_upload pipeline "+pipelineID+" is not allowed by any upload token")
+		}
+	}
+	return errs
+}
+
 func validateBackend(errs ValidationErrors, context string, backend backendView) ValidationErrors {
 	switch backend.Backend {
 	case "":

pkg/upload/client.go

@@ -15,6 +15,7 @@ import (
 	"os"
 	"path"
 	"path/filepath"
+	"regexp"
 	"strings"
 	"time"
 
@@ -22,7 +23,6 @@ import (
 )
 
 const (
-	uploadPath            = "upload"
 	runsPath              = "runs"
 	idempotencyKeyHeader  = "Idempotency-Key"
 	defaultHTTPTimeout    = 30 * time.Second
@@ -34,6 +34,8 @@ const (
 	redactedSecret        = "[redacted]"
 )
 
+var pipelineIDPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]*$`)
+
 func NewClient(opts ClientOptions) (*Client, error) {
 	endpoint, err := cleanEndpoint(opts.Endpoint)
 	if err != nil {
@@ -65,6 +67,9 @@ func (c *Client) UploadBundle(ctx context.Context, opts UploadBundleOptions) (Re
 	if opts.Validate && opts.DisableValidation {
 		return Result{}, fmt.Errorf("validate and disable validation cannot both be set")
 	}
+	if err := validatePipelineID(opts.PipelineID); err != nil {
+		return Result{}, err
+	}
 	if opts.Root == "" {
 		return Result{}, fmt.Errorf("root is required")
 	}
@@ -85,7 +90,7 @@ func (c *Client) UploadBundle(ctx context.Context, opts UploadBundleOptions) (Re
 	if err != nil {
 		return Result{}, err
 	}
-	return c.uploadArchive(ctx, archive, key)
+	return c.uploadArchive(ctx, opts.PipelineID, archive, key)
 }
 
 func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Result, error) {
@@ -95,6 +100,9 @@ func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Resu
 	if opts.Validate && opts.DisableValidation {
 		return Result{}, fmt.Errorf("validate and disable validation cannot both be set")
 	}
+	if err := validatePipelineID(opts.PipelineID); err != nil {
+		return Result{}, err
+	}
 	if opts.ID == "" {
 		return Result{}, fmt.Errorf("id is required")
 	}
@@ -131,7 +139,7 @@ func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Resu
 	if err != nil {
 		return Result{}, err
 	}
-	return c.uploadArchive(ctx, archive, key)
+	return c.uploadArchive(ctx, opts.PipelineID, archive, key)
 }
 
 func (c *Client) Status(ctx context.Context, runID string) (RunStatus, error) {
@@ -167,7 +175,7 @@ func (c *Client) Status(ctx context.Context, runID string) (RunStatus, error) {
 	return status, nil
 }
 
-func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyKey string) (Result, error) {
+func (c *Client) uploadArchive(ctx context.Context, pipelineID string, archive []byte, idempotencyKey string) (Result, error) {
 	if ctx == nil {
 		ctx = context.Background()
 	}
@@ -176,7 +184,7 @@ func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyK
 		if err := ctx.Err(); err != nil {
 			return Result{}, err
 		}
-		result, retry, err := c.uploadAttempt(ctx, archive, idempotencyKey)
+		result, retry, err := c.uploadAttempt(ctx, pipelineID, archive, idempotencyKey)
 		if err == nil {
 			return result, nil
 		}
@@ -191,8 +199,8 @@ func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyK
 	return Result{}, lastErr
 }
 
-func (c *Client) uploadAttempt(ctx context.Context, archive []byte, idempotencyKey string) (Result, bool, error) {
+func (c *Client) uploadAttempt(ctx context.Context, pipelineID string, archive []byte, idempotencyKey string) (Result, bool, error) {
-	request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.uploadURL(), bytes.NewReader(archive))
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.uploadURL(pipelineID), bytes.NewReader(archive))
 	if err != nil {
 		return Result{}, false, c.redactError(err)
 	}
@@ -227,8 +235,8 @@ func (c *Client) authorize(request *http.Request) {
 	request.Header.Set("Authorization", authorizationPrefix+c.token)
 }
 
-func (c *Client) uploadURL() string {
+func (c *Client) uploadURL(pipelineID string) string {
-	return joinEndpointPath(c.endpoint, uploadPath)
+	return joinEndpointPath(c.endpoint, "v1", "pipelines", pipelineID, "upload")
 }
 
 func (c *Client) statusURL(runID string) string {
@@ -343,6 +351,16 @@ func uploadIdempotencyKey(value string) (string, error) {
 	return value, nil
 }
 
+func validatePipelineID(value string) error {
+	if value == "" {
+		return fmt.Errorf("pipeline id is required")
+	}
+	if !pipelineIDPattern.MatchString(value) {
+		return fmt.Errorf("pipeline id must be a slug-like identifier")
+	}
+	return nil
+}
+
 func validateIdempotencyKey(value string) error {
 	if value == "" {
 		return fmt.Errorf("idempotency key is required")

pkg/upload/client_test.go

@@ -47,7 +47,7 @@ func TestNewClientValidatesOptions(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewClient() error = %v", err)
 	}
-	if got, want := client.uploadURL(), "http://127.0.0.1:8080/base/upload"; got != want {
+	if got, want := client.uploadURL("reports.daily"), "http://127.0.0.1:8080/base/v1/pipelines/reports.daily/upload"; got != want {
 		t.Fatalf("upload URL = %q, want %q", got, want)
 	}
 	if client.httpClient == nil || client.httpClient.Timeout == 0 {
@@ -65,7 +65,7 @@ func TestUploadBundleSendsCallerKeyAndManifestArchive(t *testing.T) {
 	}
 
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if got, want := r.URL.Path, "/upload"; got != want {
+		if got, want := r.URL.Path, "/v1/pipelines/reports.daily/upload"; got != want {
 			t.Fatalf("path = %q, want %q", got, want)
 		}
 		if got, want := r.Header.Get("Authorization"), "Bearer secret-token"; got != want {
@@ -93,6 +93,7 @@ func TestUploadBundleSendsCallerKeyAndManifestArchive(t *testing.T) {
 		t.Fatalf("NewClient() error = %v", err)
 	}
 	result, err := client.UploadBundle(context.Background(), UploadBundleOptions{
+		PipelineID:     "reports.daily",
 		Root:           root,
 		IdempotencyKey: "producer.retry:one",
 	})
@@ -112,6 +113,9 @@ func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) {
 	}
 	tempDir := t.TempDir()
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if got, want := r.URL.Path, "/v1/pipelines/reports.files/upload"; got != want {
+			t.Fatalf("path = %q, want %q", got, want)
+		}
 		entries := readArchiveEntries(t, r.Body)
 		if got := string(entries["manifest.json"]); !strings.Contains(got, `"id": "reports.from.files"`) {
 			t.Fatalf("manifest = %s, want uploaded id", got)
@@ -131,6 +135,7 @@ func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) {
 		t.Fatalf("NewClient() error = %v", err)
 	}
 	_, err = client.UploadFiles(context.Background(), UploadFilesOptions{
+		PipelineID: "reports.files",
 		ID:         "reports.from.files",
 		Files: []sourcebundle.BundleFile{{
 			SourcePath: sourcePath,
@@ -153,6 +158,77 @@ func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) {
 	}
 }
 
+func TestUploadMethodsRequirePipelineIDBeforeLocalWork(t *testing.T) {
+	var requests atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		t.Fatal("server should not receive request")
+	}))
+	defer server.Close()
+
+	client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+
+	missingRoot := filepath.Join(t.TempDir(), "missing")
+	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: missingRoot}); err == nil || !strings.Contains(err.Error(), "pipeline id is required") {
+		t.Fatalf("UploadBundle() error = %v, want missing pipeline id", err)
+	}
+
+	tempDir := t.TempDir()
+	sourcePath := filepath.Join(t.TempDir(), "report.md")
+	if err := os.WriteFile(sourcePath, []byte("data"), 0o600); err != nil {
+		t.Fatalf("write source: %v", err)
+	}
+	if _, err := client.UploadFiles(context.Background(), UploadFilesOptions{
+		ID: "reports.from.files",
+		Files: []sourcebundle.BundleFile{{
+			SourcePath: sourcePath,
+			Path:       "report.md",
+		}},
+		TempDir: tempDir,
+	}); err == nil || !strings.Contains(err.Error(), "pipeline id is required") {
+		t.Fatalf("UploadFiles() error = %v, want missing pipeline id", err)
+	}
+	entries, err := os.ReadDir(tempDir)
+	if err != nil {
+		t.Fatalf("read temp dir: %v", err)
+	}
+	if len(entries) != 0 {
+		t.Fatalf("temp dir entries = %d, want no local bundle work", len(entries))
+	}
+	if got := requests.Load(); got != 0 {
+		t.Fatalf("requests = %d, want 0", got)
+	}
+}
+
+func TestUploadMethodsRejectInvalidPipelineIDBeforeHTTPRequest(t *testing.T) {
+	root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}})
+	var requests atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		t.Fatal("server should not receive request")
+	}))
+	defer server.Close()
+
+	client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	for _, pipelineID := range []string{".reports", "reports/daily", "reports daily"} {
+		t.Run(pipelineID, func(t *testing.T) {
+			_, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: pipelineID, Root: root})
+			if err == nil || !strings.Contains(err.Error(), "pipeline id must be a slug-like identifier") {
+				t.Fatalf("UploadBundle() error = %v, want invalid pipeline id", err)
+			}
+		})
+	}
+	if got := requests.Load(); got != 0 {
+		t.Fatalf("requests = %d, want 0", got)
+	}
+}
+
 func TestUploadBundleValidationFailurePreventsHTTPRequest(t *testing.T) {
 	root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "original"}})
 	if err := os.WriteFile(filepath.Join(root, "report.md"), []byte("changed"), 0o600); err != nil {
@@ -169,7 +245,7 @@ func TestUploadBundleValidationFailurePreventsHTTPRequest(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewClient() error = %v", err)
 	}
-	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root}); err == nil {
+	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root}); err == nil {
 		t.Fatal("UploadBundle() error = nil, want validation error")
 	}
 	if got := requests.Load(); got != 0 {
@@ -193,7 +269,7 @@ func TestUploadBundleCanDisableLocalValidation(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewClient() error = %v", err)
 	}
-	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, DisableValidation: true}); err != nil {
+	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, DisableValidation: true}); err != nil {
 		t.Fatalf("UploadBundle() error = %v", err)
 	}
 	if got := requests.Load(); got != 1 {
@@ -206,6 +282,9 @@ func TestGeneratedIdempotencyKeyIsReusedAcrossRetry(t *testing.T) {
 	var attempts atomic.Int64
 	var keys []string
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if got, want := r.URL.Path, "/v1/pipelines/reports/upload"; got != want {
+			t.Fatalf("path = %q, want %q", got, want)
+		}
 		keys = append(keys, r.Header.Get(idempotencyKeyHeader))
 		if attempts.Add(1) == 1 {
 			writeJSONError(w, http.StatusServiceUnavailable, "busy", false)
@@ -224,7 +303,7 @@ func TestGeneratedIdempotencyKeyIsReusedAcrossRetry(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewClient() error = %v", err)
 	}
-	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root}); err != nil {
+	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root}); err != nil {
 		t.Fatalf("UploadBundle() error = %v", err)
 	}
 	if got, want := attempts.Load(), int64(2); got != want {
@@ -276,7 +355,7 @@ func TestUploadResponseParsingAndNoRetryStatuses(t *testing.T) {
 			if err != nil {
 				t.Fatalf("NewClient() error = %v", err)
 			}
-			_, err = client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "key"})
+			_, err = client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "key"})
 			if err == nil {
 				t.Fatal("UploadBundle() error = nil, want error")
 			}
@@ -309,7 +388,7 @@ func TestTokenRedactedFromHTTPError(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewClient() error = %v", err)
 	}
-	_, err = client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "key"})
+	_, err = client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "key"})
 	if err == nil {
 		t.Fatal("UploadBundle() error = nil, want error")
 	}
@@ -330,6 +409,9 @@ func TestNetworkRetryUsesSameIdempotencyKey(t *testing.T) {
 		Token:    "secret",
 		HTTPClient: &http.Client{Transport: roundTripFunc(func(request *http.Request) (*http.Response, error) {
 			keys = append(keys, request.Header.Get(idempotencyKeyHeader))
+			if got, want := request.URL.Path, "/v1/pipelines/reports/upload"; got != want {
+				t.Fatalf("path = %q, want %q", got, want)
+			}
 			if attempts.Add(1) == 1 {
 				return nil, temporaryNetworkError{}
 			}
@@ -346,7 +428,7 @@ func TestNetworkRetryUsesSameIdempotencyKey(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewClient() error = %v", err)
 	}
-	result, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "network-retry"})
+	result, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "network-retry"})
 	if err != nil {
 		t.Fatalf("UploadBundle() error = %v", err)
 	}
@@ -381,7 +463,7 @@ func TestContextCancellationDuringRetryBackoff(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewClient() error = %v", err)
 	}
-	_, err = client.UploadBundle(ctx, UploadBundleOptions{Root: root, IdempotencyKey: "cancel"})
+	_, err = client.UploadBundle(ctx, UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "cancel"})
 	if !errors.Is(err, context.Canceled) {
 		t.Fatalf("UploadBundle() error = %v, want context.Canceled", err)
 	}
@@ -446,7 +528,7 @@ func TestInvalidCallerIdempotencyKeyPreventsHTTPRequest(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewClient() error = %v", err)
 	}
-	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "bad key"}); err == nil {
+	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{PipelineID: "reports", Root: root, IdempotencyKey: "bad key"}); err == nil {
 		t.Fatal("UploadBundle() error = nil, want invalid key error")
 	}
 	if got := requests.Load(); got != 0 {

pkg/upload/doc.go

@@ -12,9 +12,10 @@
 //
 // NewClient creates a Client from ClientOptions. Endpoint is required and must
 // be an http or https distributor server base URL without userinfo, query, or
-// fragment. The client derives /upload for submissions and /runs/<run-id> for
+// fragment. The client derives /v1/pipelines/<pipeline-id>/upload for
-// status checks. Token is required and is sent as Authorization: Bearer <token>.
+// submissions and /runs/<run-id> for status checks. Token is required and is
-// Token values are redacted from errors produced by the client.
+// sent as Authorization: Bearer <token>. Token values are redacted from errors
+// produced by the client.
 //
 // HTTPClient is optional. When omitted, the package uses a client with a
 // conservative timeout. Retry is optional; zero values select safe defaults.
@@ -23,20 +24,28 @@
 //
 // # Upload Workflows
 //
-// UploadBundle uploads an existing local source bundle root. The root must
+// UploadBundle uploads an existing local source bundle root to the configured
-// contain manifest.json. By default, UploadBundle loads the manifest and
+// PipelineID. PipelineID is required and must match the server's slug-like
-// validates the complete local bundle with pkg/bundle before making any HTTP
+// pipeline id syntax. The root must contain manifest.json. By default,
-// request. The generated gzip-compressed tar archive contains manifest.json and
+// UploadBundle loads the manifest and validates the complete local bundle with
-// exactly the manifest-listed files; unlisted files are not uploaded.
+// pkg/bundle before making any HTTP request. The generated gzip-compressed tar
+// archive contains manifest.json and exactly the manifest-listed files;
+// unlisted files are not uploaded.
 //
 // UploadFiles is the convenience workflow for producer applications that have
-// generated files but have not yet assembled a bundle directory. It uses
+// generated files but have not yet assembled a bundle directory. PipelineID is
+// required and selects the configured distributor workflow. UploadFiles uses
 // pkg/bundle to create a temporary complete bundle from explicit
 // bundle.BundleFile values, validates it by default, archives it, uploads it,
 // and removes temporary files when the call returns. UploadFiles does not write
 // into producer source directories. A zero Created timestamp follows
 // pkg/bundle defaulting behavior.
 //
+// The producer contract has four separate identifiers: the bearer token
+// authenticates the client, PipelineID selects the distributor workflow, the
+// source manifest ID identifies the logical artifact within that workflow, and
+// IdempotencyKey identifies one producer run and retry group.
+//
 // Validation is enabled by default. Set DisableValidation when the application
 // has already performed equivalent local validation and wants to skip the
 // package's validation step. Validate and DisableValidation must not both be
@@ -91,6 +100,7 @@
 //	}
 //
 //	result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{
+//		PipelineID:     "reports.daily",
 //		ID:             "reports.daily.2026-06-06",
 //		IdempotencyKey: "reports.daily.2026-06-06",
 //		Files: []bundle.BundleFile{
@@ -115,6 +125,7 @@
 // Example: upload an existing bundle root.
 //
 //	result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{
+//		PipelineID:     "reports.daily",
 //		Root:           "/var/lib/reports/daily-2026-06-06",
 //		IdempotencyKey: "reports.daily.2026-06-06",
 //	})

pkg/upload/types.go

@@ -30,6 +30,7 @@ type RetryOptions struct {
 }
 
 type UploadBundleOptions struct {
+	PipelineID        string
 	Root              string
 	Validate          bool
 	DisableValidation bool
@@ -37,6 +38,7 @@ type UploadBundleOptions struct {
 }
 
 type UploadFilesOptions struct {
+	PipelineID        string
 	ID                string
 	Created           time.Time
 	Files             []bundle.BundleFile