Consolidate future roadmap documentation

README.md

@@ -10,7 +10,7 @@ Run the maintained local example:
 go run ./cmd/distributor run --config examples/local-publish.yml
 ```
 
-Go producers can use `gitea.maximumdirect.net/eric/distributor/pkg/bundle` to build, write, parse, and validate local source bundles with the same manifest contract used by the CLI. See [Source bundle contract](docs/integrations/source-bundle.md).
+Go producers can use `gitea.maximumdirect.net/eric/distributor/pkg/bundle` to build, write, parse, and validate local source bundles with the same manifest contract used by the CLI. They can use `gitea.maximumdirect.net/eric/distributor/pkg/upload` to build or validate a bundle and submit it to `distributor serve` with bearer authentication and idempotency keys. See [Source bundle contract](docs/integrations/source-bundle.md) and [HTTP upload contract](docs/integrations/http-upload.md).
 
 - [CLI reference](docs/cli.md)
 - [Configuration reference](docs/config.md)

docs/integrations/http-upload.md

@@ -30,6 +30,14 @@ Returns `200 OK` when the server is running:
 
 Accepts one source bundle archive and returns after the archive is staged and validated.
 
+Producers may include:
+
+```text
+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`.
+
 Accepted content types:
 
 - `application/x-tar`
@@ -44,8 +52,9 @@ Successful admission returns `202 Accepted`:
 
 Common error responses:
 
-- `400`: pipeline query supplied, archive rejected, malformed archive, or invalid staged source bundle.
+- `400`: pipeline query supplied, invalid idempotency key, archive rejected, malformed archive, or invalid staged source bundle.
 - `401`: missing, empty, or unknown bearer token.
+- `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.
 - `503`: upload queue is full.
@@ -56,6 +65,14 @@ Error bodies use:
 {"error":"<message>"}
 ```
 
+Retryable idempotency conflicts include:
+
+```json
+{"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`.
+
 ### `GET /runs/<run-id>`
 
 Returns an in-memory status record while retained:
@@ -89,12 +106,36 @@ Archive entry rules:
 
 The uploaded archive size and extracted bundle size are bounded by the selected pipeline's `source.max_upload_size`. Extracted file count is also bounded by the implementation.
 
+## Go Producer Helper
+
+Go producers can use `gitea.maximumdirect.net/eric/distributor/pkg/upload` to build or validate source bundles, package them as gzip-compressed tar archives, and submit them to this API:
+
+```go
+client, err := upload.NewClient(upload.ClientOptions{
+	Endpoint: "http://127.0.0.1:8080",
+	Token:    token,
+})
+if err != nil {
+	return err
+}
+result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{
+	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.
+
+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.
+
 ## Queue And Retention
 
 `server.http.queue_size` bounds accepted-but-not-started uploads plus uploads being staged. `server.http.max_concurrency` bounds publishing concurrency. The coordinator does not run two uploads for the same pipeline concurrently.
 
 Completed status records expire after `server.http.retention`; expiration removes committed staged bundle directories for completed uploads. Server restart clears queue state and status records.
 
+Idempotency records are memory-only, expire with completed upload status records, and are cleared by server restart.
+
 ## 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.
@@ -104,5 +145,5 @@ The HTTP API does not expose pipeline selection by request parameter, TLS, publi
 Before changing this contract, inspect and run:
 
 ```sh
-go test ./internal/app ./internal/ingest
+go test ./internal/app ./internal/ingest ./pkg/upload
 ```

docs/integrations/source-bundle.md

@@ -68,6 +68,8 @@ Go producers can use `gitea.maximumdirect.net/eric/distributor/pkg/bundle` to bu
 - `LoadManifest`, `ParseManifest`, `ValidateManifest`, and `ValidateBundle`: parse and validate local bundles.
 - `FileDigest`, `BundleDigest`, and `ValidateDigest`: digest helpers.
 
+Go producers that submit bundles to `distributor serve` can use `gitea.maximumdirect.net/eric/distributor/pkg/upload`. It builds on `pkg/bundle`, packages valid bundles as gzip-compressed tar uploads, sends bearer authentication, and includes idempotency keys for safe retry behavior. See [HTTP Upload API Contract](http-upload.md).
+
 CLI producers can use:
 
 ```sh
@@ -88,5 +90,5 @@ The source bundle manifest does not configure routing, destination selection, pu
 Before changing this contract, inspect and run:
 
 ```sh
-go test ./pkg/bundle ./internal/bundle
+go test ./pkg/bundle ./pkg/upload ./internal/bundle
 ```

docs/internal/app.md

@@ -34,19 +34,21 @@ The app layer registers default transforms, including Markdown-to-HTML, and supp
 
 Run workflows discover and validate source bundles through `internal/bundle`. Destination state actions are prepared and written through `internal/publish` and `internal/state`; the app layer records report projections of those actions and results.
 
-HTTP uploads stage and validate archives before enqueueing a pipeline run with a local staged source root.
+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.
 
 ## Skip And Resume Behavior
 
 Fan-out destinations are independent. A destination failure is recorded and does not prevent later destinations from being attempted. Dry-run builds plans and reports without destination writes, destination state writes, notifier calls, or SSH known-host persistence.
 
-HTTP upload status is in memory. Accepted jobs move through accepted, queued, running, succeeded, or failed states and expire after configured retention.
+HTTP upload status is in memory. Accepted jobs move through accepted, queued, running, succeeded, or failed states and expire after configured retention. Upload idempotency records are also memory-only, expire with the completed status record for their accepted run, and are cleared by process restart.
 
 ## Failure Behavior
 
 Runtime setup fails for config load, config validation, secret loading, or credential resolution errors. Source setup failures stop the affected run before destination planning. Destination open, planning, execution, and notification failures are recorded as destination failures where a partial result exists.
 
-HTTP upload startup fails if upload tokens are missing, empty, or duplicated. Upload requests can fail during authentication, content-type validation, queue admission, archive staging, source validation, or later publish execution.
+HTTP upload startup fails if upload tokens are missing, empty, or duplicated. Upload requests can fail during authentication, idempotency-key validation, content-type validation, idempotency conflict checks, queue admission, archive staging, source validation, or later publish execution.
 
 ## Tests To Inspect
 
@@ -63,4 +65,5 @@ HTTP upload startup fails if upload tokens are missing, empty, or duplicated. Up
 - Fan-out destinations remain independent after a destination-scoped failure.
 - Secret values are never printed; warnings may name variables only.
 - Upload admission stages and validates a bundle before returning a run id.
+- Idempotent upload retries compare normalized source manifest identity, not archive bytes.
 - Runtime backend registration remains app-owned.

docs/operations.md

@@ -138,6 +138,35 @@ curl -X POST http://127.0.0.1:8080/upload \
   --data-binary @bundle.tar.gz
 ```
 
+For safe producer retries, include an idempotency key that is stable for the producer operation:
+
+```sh
+curl -X POST http://127.0.0.1:8080/upload \
+  -H "Authorization: Bearer $DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN" \
+  -H "Content-Type: application/gzip" \
+  -H "Idempotency-Key: producer.run.20260604T120000Z" \
+  --data-binary @bundle.tar.gz
+```
+
+Go producer applications can use `pkg/upload` instead of constructing archives and HTTP requests directly. The package sends `Idempotency-Key` on every upload, derives `/upload` from the configured endpoint, and reuses the same key and replayable request body for safe retries:
+
+```go
+client, err := upload.NewClient(upload.ClientOptions{
+	Endpoint: "http://127.0.0.1:8080",
+	Token:    token,
+})
+result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{
+	Root:           "examples/source-bundle",
+	IdempotencyKey: "producer.run.20260604T120000Z",
+})
+```
+
+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 a retry must be stable across separate process runs.
+
+```sh
+go run ./examples/upload-client
+```
+
 Accepted uploads return after the archive is staged and validated:
 
 ```json
@@ -154,6 +183,8 @@ 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.
+
 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.
 
 The default bind address is private loopback. Put TLS, public routing, rate limiting, and external access policy in a reverse proxy or deployment layer.

docs/policy/development.md

@@ -7,6 +7,7 @@ Use it with `docs/policy/architecture.md` and `docs/policy/documentation.md`.
 
 - `cmd/distributor`: executable entrypoint only.
 - `pkg/bundle`: public producer-facing source manifest and local bundle writer helpers.
+- `pkg/upload`: public producer-facing HTTP upload client built on `pkg/bundle`.
 - `internal/app`: top-level use cases for `run`, `validate`, and `inspect`.
 - `internal/cli`: standard-library command parsing, flags, help text, and command wiring.
 - `internal/config`: YAML configuration structs, loading, defaults, and validation.
@@ -26,9 +27,9 @@ Use it with `docs/policy/architecture.md` and `docs/policy/documentation.md`.
 - `examples`: copyable example configs and source bundles.
 
 Do not create new top-level package families such as public `pkg/...` packages
-beyond `pkg/bundle`, generic workflow containers, or service-specific adapter
+beyond `pkg/bundle` and `pkg/upload`, generic workflow containers, or
-directories unless the architecture policy or a current roadmap explicitly
+service-specific adapter directories unless the architecture policy or a
-calls for them.
+current roadmap explicitly calls for them.
 
 ## Common Commands
 
@@ -45,6 +46,7 @@ go test ./internal/config
 go test ./internal/cli ./internal/app
 go test ./internal/publish ./internal/state
 go test ./internal/transform/markdown
+go test ./pkg/bundle ./pkg/upload
 ```
 
 Run the CLI against an example config:
@@ -76,6 +78,7 @@ GOCACHE=/private/tmp/distributor-gocache GOMODCACHE=/private/tmp/distributor-gom
 - Preserve public CLI behavior, config semantics, manifest schema, destination state schema, and implemented backend behavior unless the current task explicitly changes them.
 - Use `storage.DisplayPath`, `storage.StateFileName`, `storage.StatePath`, and `storage.ManagedBundleTargets` instead of duplicating those conventions.
 - Use `pkg/bundle` for normalized source manifest semantics. Internal packages should reach those rules through `internal/bundle` when they also need storage-backed bundle discovery or validation.
+- Keep `pkg/upload` as a producer-facing HTTP client. It should depend on `pkg/bundle` and standard HTTP/archive primitives, not on `internal/app`, `internal/ingest`, server config, storage backends, or destination state types.
 - Use `config.ValidatePublishTransformPolicy` for publish and transform policy combinations.
 - Do not import concrete transform implementations from `internal/publish`; app-level wiring owns transform registration.
 - Do not import `internal/testutil` from production code.
@@ -171,6 +174,7 @@ Test close to the behavior being changed:
 - Use `internal/app` and `internal/cli` tests for user-facing workflows.
 - Use `internal/testutil` for shared valid fixtures only; keep edge cases near the package under test.
 - Run `go test ./...` after cross-package changes or documentation/example changes tied to tests.
+- Run `go test ./pkg/bundle ./pkg/upload` after changing producer-facing bundle or upload APIs.
 
 Live integration tests must be opt-in and skipped during normal `go test ./...`
 unless their required environment variables are set. Test-only environment

docs/roadmap/future.md

@@ -0,0 +1,80 @@
+# Future Roadmap
+
+This document records planned or deferred work that is not part of the current
+implementation. Current behavior is documented outside roadmap files in the
+README, integration contracts, operations guide, troubleshooting guide, and
+internal docs.
+
+## Durability And Recovery
+
+- Durable upload status persistence across process restarts.
+- Durable idempotency records across server restarts.
+- Database-backed upload queueing.
+- Recovery semantics for queued or running uploads after a restart.
+- Durable producer retry processing.
+
+## Producer Client Workflows
+
+- Durable client queues or background producer workers.
+- `UploadAndWait` helper.
+- Long-polling helper or equivalent wait workflow.
+
+## Run Control APIs
+
+- Run retry endpoints.
+- Run cancellation endpoints.
+- Run listing endpoints.
+
+## Archive And Transport Protocols
+
+- Zstandard-compressed tar archives.
+- Additional content negotiation rules for future archive formats.
+- Multipart upload support.
+- Resumable upload support.
+- Streaming upload protocols.
+
+## Destination Backends
+
+- GitHub Gist destination backend support.
+- Authentication and secret handling for GitHub API tokens.
+- Gist file mapping, update, replacement, and conflict semantics.
+- Rate-limit handling and retry behavior for GitHub API responses.
+
+## Docker Image Support
+
+- Official container image build and release workflow.
+- Runtime filesystem layout for config, secrets, staging, and local outputs.
+- Container-oriented examples for `run` and `serve`.
+- Image tagging, versioning, and upgrade guidance.
+
+## Notifications And Hooks
+
+- Email notification support for completed, failed, or partially failed
+  distribution runs.
+- SMTP configuration, authentication, secret handling, and recipient policy.
+- ntfy notification support for completed, failed, or partially failed
+  distribution runs.
+- ntfy topic, server, token, priority, and action configuration.
+- General post-distribution hook support.
+- Hook payload contract that can pass run status, summaries, destination
+  outcomes, output metadata, and public links to external tools.
+- Local executable hook adapter with bounded arguments, environment, stdin,
+  timeout, exit-code handling, and secret-redaction behavior.
+
+## Authentication And Deployment Surface
+
+- URL-token authentication for constrained clients.
+- Additional token lifecycle tooling.
+- Mutual TLS or other in-app identity mechanisms.
+- In-app TLS.
+- Public exposure defaults.
+- Browser UI.
+
+## Boundaries
+
+- `http_upload` remains source-only unless a future implementation changes that
+  contract.
+- Current upload status, queue, and idempotency state are memory-only.
+- Producers submit complete tar or gzip-compressed tar source bundles today.
+- Public access policy, TLS termination, and rate limiting belong outside
+  `distributor` unless a future implementation changes that boundary.

docs/roadmap/http.md

@@ -1,61 +0,0 @@
-# Roadmap: HTTP Upload Extensions
-
-## Purpose
-
-The HTTP upload API is implemented. Current behavior is documented in:
-
-- [CLI](../cli.md)
-- [Configuration](../config.md)
-- [Operations](../operations.md)
-- [Troubleshooting](../troubleshooting.md)
-- [HTTP upload contract](../integrations/http-upload.md)
-- [Application internals](../internal/app.md)
-- [Ingestion internals](../internal/ingest.md)
-
-This roadmap records HTTP upload extensions that are intentionally not part of
-the current implementation.
-
-## Deferred Extensions
-
-### Authentication
-
-- URL-token authentication for constrained clients.
-- Additional token lifecycle tooling.
-- Mutual TLS or other in-app identity mechanisms.
-
-### Archive Formats
-
-- Zstandard-compressed tar archives.
-- Additional content negotiation rules for future archive formats.
-
-### Status And Queue Durability
-
-- Durable status persistence across process restarts.
-- Database-backed queueing.
-- Recovery semantics for queued or running uploads after a restart.
-
-### Producer Coordination
-
-- Producer-supplied idempotency keys.
-- Run retry endpoints.
-- Run cancellation endpoints.
-- Run listing endpoints.
-
-### Deployment Surface
-
-- In-app TLS.
-- Public exposure defaults.
-- Browser UI.
-
-## Boundaries
-
-Current HTTP upload behavior remains intentionally small:
-
-- `http_upload` is source-only and is not a durable storage backend.
-- Upload status is memory-only.
-- Producers submit complete tar or gzip-compressed tar source bundles.
-- Producers authenticate with `Authorization: Bearer <token>`.
-- Public access policy, TLS termination, and rate limiting belong outside
-  `distributor` unless a future roadmap explicitly changes that boundary.
-
-Do not document deferred extensions as available outside `docs/roadmap/`.

docs/roadmap/implementation.md

@@ -1,211 +0,0 @@
-# Producer HTTP Upload Package Implementation Roadmap
-
-## Current Baseline
-
-`distributor serve` and the `http_upload` source backend are implemented.
-Producers can already submit complete tar or gzip-compressed tar source bundles
-to `POST /upload` with bearer authentication, and each bearer token maps to one
-configured upload pipeline.
-
-The public `pkg/bundle` package already provides producer-side source manifest
-semantics, digest calculation, path validation, local manifest building, local
-bundle writing, and local bundle validation helpers.
-
-The current implementation does not have a server-side producer idempotency
-contract, and it does not provide a public `pkg/upload` helper package.
-
-Future behavior remains under `docs/roadmap/` until implemented. Do not update
-README, current user docs, examples, or current-behavior internal docs until the
-corresponding stage has been implemented.
-
-This active roadmap implements the accepted producer upload package plan in
-`docs/roadmap/producer.md`. It supersedes the older `docs/roadmap/http.md`
-deferred note for producer-supplied idempotency keys; producer idempotency is
-now active roadmap work.
-
-## Active Roadmap
-
-## Stage 1: Server-Side Upload Idempotency
-
-Goal:
-
-Add `Idempotency-Key` support to `POST /upload` so safe producer retries do not
-create duplicate accepted runs.
-
-Implementation scope:
-
-- Validate optional `Idempotency-Key` headers using the syntax defined in
-  `docs/roadmap/producer.md`.
-- Scope keys by the authenticated pipeline selected through bearer-token
-  mapping.
-- Record accepted keys after archive staging and source bundle validation
-  succeed.
-- Compare normalized source manifest identity, not raw archive bytes.
-- Return the original accepted run response for the same pipeline, same key,
-  and same manifest identity.
-- Return `409 Conflict` for the same pipeline and key with a different manifest
-  identity.
-- Return a retryable conflict response when the same key is already being
-  processed concurrently for the same pipeline before manifest identity is
-  known.
-- Expire idempotency records with existing upload status retention.
-- Keep idempotency records memory-only; server restart clears them.
-- Preserve current raw HTTP behavior when no idempotency key is supplied.
-
-Current-behavior documentation updates after implementation:
-
-- `docs/integrations/http-upload.md`
-- `docs/operations.md`
-- `docs/internal/app.md`
-- `docs/troubleshooting.md`
-
-Tests:
-
-- Same key and same bundle returns the original run id and does not enqueue a
-  second run.
-- Same key and different bundle returns `409`.
-- Same key under different authenticated pipelines does not conflict.
-- Missing key preserves current raw HTTP behavior.
-- Invalid key syntax returns `400`.
-- Expiration removes idempotency records.
-- Tokens are never leaked; idempotency keys appear only where needed for
-  diagnostics.
-
-Completion criteria:
-
-- Existing upload clients continue to work.
-- The HTTP API has an implemented, tested idempotency contract.
-- Idempotent retries cannot create duplicate accepted runs.
-
-## Stage 2: Public `pkg/upload` API And Client
-
-Goal:
-
-Add a producer-facing upload package that builds or validates bundles, archives
-them, and submits them to the HTTP upload API.
-
-Implementation scope:
-
-- Add public `pkg/upload`.
-- Use `pkg/bundle` for manifest generation, digest and path semantics, local
-  validation, and temporary bundle creation.
-- Implement options-struct APIs matching `docs/roadmap/producer.md`:
-  `ClientOptions`, `RetryOptions`, `UploadBundleOptions`,
-  `UploadFilesOptions`, `Result`, `RunStatus`, `NewClient`, `UploadBundle`,
-  `UploadFiles`, and `Status`.
-- Treat `Endpoint` as the distributor server base URL, deriving `/upload` and
-  `/runs/<run-id>` internally.
-- Require bearer token authentication and redact token values from all errors.
-- Always send `Idempotency-Key`.
-- Use caller-supplied idempotency keys when provided.
-- When no key is supplied, generate one random 128-bit lowercase hex key per
-  upload operation and reuse it across retries from that call.
-- Support uploading an existing local bundle root.
-- Support building a temporary bundle from explicit `bundle.BundleFile` values
-  and uploading it.
-- Create replayable gzip-compressed tar uploads with
-  `Content-Type: application/gzip`.
-- Retry only safe cases: `503 Service Unavailable`, temporary network errors,
-  and ambiguous mid-upload failures, using the same idempotency key and
-  replayable body.
-- Do not retry `400`, `401`, `409`, `413`, or `415`.
-- Do not retry after `202 Accepted`.
-- Respect context cancellation before waiting and before each retry.
-- Close response bodies on every attempt.
-
-Current-behavior documentation updates after implementation:
-
-- Update `pkg/bundle` integration references only as needed once `pkg/upload`
-  exists.
-- Keep full user-facing docs and examples for Stage 3.
-
-Tests:
-
-- Client construction validates endpoint and token requirements.
-- Token values are redacted from errors.
-- Caller-supplied and generated idempotency keys are sent correctly.
-- Existing bundle upload includes only `manifest.json` and manifest-listed
-  files.
-- File-based upload builds a compliant temporary bundle without touching
-  producer source directories.
-- Local validation failures prevent HTTP requests.
-- Response parsing covers `202`, `400`, `401`, `409`, `413`, `415`, `503`,
-  non-JSON errors, and unexpected statuses.
-- Retry uses the same idempotency key and stops correctly.
-- Context cancellation during retry backoff is honored.
-- Custom `*http.Client` behavior is covered with `httptest`.
-
-Completion criteria:
-
-- Go producers can upload valid bundles through `pkg/upload`.
-- Safe retry behavior relies on the implemented server idempotency contract.
-- Public package tests prove upload behavior does not duplicate or drift from
-  `pkg/bundle` semantics.
-
-## Stage 3: Documentation And Examples
-
-Goal:
-
-Document implemented producer upload and idempotency behavior after the server
-contract and public package exist.
-
-Implementation scope:
-
-- Update current-behavior docs only after Stages 1 and 2 are implemented.
-- Add secret-free examples where they are safe, copyable, and describe
-  implemented behavior.
-- Keep deferred items out of current docs.
-
-Docs to update:
-
-- `README.md`
-- `docs/integrations/source-bundle.md`
-- `docs/integrations/http-upload.md`
-- `docs/operations.md`
-- `docs/internal/app.md`
-- `docs/policy/development.md`
-- `examples/`, only if examples are safe, copyable, and implemented
-
-Tests and checks:
-
-```sh
-go test ./...
-rg -n "idempotency|Idempotency-Key|pkg/upload|UploadBundle|UploadFiles" README.md docs examples
-```
-
-Completion criteria:
-
-- Current docs describe the implemented server idempotency and `pkg/upload`
-  API.
-- Completed behavior is not documented only as future work.
-- `docs/roadmap/` contains only future or deferred producer-upload work.
-
-## Deferred Work
-
-- Durable idempotency storage across server restarts.
-- Database-backed queues or durable producer retry processing.
-- `UploadAndWait`, long polling, run cancellation, run retry, or run listing
-  helpers.
-- Zstandard archives.
-- Multipart, resumable, or streaming upload protocols.
-- URL-token authentication.
-- Browser UI or public exposure defaults.
-
-## Validation
-
-For this documentation pass:
-
-```sh
-rg -n "Idempotency-Key|pkg/upload|UploadBundle|UploadFiles|Stage 1: Server-Side Upload Idempotency" docs/roadmap/implementation.md
-rg -n "Producer-supplied idempotency keys|Producer Coordination" docs/roadmap/http.md docs/roadmap/implementation.md
-rg -n "pkg/upload|UploadBundle|UploadFiles|Idempotency-Key" README.md docs examples --glob '!docs/roadmap/**'
-git status --short
-git diff -- docs/roadmap/implementation.md
-```
-
-Expected result:
-
-- New future behavior appears only under `docs/roadmap/`.
-- Existing unrelated worktree changes, including any current
-  `docs/roadmap/documentation.md` deletion, are not touched.
-- This pass changes only `docs/roadmap/implementation.md`.

docs/roadmap/producer.md

@@ -1,322 +0,0 @@
-# Roadmap: Producer HTTP Upload Package
-
-## Purpose
-
-Add a second public producer-facing package that lets Go producer applications
-build or validate a compliant source bundle, package it as a gzip-compressed tar
-archive, and submit it to the HTTP upload API with safe retry support.
-
-The current public producer package, `pkg/bundle`, owns source manifest
-semantics, digest calculation, path validation, local manifest building, local
-bundle writing, and local bundle validation. The new package must build on that
-contract instead of reimplementing it.
-
-This roadmap also adds server-side producer idempotency keys to the HTTP upload
-API. Idempotency is required for the producer upload package's retry behavior:
-the client can safely retry an upload with the same key, and the server can
-collapse duplicate accepted uploads into the original run.
-
-## Goals
-
-- Make the common Go producer workflow small and hard to misuse.
-- Reuse `pkg/bundle` for manifest generation, path normalization, SHA-256
-  calculation, digest calculation, and local validation.
-- Create upload archives that match the server's source bundle archive contract.
-- Add server-side idempotency records scoped to the authenticated pipeline.
-- Send idempotency keys from the public upload package by default.
-- Handle bearer authentication without logging or returning token values.
-- Parse successful, duplicate, conflict, and error responses into typed
-  producer-side results.
-- Retry safely using idempotency keys and bounded backoff.
-- Keep the package dependency-light and usable from ordinary Go producer
-  applications.
-
-## Non-Goals
-
-- Do not expose `internal/app`, `internal/ingest`, storage backends, server
-  config, or destination state types through the public package.
-- Do not add durable idempotency storage, durable client queues, background
-  workers, or database-backed retry processing.
-- Do not add zstd, multipart upload, resumable upload, or non-tar archive
-  formats.
-- Do not require producers to know distributor pipeline ids; server-side token
-  mapping remains authoritative.
-- Do not make the package a replacement for the existing CLI or server API
-  documentation.
-
-## Implementation Sequence
-
-Implement this feature in three stages:
-
-1. Server-side HTTP idempotency keys.
-2. Public `pkg/upload` client package.
-3. Current-behavior documentation and examples.
-
-Server idempotency should land first so the public upload package can rely on
-the final retry contract from its first release.
-
-## Stage 1: Server Idempotency Keys
-
-Goal:
-
-Extend `distributor serve` so `POST /upload` can safely accept retried producer
-uploads without creating duplicate accepted runs.
-
-HTTP contract:
-
-- Producers may send `Idempotency-Key: <key>` with `POST /upload`.
-- The public upload package must always send this header.
-- Raw HTTP clients may omit it; omitted keys preserve current behavior.
-- Keys are scoped to the authenticated pipeline selected by bearer token.
-- Valid keys are non-empty ASCII strings up to 128 bytes using
-  letters, digits, `.`, `_`, `-`, and `:`.
-- Invalid keys return `400`.
-
-Server behavior:
-
-- After archive staging and source bundle validation succeeds, record the
-  idempotency key with the accepted run id and the normalized source manifest
-  identity.
-- If the same pipeline receives the same key and the staged upload has the same
-  normalized source manifest identity, return the original accepted response
-  instead of enqueueing another run.
-- If the same pipeline receives the same key and the staged upload has a
-  different normalized source manifest identity, return `409 Conflict`.
-- If the same key is already being processed concurrently for the same pipeline
-  before a manifest identity is available, return a retryable conflict response
-  without accepting a new run.
-- Idempotency records are memory-only and expire with the existing HTTP upload
-  retention window.
-- Server restart clears idempotency records, matching the current memory-only
-  status and queue behavior.
-
-Manifest identity:
-
-- Compare normalized source manifest semantics, not raw archive bytes.
-- At minimum, compare manifest schema version, id, created timestamp, bundle
-  digest, and ordered file records.
-- Different tar metadata or gzip encoding for the same source bundle should not
-  create a conflict.
-
-Tests:
-
-- `go test ./internal/app ./internal/ingest`
-- Same token, same key, same staged bundle returns the original run id and does
-  not enqueue a second run.
-- Same token, same key, different staged bundle returns `409`.
-- Same key under different authenticated pipelines does not conflict.
-- Missing idempotency key preserves existing raw HTTP behavior.
-- Invalid key syntax returns `400`.
-- Idempotency records expire with completed run status retention.
-- Secret tokens and idempotency keys are not logged in errors beyond the key
-  value itself where required for diagnostics.
-
-Completion criteria:
-
-- The HTTP API has an implemented, tested idempotency contract.
-- Existing clients without `Idempotency-Key` continue to work.
-- Duplicate idempotent uploads cannot create duplicate accepted runs.
-
-## Stage 2: Public `pkg/upload` Client
-
-Goal:
-
-Add a new public `pkg/upload` package that submits compliant bundles to
-`distributor serve` using bearer authentication and idempotency keys.
-
-Package name:
-
-- Use `pkg/upload`.
-- Rationale: `pkg/bundle` owns bundle construction and validation; `pkg/upload`
-  owns submission to the distributor HTTP upload API.
-
-API shape:
-
-Use options-struct APIs rather than one large positional function. Initial APIs
-must cover two producer workflows:
-
-- upload an existing local bundle root;
-- build a temporary bundle from explicit producer files and upload it.
-
-Representative API shape:
-
-```go
-package upload
-
-type Client struct {
-    // unexported fields
-}
-
-type ClientOptions struct {
-    Endpoint   string
-    Token      string
-    HTTPClient *http.Client
-    Retry      RetryOptions
-}
-
-type RetryOptions struct {
-    MaxAttempts int
-    BaseDelay   time.Duration
-    MaxDelay    time.Duration
-}
-
-type UploadBundleOptions struct {
-    Root           string
-    Validate       bool
-    IdempotencyKey string
-}
-
-type UploadFilesOptions struct {
-    ID             string
-    Created        time.Time
-    Files          []bundle.BundleFile
-    Validate       bool
-    TempDir        string
-    IdempotencyKey string
-}
-
-type Result struct {
-    RunID  string
-    Status string
-}
-
-func NewClient(opts ClientOptions) (*Client, error)
-func (c *Client) UploadBundle(ctx context.Context, opts UploadBundleOptions) (Result, error)
-func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Result, error)
-func (c *Client) Status(ctx context.Context, runID string) (RunStatus, error)
-```
-
-Required API semantics:
-
-- `Endpoint` is a distributor server base URL. The client derives `/upload` and
-  `/runs/<run-id>` internally.
-- `Token` is required and is sent as `Authorization: Bearer <token>`.
-- `HTTPClient` is optional; when omitted, use a client with conservative
-  timeouts.
-- `UploadBundle` reads and packages an existing local source bundle.
-- `UploadFiles` creates a temporary complete bundle through `pkg/bundle`, then
-  packages and uploads it.
-- `Status` is optional for callers and never required by `UploadBundle` or
-  `UploadFiles`.
-- `Result` represents upload admission, not final publication success.
-
-Idempotency key behavior:
-
-- The client must send `Idempotency-Key` on every upload.
-- If the caller supplies `IdempotencyKey`, use it.
-- If omitted, generate a random 128-bit lowercase hex key once for that upload
-  operation and reuse it for all retries from that call.
-- Generated keys are not stable across process restarts or separate calls.
-- Producers that need cross-process retry safety must supply their own stable
-  key.
-- Validate caller-supplied keys before making a request.
-
-Bundle and archive behavior:
-
-- `UploadBundle` loads `manifest.json` from the bundle root and validates the
-  local bundle by default.
-- `UploadBundle` includes `manifest.json` and every manifest-listed file in the
-  tar.gz archive, and does not include unlisted files.
-- `UploadFiles` requires a non-empty bundle id and non-empty file list.
-- `UploadFiles` uses `pkg/bundle.WriteBundle` or equivalent public bundle APIs
-  in a temporary directory and preserves explicit file order.
-- Zero `Created` follows `pkg/bundle` defaulting behavior.
-- Validation is enabled by default and may be explicitly disabled only for
-  callers that already performed equivalent validation.
-- Tar entry names are slash-separated bundle-relative paths.
-- The package must not write into producer source directories.
-
-Archive and retry strategy:
-
-- Create a replayable upload body for each upload operation.
-- The implementation may either create a temporary `.tar.gz` file or regenerate
-  the tar.gz body from the validated staged bundle for each attempt.
-- Clean up all temporary bundles and archive files created by the package.
-- Use `Content-Type: application/gzip`.
-
-Retry policy:
-
-- Defaults should be safe and modest, for example three total attempts with
-  bounded exponential backoff.
-- Retry `503 Service Unavailable` because the upload was not accepted.
-- Retry temporary network errors and ambiguous mid-upload failures using the
-  same idempotency key and replayable body.
-- Do not retry `400`, `401`, `409`, `413`, or `415`.
-- Do not retry after `202 Accepted`.
-- Respect caller context cancellation before waiting and before each retry.
-- Redact the bearer token from all errors.
-
-HTTP response handling:
-
-- Parse `202 Accepted` responses into `Result`.
-- Parse JSON error bodies where available.
-- Include HTTP status codes and response messages in typed errors.
-- Treat duplicate idempotent `202` responses the same as first acceptance.
-- Treat `409 Conflict` as an idempotency conflict error.
-- Close response bodies on every attempt.
-
-Tests:
-
-- `go test ./pkg/bundle ./pkg/upload`
-- Client construction with valid and invalid base endpoints.
-- Missing token rejection and token redaction in errors.
-- Caller-supplied and generated idempotency keys.
-- Uploading an existing valid bundle root.
-- Building and uploading from `bundle.BundleFile` values.
-- Local validation failures before any HTTP request.
-- Tar.gz entry names, manifest inclusion, and exclusion of unlisted files.
-- `202`, `400`, `401`, `409`, `413`, `415`, `503`, non-JSON errors, and
-  unexpected status response parsing.
-- Safe retry with the same idempotency key for `503` and retryable network
-  failures.
-- No retry for non-retryable statuses.
-- Context cancellation during retry backoff.
-- Custom `*http.Client` behavior through `httptest.Server`.
-
-Completion criteria:
-
-- Producer applications can build or validate a bundle and upload it with one
-  package.
-- All uploads include idempotency keys.
-- Retry behavior is safe under the server idempotency contract.
-
-## Stage 3: Documentation And Examples
-
-Goal:
-
-Document the implemented producer upload package and idempotency behavior only
-after the server and public package exist.
-
-Current-behavior documentation updates:
-
-- `README.md`: mention the new producer upload package briefly.
-- `docs/integrations/source-bundle.md`: link from producer APIs to upload
-  helpers.
-- `docs/integrations/http-upload.md`: document `Idempotency-Key` and add a
-  short Go producer helper section.
-- `docs/operations.md`: add a concise producer-side example if useful.
-- `docs/internal/app.md`: document server-side idempotency record behavior.
-- `docs/policy/development.md`: document the `pkg/upload` boundary and test
-  expectations.
-
-Tests and checks:
-
-```sh
-go test ./...
-rg -n "pkg/upload|Idempotency-Key|UploadBundle|UploadFiles" README.md docs examples
-```
-
-Completion criteria:
-
-- Current docs describe implemented behavior.
-- Future-only behavior remains under `docs/roadmap/`.
-
-## Deferred Work
-
-- Durable idempotency records across server restarts.
-- Producer-supplied idempotency keys integrated with a database-backed queue.
-- `UploadAndWait` or long-polling helpers.
-- Run cancellation, retry, or listing endpoints.
-- Zstandard-compressed tar archives.
-- Multipart, resumable, or streaming object upload support.
-- URL-token authentication for constrained clients.

docs/troubleshooting.md

@@ -451,6 +451,26 @@ Safe fix: send one valid tar or tar.gz source bundle archive with `Content-Type:
 
 Reference: [Operations](operations.md#http-upload-operation).
 
+## Upload Idempotency Conflict
+
+Symptom: `POST /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.
+
+Diagnostic:
+
+```sh
+curl -i -X POST http://127.0.0.1:8080/upload \
+  -H "Authorization: Bearer $DISTRIBUTOR_UPLOAD_TOKEN" \
+  -H "Content-Type: application/gzip" \
+  -H "Idempotency-Key: <key>" \
+  --data-binary @bundle.tar.gz
+```
+
+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).
+
 ## Upload Status Is Missing
 
 Symptom: `GET /runs/<run_id>` returns `404`.

examples/upload-client/main.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/pkg/upload"
+)
+
+func main() {
+	token := os.Getenv("DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN")
+	if token == "" {
+		log.Fatal("set DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN before running this example")
+	}
+	endpoint := os.Getenv("DISTRIBUTOR_EXAMPLE_UPLOAD_ENDPOINT")
+	if endpoint == "" {
+		endpoint = "http://127.0.0.1:8080"
+	}
+	bundleRoot := "examples/source-bundle"
+	if len(os.Args) > 1 {
+		bundleRoot = os.Args[1]
+	}
+	idempotencyKey := os.Getenv("DISTRIBUTOR_EXAMPLE_UPLOAD_IDEMPOTENCY_KEY")
+
+	client, err := upload.NewClient(upload.ClientOptions{
+		Endpoint: endpoint,
+		Token:    token,
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	opts := upload.UploadBundleOptions{Root: bundleRoot}
+	if idempotencyKey != "" {
+		opts.IdempotencyKey = idempotencyKey
+	}
+	result, err := client.UploadBundle(ctx, opts)
+	if err != nil {
+		log.Fatal(err)
+	}
+	fmt.Printf("accepted run %s with status %s\n", result.RunID, result.Status)
+}

internal/app/upload_coordinator.go

@@ -13,6 +13,7 @@ import (
 
 	"gitea.maximumdirect.net/eric/distributor/internal/config"
 	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
+	sourcebundle "gitea.maximumdirect.net/eric/distributor/pkg/bundle"
 )
 
 const DefaultUploadMaxFileCount = 4096
@@ -46,6 +47,7 @@ type UploadRequest struct {
 	PipelineID     string
 	ContentType    string
 	Body           io.Reader
+	IdempotencyKey string
 	DryRun         bool
 	Force          bool
 	MaxFileCount   int
@@ -64,6 +66,22 @@ func IsUploadQueueFull(err error) bool {
 	return errors.As(err, &full)
 }
 
+type UploadIdempotencyConflictError struct {
+	Retryable bool
+}
+
+func (err UploadIdempotencyConflictError) Error() string {
+	if err.Retryable {
+		return "upload idempotency key is already being processed"
+	}
+	return "upload idempotency key conflicts with a different source manifest"
+}
+
+func IsUploadIdempotencyConflict(err error) bool {
+	var conflict UploadIdempotencyConflictError
+	return errors.As(err, &conflict)
+}
+
 type UploadCoordinator struct {
 	ctx          context.Context
 	cfg          config.Config
@@ -82,6 +100,7 @@ type UploadCoordinator struct {
 	activePipeline map[string]bool
 	pending        []*uploadJob
 	records        map[UploadRunID]UploadRunRecord
+	idempotency    map[uploadIdempotencyScope]uploadIdempotencyRecord
 }
 
 type uploadStageFunc func(context.Context, ingest.StageOptions) (ingest.StagedBundle, error)
@@ -95,6 +114,17 @@ type uploadJob struct {
 	stagedRoot string
 }
 
+type uploadIdempotencyScope struct {
+	PipelineID string
+	Key        string
+}
+
+type uploadIdempotencyRecord struct {
+	RunID    UploadRunID
+	Manifest sourcebundle.Manifest
+	Pending  bool
+}
+
 type uploadCoordinatorHooks struct {
 	stage        uploadStageFunc
 	run          uploadRunFunc
@@ -140,6 +170,7 @@ func newUploadCoordinator(ctx context.Context, cfg config.Config, hooks uploadCo
 		maxConcurrency: cfg.Server.HTTP.MaxConcurrency,
 		activePipeline: map[string]bool{},
 		records:        map[UploadRunID]UploadRunRecord{},
+		idempotency:    map[uploadIdempotencyScope]uploadIdempotencyRecord{},
 	}
 	go coordinator.dispatchLoop()
 	return coordinator
@@ -169,14 +200,26 @@ 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)
 
 	coordinator.mu.Lock()
 	coordinator.expireLocked(coordinator.now().UTC())
+	existingIdempotency, hasExistingIdempotency := coordinator.idempotency[scope]
+	if hasKey && hasExistingIdempotency && existingIdempotency.Pending {
+		coordinator.mu.Unlock()
+		return UploadRunRecord{}, UploadIdempotencyConflictError{Retryable: true}
+	}
+	needsReservation := !hasKey || !hasExistingIdempotency
+	if needsReservation {
 		if coordinator.queueFullLocked() {
 			coordinator.mu.Unlock()
 			return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize}
 		}
 		coordinator.reservedCount++
+		if hasKey {
+			coordinator.idempotency[scope] = uploadIdempotencyRecord{Pending: true}
+		}
+	}
 	coordinator.mu.Unlock()
 
 	staged, err := coordinator.stage(ctx, ingest.StageOptions{
@@ -189,13 +232,36 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
 		MaxFileCount:        uploadMaxFileCount(request.MaxFileCount),
 	})
 	if err != nil {
-		coordinator.releaseReservation()
+		if needsReservation {
+			coordinator.releaseReservation(scope, hasKey)
+		}
 		return UploadRunRecord{}, err
 	}
 
 	coordinator.mu.Lock()
 	defer coordinator.mu.Unlock()
+	if needsReservation {
 		coordinator.reservedCount--
+	}
+	if hasKey {
+		existingIdempotency, hasExistingIdempotency = coordinator.idempotency[scope]
+		if hasExistingIdempotency && !existingIdempotency.Pending {
+			if uploadManifestsEqual(existingIdempotency.Manifest, staged.Manifest) {
+				_ = os.RemoveAll(staged.Root)
+				record, ok := coordinator.records[existingIdempotency.RunID]
+				if !ok {
+					return UploadRunRecord{}, fmt.Errorf("idempotency record references missing run")
+				}
+				return record, nil
+			}
+			_ = os.RemoveAll(staged.Root)
+			return UploadRunRecord{}, UploadIdempotencyConflictError{}
+		}
+		if !hasExistingIdempotency && !needsReservation && coordinator.queueFullLocked() {
+			_ = os.RemoveAll(staged.Root)
+			return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize}
+		}
+	}
 	record := UploadRunRecord{
 		ID:         runID,
 		PipelineID: pipeline.ID,
@@ -204,6 +270,12 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
 		StagedRoot: staged.Root,
 	}
 	coordinator.records[runID] = record
+	if hasKey {
+		coordinator.idempotency[scope] = uploadIdempotencyRecord{
+			RunID:    runID,
+			Manifest: staged.Manifest,
+		}
+	}
 	coordinator.pending = append(coordinator.pending, &uploadJob{
 		recordID:   runID,
 		request:    request,
@@ -214,6 +286,13 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
 	return record, nil
 }
 
+func uploadRequestIdempotencyScope(pipelineID, key string) (uploadIdempotencyScope, bool) {
+	if key == "" {
+		return uploadIdempotencyScope{}, false
+	}
+	return uploadIdempotencyScope{PipelineID: pipelineID, Key: key}, true
+}
+
 func (coordinator *UploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bool) {
 	coordinator.mu.Lock()
 	defer coordinator.mu.Unlock()
@@ -326,10 +405,15 @@ func (coordinator *UploadCoordinator) runJob(job *uploadJob) {
 	coordinator.complete(job, &report, err)
 }
 
-func (coordinator *UploadCoordinator) releaseReservation() {
+func (coordinator *UploadCoordinator) releaseReservation(scope uploadIdempotencyScope, hasKey bool) {
 	coordinator.mu.Lock()
 	defer coordinator.mu.Unlock()
 	coordinator.reservedCount--
+	if hasKey {
+		if record, ok := coordinator.idempotency[scope]; ok && record.Pending {
+			delete(coordinator.idempotency, scope)
+		}
+	}
 }
 
 func (coordinator *UploadCoordinator) queueFullLocked() bool {
@@ -379,10 +463,31 @@ func (coordinator *UploadCoordinator) expireLocked(now time.Time) []UploadRunRec
 		record.Error = ""
 		expired = append(expired, record)
 		delete(coordinator.records, runID)
+		for scope, idempotencyRecord := range coordinator.idempotency {
+			if idempotencyRecord.RunID == runID {
+				delete(coordinator.idempotency, scope)
+			}
+		}
 	}
 	return expired
 }
 
+func uploadManifestsEqual(a, b sourcebundle.Manifest) bool {
+	if a.SchemaVersion != b.SchemaVersion ||
+		a.ID != b.ID ||
+		a.Digest != b.Digest ||
+		!a.Created.Equal(b.Created) ||
+		len(a.Files) != len(b.Files) {
+		return false
+	}
+	for index := range a.Files {
+		if a.Files[index] != b.Files[index] {
+			return false
+		}
+	}
+	return true
+}
+
 func (coordinator *UploadCoordinator) notify() {
 	select {
 	case coordinator.signal <- struct{}{}:

internal/app/upload_coordinator_test.go

@@ -15,6 +15,7 @@ import (
 
 	"gitea.maximumdirect.net/eric/distributor/internal/config"
 	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
+	sourcebundle "gitea.maximumdirect.net/eric/distributor/pkg/bundle"
 )
 
 func TestUploadCoordinatorGeneratesRunIDAndAcceptedStatus(t *testing.T) {
@@ -254,6 +255,233 @@ func TestUploadCoordinatorExpiresCompletedRecordsAndStagingDirectories(t *testin
 	}
 }
 
+func TestUploadCoordinatorIdempotencyReturnsOriginalRunForSameManifest(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var runCount atomic.Int64
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports"},
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        manifestUploadStage,
+		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+			runCount.Add(1)
+			return RunReport{}, nil
+		},
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("same"),
+		IdempotencyKey: "producer.retry:20260603",
+	})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
+
+	second, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("same"),
+		IdempotencyKey: "producer.retry:20260603",
+	})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	if second.ID != first.ID {
+		t.Fatalf("second run id = %q, want original %q", second.ID, first.ID)
+	}
+	if got := runCount.Load(); got != 1 {
+		t.Fatalf("run count = %d, want 1", got)
+	}
+}
+
+func TestUploadCoordinatorIdempotencyConflictsForDifferentManifest(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{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("one"),
+		IdempotencyKey: "same-key",
+	})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
+
+	_, err = coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("two"),
+		IdempotencyKey: "same-key",
+	})
+	if err == nil || !IsUploadIdempotencyConflict(err) {
+		t.Fatalf("second Submit() error = %v, want idempotency conflict", err)
+	}
+}
+
+func TestUploadCoordinatorIdempotencyIsScopedByPipeline(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports-one", "reports-two"},
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        manifestUploadStage,
+		run:          successfulUploadRun,
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports-one",
+		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{
+		PipelineID:     "reports-two",
+		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 pipelines: %q", second.ID)
+	}
+}
+
+func TestUploadCoordinatorWithoutIdempotencyKeyAcceptsDuplicateBodies(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{PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("same")})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	second, err := coordinator.Submit(context.Background(), UploadRequest{PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("same")})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	if second.ID == first.ID {
+		t.Fatalf("second run id = %q, want distinct run", second.ID)
+	}
+}
+
+func TestUploadCoordinatorIdempotencyReturnsRetryableConflictWhileStaging(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	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) {
+			close(entered)
+			<-release
+			return manifestUploadStage(ctx, opts)
+		},
+		run: successfulUploadRun,
+	})
+	firstErr := make(chan error, 1)
+	go func() {
+		_, err := coordinator.Submit(context.Background(), UploadRequest{
+			PipelineID:     "reports",
+			ContentType:    ingest.ContentTypeTar,
+			Body:           strings.NewReader("same"),
+			IdempotencyKey: "in-flight",
+		})
+		firstErr <- err
+	}()
+	<-entered
+
+	var reads atomic.Int64
+	_, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:  "reports",
+		ContentType: ingest.ContentTypeTar,
+		Body: readerFunc(func(data []byte) (int, error) {
+			reads.Add(1)
+			return 0, io.EOF
+		}),
+		IdempotencyKey: "in-flight",
+	})
+	var conflict UploadIdempotencyConflictError
+	if err == nil || !errors.As(err, &conflict) || !conflict.Retryable {
+		t.Fatalf("second Submit() error = %v, want retryable idempotency conflict", err)
+	}
+	if got := reads.Load(); got != 0 {
+		t.Fatalf("retryable conflict body reads = %d, want 0", got)
+	}
+	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()
+	clock := newUploadTestClock(time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC))
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports"},
+		retention:   time.Second,
+	}), uploadCoordinatorHooks{
+		now:          clock.Now,
+		randomSuffix: uploadTestSuffixes("00000001", "00000002", "00000003"),
+		stage:        manifestUploadStage,
+		run:          successfulUploadRun,
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("same"),
+		IdempotencyKey: "expires",
+	})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
+
+	clock.Advance(2 * time.Second)
+	coordinator.Expire()
+
+	second, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("same"),
+		IdempotencyKey: "expires",
+	})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	if second.ID == first.ID {
+		t.Fatalf("second run id = %q, want new run after expiry", second.ID)
+	}
+}
+
 type readerFunc func([]byte) (int, error)
 
 func (fn readerFunc) Read(data []byte) (int, error) {
@@ -268,6 +496,37 @@ func successfulUploadStage(ctx context.Context, opts ingest.StageOptions) (inges
 	return ingest.StagedBundle{Root: root}, nil
 }
 
+func manifestUploadStage(ctx context.Context, opts ingest.StageOptions) (ingest.StagedBundle, error) {
+	data, err := io.ReadAll(opts.Body)
+	if err != nil {
+		return ingest.StagedBundle{}, err
+	}
+	root := filepath.Join(opts.PipelineStagingPath, opts.RunID)
+	if err := os.MkdirAll(root, 0o755); err != nil {
+		return ingest.StagedBundle{}, err
+	}
+	return ingest.StagedBundle{
+		Root:     root,
+		Manifest: uploadTestManifest(string(data)),
+	}, nil
+}
+
+func uploadTestManifest(id string) sourcebundle.Manifest {
+	created := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC)
+	file := sourcebundle.ManifestFile{
+		Path:   "report.md",
+		SHA256: sourcebundle.FileDigest([]byte(id)),
+		Size:   int64(len(id)),
+	}
+	return sourcebundle.Manifest{
+		SchemaVersion: sourcebundle.SchemaVersion,
+		ID:            id,
+		Created:       created,
+		Files:         []sourcebundle.ManifestFile{file},
+		Digest:        sourcebundle.BundleDigest([]sourcebundle.ManifestFile{file}),
+	}
+}
+
 func successfulUploadRun(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
 	return RunReport{}, nil
 }

internal/app/upload_http.go

@@ -30,8 +30,11 @@ type uploadAcceptedResponse struct {
 
 type httpErrorResponse struct {
 	Error     string `json:"error"`
+	Retryable bool   `json:"retryable,omitempty"`
 }
 
+const idempotencyKeyHeader = "Idempotency-Key"
+
 func newUploadHTTPHandler(ctx context.Context, cfg config.Config, environment config.Environment) (http.Handler, error) {
 	config.ApplyDefaults(&cfg)
 	tokens, err := resolveUploadTokens(cfg, environment)
@@ -98,14 +101,16 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req
 		writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type")
 		return
 	}
-	if !handler.coordinator.CanAccept() {
+	idempotencyKey, err := uploadIdempotencyKey(r.Header)
-		writeHTTPError(w, http.StatusServiceUnavailable, "upload queue is full")
+	if err != nil {
+		writeHTTPError(w, http.StatusBadRequest, "invalid idempotency key")
 		return
 	}
 	record, err := handler.coordinator.Submit(r.Context(), UploadRequest{
 		PipelineID:     pipelineID,
 		ContentType:    contentType,
 		Body:           r.Body,
+		IdempotencyKey: idempotencyKey,
 	})
 	if err != nil {
 		writeUploadSubmitError(w, err)
@@ -144,10 +149,44 @@ func (handler uploadHTTPHandler) authenticate(header string) (string, bool) {
 	return pipelineID, ok
 }
 
+func uploadIdempotencyKey(header http.Header) (string, error) {
+	values := header.Values(idempotencyKeyHeader)
+	if len(values) == 0 {
+		return "", nil
+	}
+	if len(values) != 1 {
+		return "", fmt.Errorf("idempotency key must appear at most once")
+	}
+	key := values[0]
+	if key == "" {
+		return "", fmt.Errorf("idempotency key is required when header is present")
+	}
+	if len(key) > 128 {
+		return "", fmt.Errorf("idempotency key must be at most 128 bytes")
+	}
+	for index := 0; index < len(key); index++ {
+		character := key[index]
+		if character >= 'a' && character <= 'z' ||
+			character >= 'A' && character <= 'Z' ||
+			character >= '0' && character <= '9' ||
+			character == '.' ||
+			character == '_' ||
+			character == '-' ||
+			character == ':' {
+			continue
+		}
+		return "", fmt.Errorf("idempotency key contains unsupported character")
+	}
+	return key, nil
+}
+
 func writeUploadSubmitError(w http.ResponseWriter, err error) {
+	var idempotencyConflict UploadIdempotencyConflictError
 	switch {
 	case IsUploadQueueFull(err):
 		writeHTTPError(w, http.StatusServiceUnavailable, "upload queue is full")
+	case errors.As(err, &idempotencyConflict):
+		writeHTTPErrorRetryable(w, http.StatusConflict, idempotencyConflict.Error(), idempotencyConflict.Retryable)
 	case errors.Is(err, ingest.ErrUploadTooLarge):
 		writeHTTPError(w, http.StatusRequestEntityTooLarge, "upload exceeds maximum size")
 	case errors.Is(err, ingest.ErrUnsupportedContentType):
@@ -158,7 +197,11 @@ func writeUploadSubmitError(w http.ResponseWriter, err error) {
 }
 
 func writeHTTPError(w http.ResponseWriter, status int, message string) {
-	writeJSON(w, status, httpErrorResponse{Error: message})
+	writeHTTPErrorRetryable(w, status, message, false)
+}
+
+func writeHTTPErrorRetryable(w http.ResponseWriter, status int, message string, retryable bool) {
+	writeJSON(w, status, httpErrorResponse{Error: message, Retryable: retryable})
 }
 
 func writeJSON(w http.ResponseWriter, status int, value any) {

internal/app/upload_http_integration_test.go

@@ -97,6 +97,62 @@ func TestHTTPUploadInvalidArchiveIsRejectedWithoutRunID(t *testing.T) {
 	assertDirectoryEmpty(t, destination)
 }
 
+func TestHTTPUploadIdempotencyReturnsOriginalRunForSameBundle(t *testing.T) {
+	destination := t.TempDir()
+	coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
+		id:           "reports",
+		tokenEnv:     "REPORTS_TOKEN",
+		stagingPath:  filepath.Join(t.TempDir(), "reports"),
+		destinations: []string{destination},
+	}}, 4, 1))
+	handler := uploadHTTPHandler{
+		coordinator: coordinator,
+		tokens:      map[string]string{"reports-secret": "reports"},
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	firstRunID := submitHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeTar, "same-key", bundleArchive(t, false, testutil.BundleOptions{}))
+	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)
+
+	secondRunID := submitHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeGzip, "same-key", bundleArchive(t, true, testutil.BundleOptions{}))
+	if secondRunID != firstRunID {
+		t.Fatalf("second run id = %q, want original %q", secondRunID, firstRunID)
+	}
+	if got := coordinator.QueueDepth(); got != 0 {
+		t.Fatalf("queue depth = %d, want no duplicate run queued", got)
+	}
+}
+
+func TestHTTPUploadIdempotencyReturnsConflictForDifferentBundle(t *testing.T) {
+	destination := t.TempDir()
+	coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
+		id:           "reports",
+		tokenEnv:     "REPORTS_TOKEN",
+		stagingPath:  filepath.Join(t.TempDir(), "reports"),
+		destinations: []string{destination},
+	}}, 4, 1))
+	handler := uploadHTTPHandler{
+		coordinator: coordinator,
+		tokens:      map[string]string{"reports-secret": "reports"},
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	firstRunID := submitHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeTar, "same-key", bundleArchive(t, false, testutil.BundleOptions{}))
+	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)
+
+	status, body := postHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeTar, "same-key", bundleArchive(t, false, testutil.BundleOptions{
+		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)
+	}
+	if strings.Contains(body, "reports-secret") {
+		t.Fatalf("conflict response exposed token: %s", body)
+	}
+}
+
 func TestHTTPUploadOversizedArchiveIsRejectedWithoutRunID(t *testing.T) {
 	destination := t.TempDir()
 	stagingPath := filepath.Join(t.TempDir(), "reports")
@@ -278,6 +334,17 @@ 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 decodeAcceptedHTTPUpload(t, status, responseBody)
+}
+
+func submitHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, contentType, key string, body []byte) UploadRunID {
+	t.Helper()
+	status, responseBody := postHTTPUploadWithKey(t, server, token, contentType, key, body)
+	return decodeAcceptedHTTPUpload(t, status, responseBody)
+}
+
+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)
 	}
@@ -292,6 +359,11 @@ func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType
 }
 
 func postHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) (int, string) {
+	t.Helper()
+	return postHTTPUploadWithKey(t, server, 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))
 	if err != nil {
@@ -299,6 +371,9 @@ func postHTTPUpload(t *testing.T, server *httptest.Server, token, contentType st
 	}
 	request.Header.Set("Authorization", "Bearer "+token)
 	request.Header.Set("Content-Type", contentType)
+	if key != "" {
+		request.Header.Set("Idempotency-Key", key)
+	}
 	response, err := server.Client().Do(request)
 	if err != nil {
 		t.Fatalf("POST /upload error = %v", err)

internal/app/upload_http_test.go

@@ -97,7 +97,6 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
 	var submitted UploadRequest
 	handler := uploadHTTPHandler{
 		coordinator: fakeUploadCoordinator{
-			canAccept: true,
 			submit: func(_ context.Context, request UploadRequest) (UploadRunRecord, error) {
 				submitted = request
 				body, err := io.ReadAll(request.Body)
@@ -116,6 +115,7 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
 	request := httptest.NewRequest(http.MethodPost, "/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")
 
 	handler.ServeHTTP(recorder, request)
 
@@ -125,6 +125,9 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
 	if submitted.PipelineID != "reports" {
 		t.Fatalf("submitted pipeline = %q, want reports", submitted.PipelineID)
 	}
+	if submitted.IdempotencyKey != "producer.retry:20260603" {
+		t.Fatalf("submitted idempotency key = %q, want producer.retry:20260603", submitted.IdempotencyKey)
+	}
 	var response uploadAcceptedResponse
 	if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
 		t.Fatalf("decode response: %v", err)
@@ -139,7 +142,7 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
 
 func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) {
 	handler := uploadHTTPHandler{
-		coordinator: fakeUploadCoordinator{canAccept: true},
+		coordinator: fakeUploadCoordinator{},
 		tokens:      map[string]string{"valid-token": "reports"},
 	}
 
@@ -160,34 +163,56 @@ func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) {
 	}
 }
 
-func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *testing.T) {
+func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t *testing.T) {
 	tests := []struct {
 		name        string
-		canAccept   bool
 		url         string
 		contentType string
+		keyValues   []string
 		body        io.Reader
 		wantStatus  int
 	}{
 		{
 			name:        "unsupported content type",
-			canAccept:   true,
 			url:         "/upload",
 			contentType: "application/zip",
 			body:        strings.NewReader("archive"),
 			wantStatus:  http.StatusUnsupportedMediaType,
 		},
 		{
-			name:        "full queue",
+			name:        "invalid key syntax",
-			canAccept:   false,
 			url:         "/upload",
 			contentType: "application/x-tar",
-			body:        &countingReader{reader: strings.NewReader("archive")},
+			keyValues:   []string{"bad key"},
-			wantStatus:  http.StatusServiceUnavailable,
+			body:        strings.NewReader("archive"),
+			wantStatus:  http.StatusBadRequest,
+		},
+		{
+			name:        "empty key",
+			url:         "/upload",
+			contentType: "application/x-tar",
+			keyValues:   []string{""},
+			body:        strings.NewReader("archive"),
+			wantStatus:  http.StatusBadRequest,
+		},
+		{
+			name:        "too long key",
+			url:         "/upload",
+			contentType: "application/x-tar",
+			keyValues:   []string{strings.Repeat("a", 129)},
+			body:        strings.NewReader("archive"),
+			wantStatus:  http.StatusBadRequest,
+		},
+		{
+			name:        "multiple keys",
+			url:         "/upload",
+			contentType: "application/x-tar",
+			keyValues:   []string{"one", "two"},
+			body:        strings.NewReader("archive"),
+			wantStatus:  http.StatusBadRequest,
 		},
 		{
 			name:        "submitted pipeline id",
-			canAccept:   true,
 			url:         "/upload?pipeline_id=reports",
 			contentType: "application/x-tar",
 			body:        strings.NewReader("archive"),
@@ -198,7 +223,6 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t
 		t.Run(tt.name, func(t *testing.T) {
 			handler := uploadHTTPHandler{
 				coordinator: fakeUploadCoordinator{
-					canAccept: tt.canAccept,
 					submit: func(context.Context, UploadRequest) (UploadRunRecord, error) {
 						t.Fatal("Submit should not be called")
 						return UploadRunRecord{}, nil
@@ -210,15 +234,15 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t
 			request := httptest.NewRequest(http.MethodPost, tt.url, tt.body)
 			request.Header.Set("Authorization", "Bearer valid-token")
 			request.Header.Set("Content-Type", tt.contentType)
+			for _, value := range tt.keyValues {
+				request.Header.Add("Idempotency-Key", value)
+			}
 
 			handler.ServeHTTP(recorder, request)
 
 			if recorder.Code != tt.wantStatus {
 				t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String())
 			}
-			if reader, ok := tt.body.(*countingReader); ok && reader.reads != 0 {
-				t.Fatalf("full queue read body %d time(s), want zero", reader.reads)
-			}
 		})
 	}
 }
@@ -228,9 +252,13 @@ func TestUploadHTTPHandlerMapsSubmitErrors(t *testing.T) {
 		name       string
 		err        error
 		wantStatus int
+		wantBody   string
 	}{
 		{name: "oversized", err: ingest.ErrUploadTooLarge, wantStatus: http.StatusRequestEntityTooLarge},
 		{name: "unsupported", err: ingest.ErrUnsupportedContentType, wantStatus: http.StatusUnsupportedMediaType},
+		{name: "full queue", err: UploadQueueFullError{QueueSize: 1}, wantStatus: http.StatusServiceUnavailable},
+		{name: "idempotency conflict", err: UploadIdempotencyConflictError{}, wantStatus: http.StatusConflict, wantBody: "different source manifest"},
+		{name: "idempotency in progress", err: UploadIdempotencyConflictError{Retryable: true}, wantStatus: http.StatusConflict, wantBody: `"retryable":true`},
 		{name: "malformed", err: errors.New("malformed archive"), wantStatus: http.StatusBadRequest},
 	}
 	for _, tt := range tests {
@@ -254,6 +282,9 @@ func TestUploadHTTPHandlerMapsSubmitErrors(t *testing.T) {
 			if recorder.Code != tt.wantStatus {
 				t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String())
 			}
+			if tt.wantBody != "" && !strings.Contains(recorder.Body.String(), tt.wantBody) {
+				t.Fatalf("body = %q, want substring %q", recorder.Body.String(), tt.wantBody)
+			}
 		})
 	}
 }

pkg/upload/archive.go

@@ -0,0 +1,104 @@
+package upload
+
+import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/pkg/bundle"
+)
+
+func archiveBundle(root string, manifest bundle.Manifest) ([]byte, error) {
+	var output bytes.Buffer
+	gzipWriter := gzip.NewWriter(&output)
+	tarWriter := tar.NewWriter(gzipWriter)
+
+	manifestData, err := bundle.MarshalManifest(manifest)
+	if err != nil {
+		return nil, err
+	}
+	if err := writeTarEntry(tarWriter, bundle.ManifestName, manifestData, 0o600, manifest.Created); err != nil {
+		_ = tarWriter.Close()
+		_ = gzipWriter.Close()
+		return nil, err
+	}
+	for _, manifestFile := range manifest.Files {
+		fullPath := filepath.Join(root, filepath.FromSlash(manifestFile.Path))
+		info, err := os.Lstat(fullPath)
+		if err != nil {
+			_ = tarWriter.Close()
+			_ = gzipWriter.Close()
+			return nil, fmt.Errorf("file %q stat: %w", manifestFile.Path, err)
+		}
+		if !info.Mode().IsRegular() {
+			_ = tarWriter.Close()
+			_ = gzipWriter.Close()
+			return nil, fmt.Errorf("file %q must be a regular file", manifestFile.Path)
+		}
+		file, err := os.Open(fullPath)
+		if err != nil {
+			_ = tarWriter.Close()
+			_ = gzipWriter.Close()
+			return nil, fmt.Errorf("file %q open: %w", manifestFile.Path, err)
+		}
+		if err := writeTarFile(tarWriter, manifestFile.Path, file, info.Mode().Perm(), info.ModTime(), info.Size()); err != nil {
+			_ = file.Close()
+			_ = tarWriter.Close()
+			_ = gzipWriter.Close()
+			return nil, err
+		}
+		if err := file.Close(); err != nil {
+			_ = tarWriter.Close()
+			_ = gzipWriter.Close()
+			return nil, fmt.Errorf("file %q close: %w", manifestFile.Path, err)
+		}
+	}
+	if err := tarWriter.Close(); err != nil {
+		_ = gzipWriter.Close()
+		return nil, fmt.Errorf("close tar archive: %w", err)
+	}
+	if err := gzipWriter.Close(); err != nil {
+		return nil, fmt.Errorf("close gzip archive: %w", err)
+	}
+	return output.Bytes(), nil
+}
+
+func writeTarEntry(writer *tar.Writer, name string, data []byte, mode int64, modTime time.Time) error {
+	header := &tar.Header{
+		Name:    name,
+		Mode:    mode,
+		Size:    int64(len(data)),
+		ModTime: modTime,
+	}
+	if err := writer.WriteHeader(header); err != nil {
+		return fmt.Errorf("write tar header %q: %w", name, err)
+	}
+	if _, err := writer.Write(data); err != nil {
+		return fmt.Errorf("write tar entry %q: %w", name, err)
+	}
+	return nil
+}
+
+func writeTarFile(writer *tar.Writer, name string, file *os.File, mode os.FileMode, modTime time.Time, size int64) error {
+	if mode == 0 {
+		mode = 0o600
+	}
+	header := &tar.Header{
+		Name:    name,
+		Mode:    int64(mode),
+		Size:    size,
+		ModTime: modTime,
+	}
+	if err := writer.WriteHeader(header); err != nil {
+		return fmt.Errorf("write tar header %q: %w", name, err)
+	}
+	if _, err := io.Copy(writer, file); err != nil {
+		return fmt.Errorf("write tar entry %q: %w", name, err)
+	}
+	return nil
+}

pkg/upload/client.go

@@ -0,0 +1,426 @@
+package upload
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"path"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/pkg/bundle"
+)
+
+const (
+	uploadPath            = "upload"
+	runsPath              = "runs"
+	idempotencyKeyHeader  = "Idempotency-Key"
+	defaultHTTPTimeout    = 30 * time.Second
+	defaultRetryAttempts  = 3
+	defaultRetryBaseDelay = 100 * time.Millisecond
+	defaultRetryMaxDelay  = time.Second
+	uploadContentTypeGzip = "application/gzip"
+	authorizationPrefix   = "Bearer "
+	redactedSecret        = "[redacted]"
+)
+
+func NewClient(opts ClientOptions) (*Client, error) {
+	endpoint, err := cleanEndpoint(opts.Endpoint)
+	if err != nil {
+		return nil, err
+	}
+	if opts.Token == "" {
+		return nil, fmt.Errorf("token is required")
+	}
+	retry, err := cleanRetryOptions(opts.Retry)
+	if err != nil {
+		return nil, err
+	}
+	httpClient := opts.HTTPClient
+	if httpClient == nil {
+		httpClient = &http.Client{Timeout: defaultHTTPTimeout}
+	}
+	return &Client{
+		endpoint:   endpoint,
+		token:      opts.Token,
+		httpClient: httpClient,
+		retry:      retry,
+	}, nil
+}
+
+func (c *Client) UploadBundle(ctx context.Context, opts UploadBundleOptions) (Result, error) {
+	if c == nil {
+		return Result{}, fmt.Errorf("client is nil")
+	}
+	if opts.Validate && opts.DisableValidation {
+		return Result{}, fmt.Errorf("validate and disable validation cannot both be set")
+	}
+	if opts.Root == "" {
+		return Result{}, fmt.Errorf("root is required")
+	}
+	manifest, err := bundle.LoadManifest(opts.Root)
+	if err != nil {
+		return Result{}, c.redactError(err)
+	}
+	if shouldValidateBundle(opts.Validate, opts.DisableValidation) {
+		if err := bundle.ValidateBundle(opts.Root, manifest); err != nil {
+			return Result{}, c.redactError(err)
+		}
+	}
+	archive, err := archiveBundle(opts.Root, manifest)
+	if err != nil {
+		return Result{}, c.redactError(err)
+	}
+	key, err := uploadIdempotencyKey(opts.IdempotencyKey)
+	if err != nil {
+		return Result{}, err
+	}
+	return c.uploadArchive(ctx, archive, key)
+}
+
+func (c *Client) UploadFiles(ctx context.Context, opts UploadFilesOptions) (Result, error) {
+	if c == nil {
+		return Result{}, fmt.Errorf("client is nil")
+	}
+	if opts.Validate && opts.DisableValidation {
+		return Result{}, fmt.Errorf("validate and disable validation cannot both be set")
+	}
+	if opts.ID == "" {
+		return Result{}, fmt.Errorf("id is required")
+	}
+	if len(opts.Files) == 0 {
+		return Result{}, fmt.Errorf("files is required")
+	}
+	tempRoot, err := os.MkdirTemp(opts.TempDir, "distributor-upload-*")
+	if err != nil {
+		return Result{}, c.redactError(fmt.Errorf("create temporary bundle root: %w", err))
+	}
+	defer func() {
+		_ = os.RemoveAll(tempRoot)
+	}()
+	localBundleRoot := filepath.Join(tempRoot, "bundle")
+	manifest, err := bundle.WriteBundle(bundle.WriteBundleOptions{
+		Root:    localBundleRoot,
+		ID:      opts.ID,
+		Created: opts.Created,
+		Files:   opts.Files,
+	})
+	if err != nil {
+		return Result{}, c.redactError(err)
+	}
+	if shouldValidateBundle(opts.Validate, opts.DisableValidation) {
+		if err := bundle.ValidateBundle(localBundleRoot, manifest); err != nil {
+			return Result{}, c.redactError(err)
+		}
+	}
+	archive, err := archiveBundle(localBundleRoot, manifest)
+	if err != nil {
+		return Result{}, c.redactError(err)
+	}
+	key, err := uploadIdempotencyKey(opts.IdempotencyKey)
+	if err != nil {
+		return Result{}, err
+	}
+	return c.uploadArchive(ctx, archive, key)
+}
+
+func (c *Client) Status(ctx context.Context, runID string) (RunStatus, error) {
+	if c == nil {
+		return RunStatus{}, fmt.Errorf("client is nil")
+	}
+	if runID == "" {
+		return RunStatus{}, fmt.Errorf("run id is required")
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if err := ctx.Err(); err != nil {
+		return RunStatus{}, err
+	}
+	request, err := http.NewRequestWithContext(ctx, http.MethodGet, c.statusURL(runID), nil)
+	if err != nil {
+		return RunStatus{}, c.redactError(err)
+	}
+	c.authorize(request)
+	response, err := c.httpClient.Do(request)
+	if err != nil {
+		return RunStatus{}, c.redactError(err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		return RunStatus{}, c.responseError(response)
+	}
+	var status RunStatus
+	if err := json.NewDecoder(response.Body).Decode(&status); err != nil {
+		return RunStatus{}, c.redactError(fmt.Errorf("decode run status: %w", err))
+	}
+	return status, nil
+}
+
+func (c *Client) uploadArchive(ctx context.Context, archive []byte, idempotencyKey string) (Result, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	var lastErr error
+	for attempt := 1; attempt <= c.retry.MaxAttempts; attempt++ {
+		if err := ctx.Err(); err != nil {
+			return Result{}, err
+		}
+		result, retry, err := c.uploadAttempt(ctx, archive, idempotencyKey)
+		if err == nil {
+			return result, nil
+		}
+		lastErr = err
+		if !retry || attempt == c.retry.MaxAttempts {
+			return Result{}, err
+		}
+		if err := waitForRetry(ctx, retryDelay(c.retry, attempt)); err != nil {
+			return Result{}, err
+		}
+	}
+	return Result{}, lastErr
+}
+
+func (c *Client) uploadAttempt(ctx context.Context, archive []byte, idempotencyKey string) (Result, bool, error) {
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.uploadURL(), bytes.NewReader(archive))
+	if err != nil {
+		return Result{}, false, c.redactError(err)
+	}
+	c.authorize(request)
+	request.Header.Set("Content-Type", uploadContentTypeGzip)
+	request.Header.Set(idempotencyKeyHeader, idempotencyKey)
+
+	response, err := c.httpClient.Do(request)
+	if err != nil {
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			return Result{}, false, ctxErr
+		}
+		return Result{}, isRetryableNetworkError(err), c.redactError(err)
+	}
+	defer response.Body.Close()
+
+	if response.StatusCode == http.StatusAccepted {
+		var result Result
+		if err := json.NewDecoder(response.Body).Decode(&result); err != nil {
+			return Result{}, false, c.redactError(fmt.Errorf("decode upload response: %w", err))
+		}
+		if result.RunID == "" {
+			return Result{}, false, fmt.Errorf("upload response run_id is required")
+		}
+		return result, false, nil
+	}
+	err = c.responseError(response)
+	return Result{}, response.StatusCode == http.StatusServiceUnavailable, err
+}
+
+func (c *Client) authorize(request *http.Request) {
+	request.Header.Set("Authorization", authorizationPrefix+c.token)
+}
+
+func (c *Client) uploadURL() string {
+	return joinEndpointPath(c.endpoint, uploadPath)
+}
+
+func (c *Client) statusURL(runID string) string {
+	return joinEndpointPath(c.endpoint, runsPath, runID)
+}
+
+func (c *Client) responseError(response *http.Response) error {
+	body, readErr := io.ReadAll(response.Body)
+	message := http.StatusText(response.StatusCode)
+	retryable := false
+	if readErr == nil && len(body) > 0 {
+		var decoded struct {
+			Error     string `json:"error"`
+			Retryable bool   `json:"retryable"`
+		}
+		if err := json.Unmarshal(body, &decoded); err == nil && decoded.Error != "" {
+			message = decoded.Error
+			retryable = decoded.Retryable
+		} else if trimmed := strings.TrimSpace(string(body)); trimmed != "" {
+			message = trimmed
+		}
+	}
+	message = c.redactString(message)
+	status := c.redactString(response.Status)
+	httpErr := HTTPError{
+		StatusCode: response.StatusCode,
+		Status:     status,
+		Message:    message,
+		Retryable:  retryable,
+	}
+	if response.StatusCode == http.StatusConflict {
+		return &IdempotencyConflictError{HTTPError: httpErr}
+	}
+	return &httpErr
+}
+
+func (c *Client) redactError(err error) error {
+	if err == nil {
+		return nil
+	}
+	message := c.redactString(err.Error())
+	if message == err.Error() {
+		return err
+	}
+	return errors.New(message)
+}
+
+func (c *Client) redactString(value string) string {
+	if c == nil || c.token == "" {
+		return value
+	}
+	return strings.ReplaceAll(value, c.token, redactedSecret)
+}
+
+func cleanEndpoint(value string) (string, error) {
+	if value == "" {
+		return "", fmt.Errorf("endpoint is required")
+	}
+	parsed, err := url.Parse(value)
+	if err != nil {
+		return "", fmt.Errorf("endpoint is invalid: %w", err)
+	}
+	if parsed.Scheme != "http" && parsed.Scheme != "https" {
+		return "", fmt.Errorf("endpoint scheme must be http or https")
+	}
+	if parsed.Host == "" {
+		return "", fmt.Errorf("endpoint host is required")
+	}
+	if parsed.User != nil {
+		return "", fmt.Errorf("endpoint userinfo is not supported")
+	}
+	if parsed.RawQuery != "" || parsed.Fragment != "" {
+		return "", fmt.Errorf("endpoint must not include query or fragment")
+	}
+	parsed.Path = strings.TrimRight(parsed.Path, "/")
+	parsed.RawPath = ""
+	return parsed.String(), nil
+}
+
+func cleanRetryOptions(opts RetryOptions) (RetryOptions, error) {
+	if opts.MaxAttempts < 0 {
+		return RetryOptions{}, fmt.Errorf("retry max attempts must be non-negative")
+	}
+	if opts.BaseDelay < 0 {
+		return RetryOptions{}, fmt.Errorf("retry base delay must be non-negative")
+	}
+	if opts.MaxDelay < 0 {
+		return RetryOptions{}, fmt.Errorf("retry max delay must be non-negative")
+	}
+	if opts.MaxAttempts == 0 {
+		opts.MaxAttempts = defaultRetryAttempts
+	}
+	if opts.BaseDelay == 0 {
+		opts.BaseDelay = defaultRetryBaseDelay
+	}
+	if opts.MaxDelay == 0 {
+		opts.MaxDelay = defaultRetryMaxDelay
+	}
+	if opts.MaxDelay < opts.BaseDelay {
+		return RetryOptions{}, fmt.Errorf("retry max delay must be greater than or equal to base delay")
+	}
+	return opts, nil
+}
+
+func uploadIdempotencyKey(value string) (string, error) {
+	if value == "" {
+		return randomIdempotencyKey()
+	}
+	if err := validateIdempotencyKey(value); err != nil {
+		return "", err
+	}
+	return value, nil
+}
+
+func validateIdempotencyKey(value string) error {
+	if value == "" {
+		return fmt.Errorf("idempotency key is required")
+	}
+	if len(value) > 128 {
+		return fmt.Errorf("idempotency key must be at most 128 bytes")
+	}
+	for index := 0; index < len(value); index++ {
+		character := value[index]
+		if character >= 'a' && character <= 'z' ||
+			character >= 'A' && character <= 'Z' ||
+			character >= '0' && character <= '9' ||
+			character == '.' ||
+			character == '_' ||
+			character == '-' ||
+			character == ':' {
+			continue
+		}
+		return fmt.Errorf("idempotency key contains unsupported character")
+	}
+	return nil
+}
+
+func randomIdempotencyKey() (string, error) {
+	var data [16]byte
+	if _, err := rand.Read(data[:]); err != nil {
+		return "", fmt.Errorf("generate idempotency key: %w", err)
+	}
+	return hex.EncodeToString(data[:]), nil
+}
+
+func shouldValidateBundle(validate, disable bool) bool {
+	return validate || !disable
+}
+
+func retryDelay(opts RetryOptions, attempt int) time.Duration {
+	delay := opts.BaseDelay
+	for index := 1; index < attempt; index++ {
+		delay *= 2
+		if delay >= opts.MaxDelay {
+			return opts.MaxDelay
+		}
+	}
+	return delay
+}
+
+func waitForRetry(ctx context.Context, delay time.Duration) error {
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
+func isRetryableNetworkError(err error) bool {
+	if err == nil {
+		return false
+	}
+	var netErr net.Error
+	if errors.As(err, &netErr) {
+		return netErr.Timeout() || netErr.Temporary()
+	}
+	return true
+}
+
+func joinEndpointPath(endpoint string, elements ...string) string {
+	parsed, err := url.Parse(endpoint)
+	if err != nil {
+		return endpoint
+	}
+	parts := []string{}
+	if parsed.Path != "" && parsed.Path != "/" {
+		parts = append(parts, strings.Trim(parsed.Path, "/"))
+	}
+	parts = append(parts, elements...)
+	parsed.Path = "/" + path.Join(parts...)
+	return parsed.String()
+}

pkg/upload/client_test.go

@@ -0,0 +1,559 @@
+package upload
+
+import (
+	"archive/tar"
+	"compress/gzip"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	sourcebundle "gitea.maximumdirect.net/eric/distributor/pkg/bundle"
+)
+
+func TestNewClientValidatesOptions(t *testing.T) {
+	tests := []struct {
+		name string
+		opts ClientOptions
+	}{
+		{name: "missing endpoint", opts: ClientOptions{Token: "secret"}},
+		{name: "missing token", opts: ClientOptions{Endpoint: "http://127.0.0.1:8080"}},
+		{name: "bad scheme", opts: ClientOptions{Endpoint: "ftp://127.0.0.1:8080", Token: "secret"}},
+		{name: "missing host", opts: ClientOptions{Endpoint: "http:///upload", Token: "secret"}},
+		{name: "query", opts: ClientOptions{Endpoint: "http://127.0.0.1:8080?x=1", Token: "secret"}},
+		{name: "userinfo", opts: ClientOptions{Endpoint: "http://user@127.0.0.1:8080", Token: "secret"}},
+		{name: "negative attempts", opts: ClientOptions{Endpoint: "http://127.0.0.1:8080", Token: "secret", Retry: RetryOptions{MaxAttempts: -1}}},
+		{name: "negative delay", opts: ClientOptions{Endpoint: "http://127.0.0.1:8080", Token: "secret", Retry: RetryOptions{BaseDelay: -1}}},
+		{name: "max below base", opts: ClientOptions{Endpoint: "http://127.0.0.1:8080", Token: "secret", Retry: RetryOptions{BaseDelay: time.Second, MaxDelay: time.Millisecond}}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if _, err := NewClient(tt.opts); err == nil {
+				t.Fatal("NewClient() error = nil, want error")
+			}
+		})
+	}
+
+	client, err := NewClient(ClientOptions{Endpoint: "http://127.0.0.1:8080/base/", Token: "secret"})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	if got, want := client.uploadURL(), "http://127.0.0.1:8080/base/upload"; got != want {
+		t.Fatalf("upload URL = %q, want %q", got, want)
+	}
+	if client.httpClient == nil || client.httpClient.Timeout == 0 {
+		t.Fatalf("default HTTP client = %#v, want timeout", client.httpClient)
+	}
+}
+
+func TestUploadBundleSendsCallerKeyAndManifestArchive(t *testing.T) {
+	root := writeTestBundle(t, "reports.daily", []testFile{
+		{path: "report.md", data: "# Report\n"},
+		{path: "nested/summary.txt", data: "Summary\n"},
+	})
+	if err := os.WriteFile(filepath.Join(root, "unlisted.txt"), []byte("nope"), 0o600); err != nil {
+		t.Fatalf("write unlisted file: %v", err)
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if got, want := r.URL.Path, "/upload"; got != want {
+			t.Fatalf("path = %q, want %q", got, want)
+		}
+		if got, want := r.Header.Get("Authorization"), "Bearer secret-token"; got != want {
+			t.Fatalf("authorization = %q, want %q", got, want)
+		}
+		if got, want := r.Header.Get("Content-Type"), uploadContentTypeGzip; got != want {
+			t.Fatalf("content type = %q, want %q", got, want)
+		}
+		if got, want := r.Header.Get(idempotencyKeyHeader), "producer.retry:one"; got != want {
+			t.Fatalf("idempotency key = %q, want %q", got, want)
+		}
+		entries := readArchiveEntries(t, r.Body)
+		if got, want := strings.Join(entryNames(entries), ","), "manifest.json,report.md,nested/summary.txt"; got != want {
+			t.Fatalf("archive entries = %q, want %q", got, want)
+		}
+		if _, ok := entries["unlisted.txt"]; ok {
+			t.Fatal("archive included unlisted file")
+		}
+		writeAccepted(t, w, "reports.20260604T120000Z.abcdef12")
+	}))
+	defer server.Close()
+
+	client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret-token", HTTPClient: server.Client()})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	result, err := client.UploadBundle(context.Background(), UploadBundleOptions{
+		Root:           root,
+		IdempotencyKey: "producer.retry:one",
+	})
+	if err != nil {
+		t.Fatalf("UploadBundle() error = %v", err)
+	}
+	if result.RunID != "reports.20260604T120000Z.abcdef12" || result.Status != "accepted" {
+		t.Fatalf("result = %#v, want accepted run", result)
+	}
+}
+
+func TestUploadFilesBuildsTemporaryBundleWithoutTouchingSources(t *testing.T) {
+	sourceRoot := t.TempDir()
+	sourcePath := filepath.Join(sourceRoot, "producer-output.md")
+	if err := os.WriteFile(sourcePath, []byte("producer data\n"), 0o600); err != nil {
+		t.Fatalf("write source: %v", err)
+	}
+	tempDir := t.TempDir()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		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)
+		}
+		if got, want := string(entries["reports/report.md"]), "producer data\n"; got != want {
+			t.Fatalf("uploaded file = %q, want %q", got, want)
+		}
+		if _, ok := entries["producer-output.md"]; ok {
+			t.Fatal("archive used producer source path instead of bundle path")
+		}
+		writeAccepted(t, w, "reports.20260604T120000Z.abcdef12")
+	}))
+	defer server.Close()
+
+	client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	_, err = client.UploadFiles(context.Background(), UploadFilesOptions{
+		ID: "reports.from.files",
+		Files: []sourcebundle.BundleFile{{
+			SourcePath: sourcePath,
+			Path:       "reports/report.md",
+		}},
+		TempDir: tempDir,
+	})
+	if err != nil {
+		t.Fatalf("UploadFiles() error = %v", err)
+	}
+	if _, err := os.Stat(filepath.Join(sourceRoot, sourcebundle.ManifestName)); !os.IsNotExist(err) {
+		t.Fatalf("producer source manifest stat = %v, want not exist", 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 cleanup", len(entries))
+	}
+}
+
+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 {
+		t.Fatalf("mutate bundle file: %v", err)
+	}
+	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)
+	}
+	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root}); err == nil {
+		t.Fatal("UploadBundle() error = nil, want validation error")
+	}
+	if got := requests.Load(); got != 0 {
+		t.Fatalf("requests = %d, want 0", got)
+	}
+}
+
+func TestUploadBundleCanDisableLocalValidation(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 {
+		t.Fatalf("mutate bundle file: %v", err)
+	}
+	var requests atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		writeAccepted(t, w, "reports.20260604T120000Z.abcdef12")
+	}))
+	defer server.Close()
+
+	client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, DisableValidation: true}); err != nil {
+		t.Fatalf("UploadBundle() error = %v", err)
+	}
+	if got := requests.Load(); got != 1 {
+		t.Fatalf("requests = %d, want 1", got)
+	}
+}
+
+func TestGeneratedIdempotencyKeyIsReusedAcrossRetry(t *testing.T) {
+	root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}})
+	var attempts atomic.Int64
+	var keys []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		keys = append(keys, r.Header.Get(idempotencyKeyHeader))
+		if attempts.Add(1) == 1 {
+			writeJSONError(w, http.StatusServiceUnavailable, "busy", false)
+			return
+		}
+		writeAccepted(t, w, "reports.20260604T120000Z.abcdef12")
+	}))
+	defer server.Close()
+
+	client, err := NewClient(ClientOptions{
+		Endpoint:   server.URL,
+		Token:      "secret",
+		HTTPClient: server.Client(),
+		Retry:      RetryOptions{MaxAttempts: 2, BaseDelay: time.Millisecond, MaxDelay: time.Millisecond},
+	})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root}); err != nil {
+		t.Fatalf("UploadBundle() error = %v", err)
+	}
+	if got, want := attempts.Load(), int64(2); got != want {
+		t.Fatalf("attempts = %d, want %d", got, want)
+	}
+	if len(keys) != 2 || keys[0] == "" || keys[0] != keys[1] {
+		t.Fatalf("idempotency keys = %#v, want same generated key", keys)
+	}
+	if !regexp.MustCompile(`^[0-9a-f]{32}$`).MatchString(keys[0]) {
+		t.Fatalf("generated key = %q, want 128-bit lowercase hex", keys[0])
+	}
+}
+
+func TestUploadResponseParsingAndNoRetryStatuses(t *testing.T) {
+	root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}})
+	tests := []struct {
+		name          string
+		status        int
+		body          string
+		wantConflict  bool
+		wantMessage   string
+		wantRetryable bool
+	}{
+		{name: "bad request", status: http.StatusBadRequest, body: `{"error":"bad bundle"}`, wantMessage: "bad bundle"},
+		{name: "unauthorized", status: http.StatusUnauthorized, body: `{"error":"bad token"}`, wantMessage: "bad token"},
+		{name: "conflict", status: http.StatusConflict, body: `{"error":"different manifest","retryable":true}`, wantConflict: true, wantMessage: "different manifest", wantRetryable: true},
+		{name: "too large", status: http.StatusRequestEntityTooLarge, body: `{"error":"too large"}`, wantMessage: "too large"},
+		{name: "unsupported", status: http.StatusUnsupportedMediaType, body: `{"error":"unsupported"}`, wantMessage: "unsupported"},
+		{name: "service unavailable", status: http.StatusServiceUnavailable, body: `{"error":"busy"}`, wantMessage: "busy"},
+		{name: "non json", status: http.StatusBadRequest, body: `plain failure`, wantMessage: "plain failure"},
+		{name: "unexpected", status: http.StatusTeapot, body: ``, wantMessage: "I'm a teapot"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var attempts atomic.Int64
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				attempts.Add(1)
+				w.WriteHeader(tt.status)
+				_, _ = w.Write([]byte(tt.body))
+			}))
+			defer server.Close()
+
+			client, err := NewClient(ClientOptions{
+				Endpoint:   server.URL,
+				Token:      "secret",
+				HTTPClient: server.Client(),
+				Retry:      RetryOptions{MaxAttempts: 1},
+			})
+			if err != nil {
+				t.Fatalf("NewClient() error = %v", err)
+			}
+			_, err = client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "key"})
+			if err == nil {
+				t.Fatal("UploadBundle() error = nil, want error")
+			}
+			var httpErr *HTTPError
+			if !errors.As(err, &httpErr) {
+				t.Fatalf("error = %T %v, want HTTPError", err, err)
+			}
+			if httpErr.StatusCode != tt.status || !strings.Contains(httpErr.Message, tt.wantMessage) || httpErr.Retryable != tt.wantRetryable {
+				t.Fatalf("HTTPError = %#v, want status %d message %q retryable %t", httpErr, tt.status, tt.wantMessage, tt.wantRetryable)
+			}
+			var conflict *IdempotencyConflictError
+			if got := errors.As(err, &conflict); got != tt.wantConflict {
+				t.Fatalf("conflict error = %t, want %t", got, tt.wantConflict)
+			}
+			if got := attempts.Load(); got != 1 {
+				t.Fatalf("attempts = %d, want 1", got)
+			}
+		})
+	}
+}
+
+func TestTokenRedactedFromHTTPError(t *testing.T) {
+	root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}})
+	token := "super-secret-token"
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		writeJSONError(w, http.StatusBadRequest, "token "+token+" rejected", false)
+	}))
+	defer server.Close()
+	client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: token, HTTPClient: server.Client(), Retry: RetryOptions{MaxAttempts: 1}})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	_, err = client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "key"})
+	if err == nil {
+		t.Fatal("UploadBundle() error = nil, want error")
+	}
+	if strings.Contains(err.Error(), token) {
+		t.Fatalf("error exposed token: %v", err)
+	}
+	if !strings.Contains(err.Error(), redactedSecret) {
+		t.Fatalf("error = %v, want redaction marker", err)
+	}
+}
+
+func TestNetworkRetryUsesSameIdempotencyKey(t *testing.T) {
+	root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}})
+	var attempts atomic.Int64
+	var keys []string
+	client, err := NewClient(ClientOptions{
+		Endpoint: "http://upload.example",
+		Token:    "secret",
+		HTTPClient: &http.Client{Transport: roundTripFunc(func(request *http.Request) (*http.Response, error) {
+			keys = append(keys, request.Header.Get(idempotencyKeyHeader))
+			if attempts.Add(1) == 1 {
+				return nil, temporaryNetworkError{}
+			}
+			return &http.Response{
+				StatusCode: http.StatusAccepted,
+				Status:     "202 Accepted",
+				Header:     make(http.Header),
+				Body:       io.NopCloser(strings.NewReader(`{"run_id":"reports.20260604T120000Z.abcdef12","status":"accepted"}`)),
+				Request:    request,
+			}, nil
+		})},
+		Retry: RetryOptions{MaxAttempts: 2, BaseDelay: time.Millisecond, MaxDelay: time.Millisecond},
+	})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	result, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "network-retry"})
+	if err != nil {
+		t.Fatalf("UploadBundle() error = %v", err)
+	}
+	if result.RunID == "" {
+		t.Fatalf("result = %#v, want run id", result)
+	}
+	if got, want := attempts.Load(), int64(2); got != want {
+		t.Fatalf("attempts = %d, want %d", got, want)
+	}
+	if got, want := strings.Join(keys, ","), "network-retry,network-retry"; got != want {
+		t.Fatalf("keys = %q, want %q", got, want)
+	}
+}
+
+func TestContextCancellationDuringRetryBackoff(t *testing.T) {
+	root := writeTestBundle(t, "reports.daily", []testFile{{path: "report.md", data: "data"}})
+	ctx, cancel := context.WithCancel(context.Background())
+	var attempts atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		attempts.Add(1)
+		cancel()
+		writeJSONError(w, http.StatusServiceUnavailable, "busy", false)
+	}))
+	defer server.Close()
+
+	client, err := NewClient(ClientOptions{
+		Endpoint:   server.URL,
+		Token:      "secret",
+		HTTPClient: server.Client(),
+		Retry:      RetryOptions{MaxAttempts: 2, BaseDelay: time.Hour, MaxDelay: time.Hour},
+	})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	_, err = client.UploadBundle(ctx, UploadBundleOptions{Root: root, IdempotencyKey: "cancel"})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("UploadBundle() error = %v, want context.Canceled", err)
+	}
+	if got := attempts.Load(); got != 1 {
+		t.Fatalf("attempts = %d, want 1", got)
+	}
+}
+
+func TestStatusParsesRunStatusAndErrors(t *testing.T) {
+	acceptedAt := time.Date(2026, 6, 4, 12, 0, 0, 0, time.UTC)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if got, want := r.URL.Path, "/runs/reports.20260604T120000Z.abcdef12"; got != want {
+			t.Fatalf("path = %q, want %q", got, want)
+		}
+		if got, want := r.Header.Get("Authorization"), "Bearer secret"; got != want {
+			t.Fatalf("authorization = %q, want %q", got, want)
+		}
+		_ = json.NewEncoder(w).Encode(RunStatus{
+			RunID:      "reports.20260604T120000Z.abcdef12",
+			PipelineID: "reports",
+			Status:     "succeeded",
+			AcceptedAt: acceptedAt,
+		})
+	}))
+	defer server.Close()
+
+	client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	status, err := client.Status(context.Background(), "reports.20260604T120000Z.abcdef12")
+	if err != nil {
+		t.Fatalf("Status() error = %v", err)
+	}
+	if status.RunID != "reports.20260604T120000Z.abcdef12" || status.PipelineID != "reports" || status.Status != "succeeded" || !status.AcceptedAt.Equal(acceptedAt) {
+		t.Fatalf("status = %#v, want succeeded run", status)
+	}
+
+	errorServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		writeJSONError(w, http.StatusNotFound, "run not found", false)
+	}))
+	defer errorServer.Close()
+	client, err = NewClient(ClientOptions{Endpoint: errorServer.URL, Token: "secret", HTTPClient: errorServer.Client()})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	_, err = client.Status(context.Background(), "missing")
+	var httpErr *HTTPError
+	if err == nil || !errors.As(err, &httpErr) || httpErr.StatusCode != http.StatusNotFound {
+		t.Fatalf("Status() error = %v, want 404 HTTPError", err)
+	}
+}
+
+func TestInvalidCallerIdempotencyKeyPreventsHTTPRequest(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)
+	}))
+	defer server.Close()
+	client, err := NewClient(ClientOptions{Endpoint: server.URL, Token: "secret", HTTPClient: server.Client()})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	if _, err := client.UploadBundle(context.Background(), UploadBundleOptions{Root: root, IdempotencyKey: "bad key"}); err == nil {
+		t.Fatal("UploadBundle() error = nil, want invalid key error")
+	}
+	if got := requests.Load(); got != 0 {
+		t.Fatalf("requests = %d, want 0", got)
+	}
+}
+
+type testFile struct {
+	path string
+	data string
+}
+
+func writeTestBundle(t *testing.T, id string, files []testFile) string {
+	t.Helper()
+	sourceRoot := t.TempDir()
+	bundleFiles := make([]sourcebundle.BundleFile, 0, len(files))
+	for _, file := range files {
+		sourcePath := filepath.Join(sourceRoot, filepath.FromSlash(file.path))
+		if err := os.MkdirAll(filepath.Dir(sourcePath), 0o755); err != nil {
+			t.Fatalf("mkdir source parent: %v", err)
+		}
+		if err := os.WriteFile(sourcePath, []byte(file.data), 0o600); err != nil {
+			t.Fatalf("write source file: %v", err)
+		}
+		bundleFiles = append(bundleFiles, sourcebundle.BundleFile{
+			SourcePath: sourcePath,
+			Path:       file.path,
+		})
+	}
+	root := filepath.Join(t.TempDir(), "bundle")
+	if _, err := sourcebundle.WriteBundle(sourcebundle.WriteBundleOptions{
+		Root:    root,
+		ID:      id,
+		Created: time.Date(2026, 6, 4, 12, 0, 0, 0, time.UTC),
+		Files:   bundleFiles,
+	}); err != nil {
+		t.Fatalf("WriteBundle() error = %v", err)
+	}
+	return root
+}
+
+func readArchiveEntries(t *testing.T, body io.Reader) map[string][]byte {
+	t.Helper()
+	gzipReader, err := gzip.NewReader(body)
+	if err != nil {
+		t.Fatalf("open gzip archive: %v", err)
+	}
+	defer gzipReader.Close()
+	tarReader := tar.NewReader(gzipReader)
+	entries := map[string][]byte{}
+	for {
+		header, err := tarReader.Next()
+		if errors.Is(err, io.EOF) {
+			return entries
+		}
+		if err != nil {
+			t.Fatalf("read tar archive: %v", err)
+		}
+		data, err := io.ReadAll(tarReader)
+		if err != nil {
+			t.Fatalf("read tar entry %q: %v", header.Name, err)
+		}
+		entries[header.Name] = data
+	}
+}
+
+func entryNames(entries map[string][]byte) []string {
+	ordered := []string{}
+	for _, name := range []string{"manifest.json", "report.md", "nested/summary.txt", "reports/report.md", "unlisted.txt"} {
+		if _, ok := entries[name]; ok {
+			ordered = append(ordered, name)
+		}
+	}
+	return ordered
+}
+
+func writeAccepted(t *testing.T, w http.ResponseWriter, runID string) {
+	t.Helper()
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusAccepted)
+	if err := json.NewEncoder(w).Encode(Result{RunID: runID, Status: "accepted"}); err != nil {
+		t.Fatalf("write accepted response: %v", err)
+	}
+}
+
+func writeJSONError(w http.ResponseWriter, status int, message string, retryable bool) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(map[string]any{"error": message, "retryable": retryable})
+}
+
+type roundTripFunc func(*http.Request) (*http.Response, error)
+
+func (fn roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) {
+	return fn(request)
+}
+
+type temporaryNetworkError struct{}
+
+func (temporaryNetworkError) Error() string {
+	return "temporary network failure"
+}
+
+func (temporaryNetworkError) Timeout() bool {
+	return false
+}
+
+func (temporaryNetworkError) Temporary() bool {
+	return true
+}

pkg/upload/types.go

@@ -0,0 +1,92 @@
+package upload
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/pkg/bundle"
+)
+
+type Client struct {
+	endpoint   string
+	token      string
+	httpClient *http.Client
+	retry      RetryOptions
+}
+
+type ClientOptions struct {
+	Endpoint   string
+	Token      string
+	HTTPClient *http.Client
+	Retry      RetryOptions
+}
+
+type RetryOptions struct {
+	MaxAttempts int
+	BaseDelay   time.Duration
+	MaxDelay    time.Duration
+}
+
+type UploadBundleOptions struct {
+	Root              string
+	Validate          bool
+	DisableValidation bool
+	IdempotencyKey    string
+}
+
+type UploadFilesOptions struct {
+	ID                string
+	Created           time.Time
+	Files             []bundle.BundleFile
+	Validate          bool
+	DisableValidation bool
+	TempDir           string
+	IdempotencyKey    string
+}
+
+type Result struct {
+	RunID  string `json:"run_id"`
+	Status string `json:"status"`
+}
+
+type RunStatus struct {
+	RunID      string          `json:"run_id"`
+	PipelineID string          `json:"pipeline_id"`
+	Status     string          `json:"status"`
+	AcceptedAt time.Time       `json:"accepted_at"`
+	StartedAt  *time.Time      `json:"started_at,omitempty"`
+	FinishedAt *time.Time      `json:"finished_at,omitempty"`
+	Report     json.RawMessage `json:"report,omitempty"`
+	Error      string          `json:"error,omitempty"`
+}
+
+type HTTPError struct {
+	StatusCode int
+	Status     string
+	Message    string
+	Retryable  bool
+}
+
+func (err *HTTPError) Error() string {
+	if err == nil {
+		return "<nil>"
+	}
+	if err.Message == "" {
+		return fmt.Sprintf("upload request failed: %s", err.Status)
+	}
+	return fmt.Sprintf("upload request failed: %s: %s", err.Status, err.Message)
+}
+
+type IdempotencyConflictError struct {
+	HTTPError
+}
+
+func (err *IdempotencyConflictError) Error() string {
+	return (*HTTPError)(&err.HTTPError).Error()
+}
+
+func (err *IdempotencyConflictError) Unwrap() error {
+	return &err.HTTPError
+}