Document pipeline-scoped upload behavior

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`: