diff --git a/docs/config.md b/docs/config.md
index 19092e9..1e1a30d 100644
--- a/docs/config.md
+++ b/docs/config.md
@@ -92,16 +92,22 @@ weatherreporter uploads one distributor bundle per generated report after
 - `timeout`: distributor operation timeout. Must be greater than zero when
   enabled. Default: `30s`.
 - `failure_policy`: must be `error` when enabled. Default: `error`.
+- `pipeline_id_template`: template for the distributor pipeline ID. Required
+  when enabled. Default: empty.
 - `bundle_id_template`: template for distributor bundle IDs. Default:
-  `weatherreporter.{location_id}.{report_id}.{run_id}`.
+  `weatherreporter.{location_id}.{report_id}`.
 - `idempotency_key_template`: template for distributor idempotency keys.
-  Default: `{bundle_id}`.
+  Default: `{bundle_id}.{run_id}`.
 - `report_path_template`: template for the Markdown report path inside the
   distributor bundle. Default: `{batch_output_name}`.
 
 Supported template variables are `location_id`, `report_id`, `run_id`,
-`artifact_group`, and `batch_output_name`. `idempotency_key_template` may also
-use `bundle_id`.
+`artifact_group`, and `batch_output_name`. `pipeline_id_template` and
+`idempotency_key_template` may also use `bundle_id`.
+
+The rendered pipeline ID selects the configured distributor `http_upload`
+workflow. The rendered bundle ID is the stable logical source identity for the
+report stream. The rendered idempotency key is the per-run retry identity.
 
 Rendered report paths must be relative paths with `/` separators. They must not
 contain backslashes, empty path segments, `.`, `..`, `manifest.json`, or
diff --git a/docs/integrations/distributor/api.md b/docs/integrations/distributor/api.md
index 365ffb8..2871db3 100644
--- a/docs/integrations/distributor/api.md
+++ b/docs/integrations/distributor/api.md
@@ -9,13 +9,16 @@ This document is the copyable implementation guide for submitting producer outpu
 The upstream application needs these values from deployment or operator configuration:
 
 - distributor endpoint: the HTTP server base URL, such as `https://distributor.example.com`;
-- upload token: bearer token for exactly one configured `http_upload` pipeline;
+- upload token: bearer token that authenticates the producer;
+- pipeline id: configured `http_upload` pipeline that should process this upload;
 - generated files: regular local files to include in the source bundle;
-- bundle id: stable identifier for this producer output;
-- idempotency key: stable key for retrying the same producer operation.
+- bundle id: stable identifier for the logical report stream or artifact;
+- idempotency key: unique key for one producer run, reused only when retrying that same run.
 
 Do not put destination routing, public URLs, transform settings, or credentials in the source manifest. Those belong in the `distributor` pipeline configuration.
 
+The token, pipeline id, bundle id, and idempotency key have different jobs. The token authenticates the producer. The pipeline id selects the configured distributor workflow, including destinations and publishing policy. The bundle id tells `distributor` whether a new upload is a newer version of the same source; keep it stable across runs that should replace the same managed destination artifact. The idempotency key tells `distributor` whether an upload request is a retry; change it for each distinct producer run so new content is enqueued.
+
 ## Recommended Workflow
 
 Use `gitea.maximumdirect.net/eric/distributor/pkg/upload`.
@@ -53,7 +56,9 @@ func SubmitReport(reportPath, summaryPath string) error {
 		return fmt.Errorf("distributor endpoint and token are required")
 	}
 
-	reportID := "weather.hourly.brentwood.2026-06-07T15"
+	pipelineID := "weather-hourly"
+	reportID := "weather.hourly.brentwood"
+	runID := time.Now().UTC().Format("20060102T150405.000000000Z")
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
@@ -66,8 +71,9 @@ func SubmitReport(reportPath, summaryPath string) error {
 	}
 
 	result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{
+		PipelineID:     pipelineID,
 		ID:             reportID,
-		IdempotencyKey: reportID,
+		IdempotencyKey: reportID + "." + runID,
 		Files: []bundle.BundleFile{
 			{SourcePath: reportPath, Path: "report.md"},
 			{SourcePath: summaryPath, Path: "summary.txt"},
@@ -88,8 +94,11 @@ func SubmitReport(reportPath, summaryPath string) error {
 
 ## Producer Responsibilities
 
-- Use a stable bundle id for the producer output, such as a report type plus logical timestamp.
-- Use a stable idempotency key for cross-process retries of the same producer operation.
+- Use a stable bundle id for the logical producer output that should replace the same destination artifact, such as `weather.hourly.brentwood`.
+- Set `PipelineID` to the configured upload pipeline that should process the bundle.
+- Do not include per-run timestamps, random values, or job ids in the bundle id unless each run should be treated as a different source.
+- Use an idempotency key that changes for every distinct producer run, such as `<bundle-id>.<run-id>`.
+- Reuse the same idempotency key only when retrying the exact same producer run with the same source manifest.
 - Map each generated file to a clean slash-separated bundle path, such as `report.md` or `assets/chart.png`.
 - Include only regular files. Symlinks, directories as files, devices, FIFOs, and sockets are rejected.
 - Keep file contents stable after upload inputs are selected. Bundle digests are calculated from file bytes.
@@ -99,9 +108,9 @@ Valid bundle paths are relative slash paths. They must not be empty, absolute, c
 
 ## Idempotency And Status
 
-`pkg/upload` sends `Idempotency-Key` on every upload. If the caller omits one, the package generates a random key for that call and reuses it for in-process retries. That is enough for transient network retry within one process.
+`pkg/upload` sends `Idempotency-Key` on every upload. If the caller omits one, the package generates a random key for that call and reuses it for in-process retries. That is enough for transient network retry within one process, but it does not give cross-process retry identity.
 
-For producer jobs that may retry after process restart, supply a stable key derived from the producer operation, such as the report id or job id. Reusing the same key with the same normalized source manifest returns the original accepted run. Reusing the same key with different source content returns a conflict.
+For producer jobs that may retry after process restart, supply a key derived from the producer run, such as `<bundle-id>.<run-id>`. Reusing the same key with the same token, pipeline id, and normalized source manifest returns the original accepted run. Reusing the same key with different source content in that scope returns a conflict. Reusing one key across multiple distinct report generations prevents those generations from being treated as new uploads.
 
 `Status` polls `/runs/<run-id>` while the distributor server retains the in-memory status record. Status values are `accepted`, `queued`, `running`, `succeeded`, and `failed`. Completed records expire according to the server's `server.http.retention` setting, and server restart clears status and idempotency records.
 
diff --git a/docs/integrations/distributor/pkg-bundle.md b/docs/integrations/distributor/pkg-bundle.md
index 87775d9..691e094 100644
--- a/docs/integrations/distributor/pkg-bundle.md
+++ b/docs/integrations/distributor/pkg-bundle.md
@@ -19,7 +19,7 @@ Use `WriteBundle` when producer-generated files live outside the final bundle ro
 ```go
 manifest, err := bundle.WriteBundle(bundle.WriteBundleOptions{
 	Root: "/var/spool/distributor/weather/hourly-2026-06-07T15",
-	ID:   "weather.hourly.brentwood.2026-06-07T15",
+	ID:   "weather.hourly.brentwood",
 	Files: []bundle.BundleFile{
 		{SourcePath: "/tmp/weather/report.md", Path: "report.md"},
 		{SourcePath: "/tmp/weather/summary.txt", Path: "summary.txt"},
@@ -41,7 +41,7 @@ Use `BuildManifest` and `WriteManifest` when files are already staged under the
 root := "/var/spool/distributor/weather/hourly-2026-06-07T15"
 manifest, err := bundle.BuildManifest(bundle.BuildOptions{
 	Root:  root,
-	ID:    "weather.hourly.brentwood.2026-06-07T15",
+	ID:    "weather.hourly.brentwood",
 	Files: []string{"report.md", "summary.txt"},
 })
 if err != nil {
@@ -72,6 +72,8 @@ Invalid paths include:
 
 Explicit file lists preserve caller order. File order is part of the bundle digest, so producers should choose it deliberately and keep it stable.
 
+The manifest `ID` is the logical source identity used by `distributor` destination comparison. Keep it stable for runs that should replace the same managed destination artifact. If every run uses a different manifest `ID`, `distributor` treats those runs as different sources and may report a destination conflict instead of replacing older output.
+
 ## Validation And Digest Helpers
 
 Use `ValidateBundle` before handing an existing local bundle to another process. It verifies manifest semantics, file existence, regular-file type, file size, per-file SHA-256 digests, and bundle digest.
diff --git a/docs/integrations/distributor/pkg-upload.md b/docs/integrations/distributor/pkg-upload.md
index d6ff35c..6ca073e 100644
--- a/docs/integrations/distributor/pkg-upload.md
+++ b/docs/integrations/distributor/pkg-upload.md
@@ -8,7 +8,7 @@ Import path:
 import "gitea.maximumdirect.net/eric/distributor/pkg/upload"
 ```
 
-`pkg/upload` is the producer-facing HTTP upload client. It builds on `pkg/bundle`, packages valid source bundles as gzip-compressed tar archives, sends bearer authentication, includes idempotency keys, and exposes a status polling helper.
+`pkg/upload` is the producer-facing HTTP upload client. It builds on `pkg/bundle`, packages valid source bundles as gzip-compressed tar archives, sends bearer authentication, routes uploads to a configured pipeline, includes idempotency keys, and exposes a status polling helper.
 
 `UploadFiles` examples also use:
 
@@ -30,7 +30,7 @@ if err != nil {
 }
 ```
 
-`Endpoint` is the distributor server base URL. The client derives `/upload` and `/runs/<run-id>`. `Token` is required and is sent as `Authorization: Bearer <token>`. Token values are redacted from client errors.
+`Endpoint` is the distributor server base URL. The client derives `/v1/pipelines/<pipeline-id>/upload` and `/runs/<run-id>`. `Token` is required and is sent as `Authorization: Bearer <token>`. Token values are redacted from client errors.
 
 `HTTPClient` and `Retry` are optional. Defaults use a 30 second HTTP timeout and safe retry settings.
 
@@ -40,8 +40,9 @@ Use `UploadFiles` when the producer has generated output files but has not assem
 
 ```go
 result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{
-	ID:             "weather.hourly.brentwood.2026-06-07T15",
-	IdempotencyKey: "weather.hourly.brentwood.2026-06-07T15",
+	PipelineID:     "weather-hourly",
+	ID:             "weather.hourly.brentwood",
+	IdempotencyKey: "weather.hourly.brentwood.20260607T150000Z",
 	Files: []bundle.BundleFile{
 		{SourcePath: "/tmp/weather/report.md", Path: "report.md"},
 		{SourcePath: "/tmp/weather/summary.txt", Path: "summary.txt"},
@@ -53,7 +54,7 @@ if err != nil {
 _ = result.RunID
 ```
 
-`UploadFiles` creates a temporary bundle, writes and validates a manifest, uploads the archive, and removes temporary files when the call returns. It does not write into producer source directories.
+`PipelineID` is required and selects the configured distributor workflow for this upload. `ID` is the source manifest id and identifies the logical artifact inside that workflow. `UploadFiles` creates a temporary bundle, writes and validates a manifest, uploads the archive, and removes temporary files when the call returns. It does not write into producer source directories.
 
 ## Upload An Existing Bundle
 
@@ -61,8 +62,9 @@ Use `UploadBundle` when the producer already has a complete local bundle root co
 
 ```go
 result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{
+	PipelineID:     "weather-hourly",
 	Root:           "/var/spool/weather/hourly-2026-06-07T15",
-	IdempotencyKey: "weather.hourly.brentwood.2026-06-07T15",
+	IdempotencyKey: "weather.hourly.brentwood.20260607T150000Z",
 })
 if err != nil {
 	return err
@@ -70,7 +72,7 @@ if err != nil {
 _ = result.RunID
 ```
 
-`UploadBundle` validates the local bundle by default and uploads only `manifest.json` plus manifest-listed files. Unlisted files are not uploaded.
+`PipelineID` is required for existing bundles too. `UploadBundle` validates the local bundle by default and uploads only `manifest.json` plus manifest-listed files. Unlisted files are not uploaded.
 
 ## Result And Status
 
@@ -94,7 +96,9 @@ Status values are `accepted`, `queued`, `running`, `succeeded`, and `failed`. Co
 
 Every upload request includes `Idempotency-Key`.
 
-If `IdempotencyKey` is omitted, the client generates a random 128-bit lowercase hexadecimal key for that upload operation and reuses it for retries within the same call. For cross-process retry safety, producers should pass a stable key derived from the producer job or report id.
+If `IdempotencyKey` is omitted, the client generates a random 128-bit lowercase hexadecimal key for that upload operation and reuses it for retries within the same call. For cross-process retry safety, producers should pass a key derived from the producer run, such as `<bundle-id>.<run-id>`.
+
+Do not reuse the same idempotency key for multiple distinct report generations. Reuse it only when retrying the exact same run with the same token, pipeline id, and source manifest. A repeated key with the same manifest in that scope returns the original accepted run instead of enqueueing another run; a repeated key with different content returns an idempotency conflict.
 
 The client retries only safe cases:
 
@@ -102,7 +106,7 @@ The client retries only safe cases:
 - temporary network errors;
 - ambiguous mid-upload failures.
 
-It does not retry after `202 Accepted` and does not retry `400`, `401`, `409`, `413`, or `415`.
+It does not retry after `202 Accepted` and does not retry `400`, `401`, `403`, `404`, `409`, `413`, or `415`.
 
 Detect conflicting key reuse with `errors.As`:
 
diff --git a/docs/internal/app-orchestration.md b/docs/internal/app-orchestration.md
index 9168ecf..b39ddf0 100644
--- a/docs/internal/app-orchestration.md
+++ b/docs/internal/app-orchestration.md
@@ -90,8 +90,8 @@ generation returns an error after writing output, the managed report and
 metadata remain inspectable. Notification is not attempted after Weather API,
 briefing, prompt input, render, Scriptorium run, or metadata-save failures.
 When notification is attempted, the debug artifact records request identity,
-accepted upload fields, distributor status fields, raw status report JSON when
-available, and redacted failure context.
+including rendered pipeline ID, accepted upload fields, distributor status
+fields, raw status report JSON when available, and redacted failure context.
 `--out` copies are never used as notification source files.
 
 ## Batch Workflow
diff --git a/docs/internal/distributor-adapter.md b/docs/internal/distributor-adapter.md
index d4e983f..62f38e1 100644
--- a/docs/internal/distributor-adapter.md
+++ b/docs/internal/distributor-adapter.md
@@ -18,6 +18,7 @@ Inputs:
 - distributor endpoint URL
 - token environment variable name
 - upload timeout
+- pipeline ID
 - bundle ID
 - idempotency key
 - source Markdown report path
@@ -54,8 +55,10 @@ The adapter is built from `notify.distributor` config:
 - `token_env`
 - `timeout`
 
-The app layer renders bundle ID, idempotency key, and bundle path from:
+The app layer renders pipeline ID, bundle ID, idempotency key, and bundle path
+from:
 
+- `pipeline_id_template`
 - `bundle_id_template`
 - `idempotency_key_template`
 - `report_path_template`
@@ -67,6 +70,7 @@ after config loading and `secrets.directory` processing.
 
 The adapter calls distributor `UploadFiles` with exactly one file:
 
+- pipeline ID: the rendered distributor workflow selector
 - source path: the managed Markdown report path selected by app orchestration
 - bundle path: the rendered bundle-relative report path
 - created: the report generation timestamp
@@ -86,12 +90,13 @@ with the status report preserved.
 
 ## Failure Behavior
 
-The adapter validates required endpoint, token env name, token value, bundle ID,
-idempotency key, source path, bundle path, and upload client inputs before
-uploading.
+The adapter validates required endpoint, token env name, token value, pipeline
+ID, bundle ID, idempotency key, source path, bundle path, and upload client
+inputs before uploading.
 
-Upload failures include endpoint, bundle ID, idempotency key, source path, and
-bundle path context. Token values are redacted from adapter errors.
+Upload failures include endpoint, pipeline ID, bundle ID, idempotency key,
+source path, and bundle path context. Token values are redacted from adapter
+errors.
 
 Distributor idempotency conflicts are exposed as a weatherreporter-owned
 `IdempotencyConflictError`, so callers do not depend on upstream distributor
diff --git a/docs/internal/state.md b/docs/internal/state.md
index 7fd8c24..6903d59 100644
--- a/docs/internal/state.md
+++ b/docs/internal/state.md
@@ -93,7 +93,8 @@ Durable JSON writes use shared atomic file helpers. Managed Markdown reports are
 prepared by creating their parent directory; Scriptorium writes the report body
 to the prepared path. Extra Markdown copies are handled by app orchestration.
 Distributor notification debug artifacts are written atomically when
-notification is attempted.
+notification is attempted and include rendered distributor pipeline ID, bundle
+ID, idempotency key, upload status, latest run status, and redacted errors.
 
 Inspection helpers read existing metadata, briefing, and data package files.
 Missing metadata directories return no inspection records or no prior snapshot
diff --git a/docs/operations.md b/docs/operations.md
index ed13622..da8699b 100644
--- a/docs/operations.md
+++ b/docs/operations.md
@@ -150,16 +150,18 @@ successfully generated report: the managed Markdown report path recorded in the
 report result and metadata. Extra copies written by `--out` or `--out-dir` are
 operator conveniences only.
 
-The default bundle ID is derived from producer name, location ID, report ID, and
-RunID:
+The rendered pipeline ID selects the configured distributor `http_upload`
+workflow. The default bundle ID is a stable logical source identity derived from
+producer name, location ID, and report ID:
 
 ```text
-weatherreporter.{location_id}.{report_id}.{run_id}
+weatherreporter.{location_id}.{report_id}
 ```
 
-The default idempotency key is the rendered bundle ID. The default bundle path
-for the Markdown file is the report definition's batch output name, such as
-`daily.md`, `tomorrow.md`, `three-day.md`, or `weekend.md`.
+The default idempotency key appends RunID to the rendered bundle ID so each
+report generation has a distinct retry identity. The default bundle path for the
+Markdown file is the report definition's batch output name, such as `daily.md`,
+`tomorrow.md`, `three-day.md`, or `weekend.md`.
 
 Notification happens after final metadata save. Weather API, briefing,
 data-package, render preflight, Scriptorium run, and metadata-save failures do
@@ -168,14 +170,14 @@ other reports continue, the failed report includes notification fields in the
 JSON summary, and the batch returns nonzero.
 
 Each notification attempt writes a debug artifact under `notifications/`. The
-artifact records the rendered bundle ID, idempotency key, managed source path,
-bundle-relative path, bundle created timestamp, accepted upload response, and
-the latest distributor run status response when available. Weatherreporter polls
-status until distributor reports `succeeded` or `failed`, or until the configured
-notification timeout expires. The run status includes the distributor status,
-error text, and raw run report JSON, which can show actions such as
-`replace_older`, `skip_same`, `skip_destination_newer`, or `failed`. Token values
-are not written.
+artifact records the rendered pipeline ID, bundle ID, idempotency key, managed
+source path, bundle-relative path, bundle created timestamp, accepted upload
+response, and the latest distributor run status response when available.
+Weatherreporter polls status until distributor reports `succeeded` or `failed`,
+or until the configured notification timeout expires. The run status includes
+the distributor status, error text, and raw run report JSON, which can show
+actions such as `replace_older`, `skip_same`, `skip_destination_newer`, or
+`failed`. Token values are not written.
 
 Weatherreporter is responsible for selecting the managed Markdown report,
 constructing a source bundle, and submitting it to the configured distributor
diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md
index 75b9a90..98e3c0a 100644
--- a/docs/troubleshooting.md
+++ b/docs/troubleshooting.md
@@ -240,17 +240,17 @@ Relevant docs: [Configuration reference](config.md),
 
 Symptom: notification fails with idempotency conflict context.
 
-Likely cause: the same idempotency key was reused for different bundle content.
-By default the key derives from bundle ID, which includes location ID, report
-ID, and RunID.
+Likely cause: the same idempotency key was reused for different bundle content
+within the same distributor token and pipeline. By default the bundle ID is a
+stable report-stream identity and the idempotency key appends RunID.
 
-Diagnostic: inspect the failed batch JSON or stderr line for bundle and
-idempotency context. Compare the configured templates with the report RunID and
-report path.
+Diagnostic: inspect the failed batch JSON or stderr line for pipeline, bundle,
+and idempotency context. Compare the configured templates with the report RunID
+and report path.
 
 Also inspect the notification artifact linked from metadata. It records the
-rendered bundle ID, idempotency key, upload result, distributor run status,
-status error, and raw run report JSON when available.
+rendered pipeline ID, bundle ID, idempotency key, upload result, distributor run
+status, status error, and raw run report JSON when available.
 
 Safe fix: keep idempotency templates stable for retries of the same generated
 report, but do not reuse the same rendered key for different generated report
@@ -264,11 +264,12 @@ Relevant docs: [Operations guide](operations.md),
 Symptom: notification fails with distributor upload rejection, HTTP status, or
 bundle validation context.
 
-Likely cause: the distributor endpoint rejected the token, bundle ID,
-idempotency key, source file, or bundle path.
+Likely cause: the distributor endpoint rejected the token, pipeline ID, bundle
+ID, idempotency key, source file, or bundle path.
 
 Diagnostic: inspect stdout JSON or stderr status lines for
 `notificationError`. Confirm `notify.distributor.endpoint`,
+`notify.distributor.pipeline_id_template`,
 `notify.distributor.report_path_template`, and token configuration. Token values
 are redacted from weatherreporter errors.
 
diff --git a/examples/config.yml b/examples/config.yml
index a72571b..8195bd8 100644
--- a/examples/config.yml
+++ b/examples/config.yml
@@ -21,6 +21,7 @@ notify:
     token_env: DISTRIBUTOR_UPLOAD_TOKEN
     timeout: 30s
     failure_policy: error
+    pipeline_id_template: "weatherreporter.{artifact_group}"
     bundle_id_template: "weatherreporter.{location_id}.{report_id}"
     idempotency_key_template: "{bundle_id}.{run_id}"
     report_path_template: "{batch_output_name}"
diff --git a/go.mod b/go.mod
index 2278527..5d64a6c 100644
--- a/go.mod
+++ b/go.mod
@@ -4,4 +4,4 @@ go 1.26
 
 require gopkg.in/yaml.v3 v3.0.1
 
-require gitea.maximumdirect.net/eric/distributor v0.4.0
+require gitea.maximumdirect.net/eric/distributor v0.5.0
diff --git a/go.sum b/go.sum
index 18a2d0a..7f0f065 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,5 @@
-gitea.maximumdirect.net/eric/distributor v0.4.0 h1:SRrTFjVLMv4wFZmMLwnaYtJaQlQ/wsB4OzcaSEEb524=
-gitea.maximumdirect.net/eric/distributor v0.4.0/go.mod h1:G03FCFZPHpsUKC6SeMgTdbfNRpPQBdyTtDUj04e1Tu8=
+gitea.maximumdirect.net/eric/distributor v0.5.0 h1:+al7Bw+kMv6V35a3Sm5rUtCTQhwOn5b9x3RsclPMKJk=
+gitea.maximumdirect.net/eric/distributor v0.5.0/go.mod h1:G03FCFZPHpsUKC6SeMgTdbfNRpPQBdyTtDUj04e1Tu8=
 github.com/aws/aws-sdk-go-v2 v1.41.9 h1:/rYeyO2+HrMztAmxAq9++XJtFMqSIpSsNA0yDGALYq4=
 github.com/aws/aws-sdk-go-v2 v1.41.9/go.mod h1:+HsoOEX80qAVUitj1A2DhCNTjmb3edVyuDypb6LNEeo=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.11 h1:h5+3VT69KUBK24grGuuA5saDJTj2IIjLb9au668Fo5I=
diff --git a/internal/adapters/distributor/client.go b/internal/adapters/distributor/client.go
index 8947b0b..c1cdba8 100644
--- a/internal/adapters/distributor/client.go
+++ b/internal/adapters/distributor/client.go
@@ -25,6 +25,7 @@ type Client struct {
 }
 
 type UploadRequest struct {
+	PipelineID     string
 	BundleID       string
 	IdempotencyKey string
 	SourcePath     string
@@ -77,6 +78,7 @@ type uploadClient interface {
 }
 
 type uploadFilesOptions struct {
+	PipelineID     string
 	BundleID       string
 	IdempotencyKey string
 	SourcePath     string
@@ -128,6 +130,9 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
 	if c.TokenEnv == "" {
 		return UploadResult{}, fmt.Errorf("distributor token environment variable is required")
 	}
+	if req.PipelineID == "" {
+		return UploadResult{}, fmt.Errorf("distributor pipeline id is required")
+	}
 	if req.BundleID == "" {
 		return UploadResult{}, fmt.Errorf("distributor bundle id is required")
 	}
@@ -165,6 +170,7 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
 	defer cancel()
 
 	result, err := uploadClient.UploadFiles(runCtx, uploadFilesOptions{
+		PipelineID:     req.PipelineID,
 		BundleID:       req.BundleID,
 		IdempotencyKey: req.IdempotencyKey,
 		SourcePath:     req.SourcePath,
@@ -174,6 +180,7 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
 	if err != nil {
 		return UploadResult{}, wrapUploadError(err, uploadErrorContext{
 			Endpoint:       c.Endpoint,
+			PipelineID:     req.PipelineID,
 			BundleID:       req.BundleID,
 			IdempotencyKey: req.IdempotencyKey,
 			SourcePath:     req.SourcePath,
@@ -265,6 +272,7 @@ func newDistributorUploadClient(endpoint, token string, timeout time.Duration) (
 
 func (c distributorUploadClient) UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error) {
 	result, err := c.client.UploadFiles(ctx, distributorupload.UploadFilesOptions{
+		PipelineID:     opts.PipelineID,
 		ID:             opts.BundleID,
 		Created:        opts.CreatedAt,
 		IdempotencyKey: opts.IdempotencyKey,
@@ -300,6 +308,7 @@ func (c distributorUploadClient) Status(ctx context.Context, runID string) (runS
 
 type uploadErrorContext struct {
 	Endpoint       string
+	PipelineID     string
 	BundleID       string
 	IdempotencyKey string
 	SourcePath     string
@@ -313,10 +322,10 @@ func wrapUploadError(err error, ctx uploadErrorContext) error {
 	err = redactToken(err, ctx.Token)
 	if isConflict {
 		return &IdempotencyConflictError{
-			Err: fmt.Errorf("upload distributor bundle %q to endpoint %q with idempotency key %q from source %q as bundle path %q: idempotency conflict: %w", ctx.BundleID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePath, ctx.BundlePath, err),
+			Err: fmt.Errorf("upload distributor bundle %q to pipeline %q at endpoint %q with idempotency key %q from source %q as bundle path %q: idempotency conflict: %w", ctx.BundleID, ctx.PipelineID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePath, ctx.BundlePath, err),
 		}
 	}
-	return fmt.Errorf("upload distributor bundle %q to endpoint %q with idempotency key %q from source %q as bundle path %q: %w", ctx.BundleID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePath, ctx.BundlePath, err)
+	return fmt.Errorf("upload distributor bundle %q to pipeline %q at endpoint %q with idempotency key %q from source %q as bundle path %q: %w", ctx.BundleID, ctx.PipelineID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePath, ctx.BundlePath, err)
 }
 
 func redactToken(err error, token string) error {
diff --git a/internal/adapters/distributor/client_test.go b/internal/adapters/distributor/client_test.go
index 6608b24..a6cf9dc 100644
--- a/internal/adapters/distributor/client_test.go
+++ b/internal/adapters/distributor/client_test.go
@@ -30,6 +30,7 @@ func TestUploadUsesConfiguredClientAndSingleFile(t *testing.T) {
 	client := newClient(cfg, factory.newClient)
 
 	result, err := client.Upload(context.Background(), UploadRequest{
+		PipelineID:     "weatherreporter.daily",
 		BundleID:       "weatherreporter.home.daily.run",
 		IdempotencyKey: "weatherreporter.home.daily.run",
 		SourcePath:     "/tmp/report.md",
@@ -55,6 +56,9 @@ func TestUploadUsesConfiguredClientAndSingleFile(t *testing.T) {
 		t.Fatalf("factory timeout = %s, want 15s", factory.timeout)
 	}
 	got := factory.client.opts
+	if got.PipelineID != "weatherreporter.daily" {
+		t.Fatalf("PipelineID = %q, want weatherreporter.daily", got.PipelineID)
+	}
 	if got.BundleID != "weatherreporter.home.daily.run" {
 		t.Fatalf("BundleID = %q, want weatherreporter.home.daily.run", got.BundleID)
 	}
@@ -91,6 +95,13 @@ func TestUploadRejectsMissingInputs(t *testing.T) {
 			},
 			wantErr: "token environment variable",
 		},
+		{
+			name: "PipelineID",
+			mutate: func(c *Client, req *UploadRequest) {
+				req.PipelineID = ""
+			},
+			wantErr: "pipeline id is required",
+		},
 		{
 			name: "SourcePath",
 			mutate: func(c *Client, req *UploadRequest) {
@@ -170,7 +181,7 @@ func TestUploadWrapsUploadFailureWithContextWithoutToken(t *testing.T) {
 	if err == nil {
 		t.Fatal("Upload() error = nil, want error")
 	}
-	for _, want := range []string{cfg.Endpoint, req.BundleID, req.IdempotencyKey, req.SourcePath, req.BundlePath} {
+	for _, want := range []string{cfg.Endpoint, req.PipelineID, req.BundleID, req.IdempotencyKey, req.SourcePath, req.BundlePath} {
 		if !strings.Contains(err.Error(), want) {
 			t.Fatalf("error = %q, want context %q", err.Error(), want)
 		}
@@ -318,6 +329,7 @@ func TestUploadPreservesIdempotencyConflictDiagnosis(t *testing.T) {
 
 func validUploadRequest() UploadRequest {
 	return UploadRequest{
+		PipelineID:     "weatherreporter.daily",
 		BundleID:       "weatherreporter.home.daily.run",
 		IdempotencyKey: "weatherreporter.home.daily.run",
 		SourcePath:     "/tmp/report.md",
diff --git a/internal/app/app.go b/internal/app/app.go
index 0334dcd..ff2b84b 100644
--- a/internal/app/app.go
+++ b/internal/app/app.go
@@ -114,24 +114,25 @@ type BatchResult struct {
 }
 
 type BatchReportResult struct {
-	ReportID           report.ID       `json:"reportId"`
-	ReportName         string          `json:"reportName"`
-	PromptID           string          `json:"promptId"`
-	RunID              string          `json:"runId"`
-	Status             string          `json:"status"`
-	Error              string          `json:"error,omitempty"`
-	NotificationStatus string          `json:"notificationStatus,omitempty"`
-	NotificationRunID  string          `json:"notificationRunId,omitempty"`
-	NotificationError  string          `json:"notificationError,omitempty"`
-	NotificationPath   string          `json:"notificationPath,omitempty"`
-	GeneratedAt        time.Time       `json:"generatedAt"`
-	ValidPeriod        timeutil.Period `json:"validPeriod"`
-	BriefingPath       string          `json:"briefingPath,omitempty"`
-	DataPackagePath    string          `json:"dataPackagePath,omitempty"`
-	PreflightPath      string          `json:"preflightPath,omitempty"`
-	ReportPath         string          `json:"reportPath,omitempty"`
-	OutputPath         string          `json:"outputPath,omitempty"`
-	MetadataPath       string          `json:"metadataPath,omitempty"`
+	ReportID               report.ID       `json:"reportId"`
+	ReportName             string          `json:"reportName"`
+	PromptID               string          `json:"promptId"`
+	RunID                  string          `json:"runId"`
+	Status                 string          `json:"status"`
+	Error                  string          `json:"error,omitempty"`
+	NotificationStatus     string          `json:"notificationStatus,omitempty"`
+	NotificationRunID      string          `json:"notificationRunId,omitempty"`
+	NotificationPipelineID string          `json:"notificationPipelineId,omitempty"`
+	NotificationError      string          `json:"notificationError,omitempty"`
+	NotificationPath       string          `json:"notificationPath,omitempty"`
+	GeneratedAt            time.Time       `json:"generatedAt"`
+	ValidPeriod            timeutil.Period `json:"validPeriod"`
+	BriefingPath           string          `json:"briefingPath,omitempty"`
+	DataPackagePath        string          `json:"dataPackagePath,omitempty"`
+	PreflightPath          string          `json:"preflightPath,omitempty"`
+	ReportPath             string          `json:"reportPath,omitempty"`
+	OutputPath             string          `json:"outputPath,omitempty"`
+	MetadataPath           string          `json:"metadataPath,omitempty"`
 }
 
 type BatchError struct {
@@ -157,6 +158,7 @@ type Notifier interface {
 type NotificationRequest struct {
 	ReportID       report.ID
 	RunID          string
+	PipelineID     string
 	BundleID       string
 	IdempotencyKey string
 	ReportPath     string
@@ -280,6 +282,7 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro
 				if errors.As(err, &notificationErr) {
 					item.NotificationStatus = "failed"
 					item.NotificationError = notificationErr.Error()
+					item.NotificationPipelineID = notificationErr.Request.PipelineID
 					if paths, pathErr := store.Paths(resolved); pathErr == nil {
 						item.NotificationPath = paths.Notification
 					}
@@ -297,6 +300,7 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro
 				if reportResult.Notification != nil {
 					item.NotificationStatus = reportResult.Notification.Status
 					item.NotificationRunID = reportResult.Notification.RunID
+					item.NotificationPipelineID = reportResult.Notification.PipelineID
 				}
 				result.Succeeded++
 			}
@@ -638,6 +642,10 @@ func buildNotificationRequest(cfg config.Config, resolved report.Resolved, repor
 		return NotificationRequest{}, err
 	}
 	values.BundleID = bundleID
+	pipelineID, err := config.RenderDistributorPipelineID(cfg.Notify.Distributor.PipelineIDTemplate, values)
+	if err != nil {
+		return NotificationRequest{}, err
+	}
 	idempotencyKey, err := config.RenderDistributorIdempotencyKey(cfg.Notify.Distributor.IdempotencyKeyTemplate, values)
 	if err != nil {
 		return NotificationRequest{}, err
@@ -649,6 +657,7 @@ func buildNotificationRequest(cfg config.Config, resolved report.Resolved, repor
 	return NotificationRequest{
 		ReportID:       resolved.Definition.ID,
 		RunID:          metadata.RunID,
+		PipelineID:     pipelineID,
 		BundleID:       bundleID,
 		IdempotencyKey: idempotencyKey,
 		ReportPath:     reportPath,
@@ -667,6 +676,7 @@ func saveNotificationArtifact(ctx context.Context, store state.Store, resolved r
 		ReportID:       resolved.Definition.ID,
 		AttemptedAt:    time.Now(),
 		Endpoint:       cfg.Notify.Distributor.Endpoint,
+		PipelineID:     req.PipelineID,
 		BundleID:       req.BundleID,
 		IdempotencyKey: req.IdempotencyKey,
 		SourcePath:     req.ReportPath,
@@ -716,6 +726,7 @@ type distributorNotifier struct {
 
 func (n distributorNotifier) Notify(ctx context.Context, req NotificationRequest) (*NotificationResult, error) {
 	result, err := n.client.Upload(ctx, distributoradapter.UploadRequest{
+		PipelineID:     req.PipelineID,
 		BundleID:       req.BundleID,
 		IdempotencyKey: req.IdempotencyKey,
 		SourcePath:     req.ReportPath,
@@ -723,6 +734,7 @@ func (n distributorNotifier) Notify(ctx context.Context, req NotificationRequest
 		CreatedAt:      req.CreatedAt,
 	})
 	notification := &NotificationResult{
+		PipelineID:     req.PipelineID,
 		BundleID:       req.BundleID,
 		IdempotencyKey: req.IdempotencyKey,
 		RunID:          result.RunID,
@@ -731,7 +743,9 @@ func (n distributorNotifier) Notify(ctx context.Context, req NotificationRequest
 		StatusError:    result.StatusError,
 	}
 	if result.RunStatus != nil {
-		notification.PipelineID = result.RunStatus.PipelineID
+		if result.RunStatus.PipelineID != "" {
+			notification.PipelineID = result.RunStatus.PipelineID
+		}
 		notification.AcceptedAt = result.RunStatus.AcceptedAt
 		notification.StartedAt = result.RunStatus.StartedAt
 		notification.FinishedAt = result.RunStatus.FinishedAt
diff --git a/internal/app/app_test.go b/internal/app/app_test.go
index ed0ec3c..47c01b9 100644
--- a/internal/app/app_test.go
+++ b/internal/app/app_test.go
@@ -288,6 +288,7 @@ func TestGenerateReportNotifiesManagedReportPath(t *testing.T) {
 	cfg := dailyTestConfig(t, server)
 	cfg.Workspace.Root = t.TempDir()
 	cfg.Notify.Distributor.Enabled = true
+	cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
 	resolved, err := ResolveGenerate(GenerateRequest{
 		Config: cfg,
 		Report: ReportDaily,
@@ -330,8 +331,12 @@ func TestGenerateReportNotifiesManagedReportPath(t *testing.T) {
 	if err != nil {
 		t.Fatalf("read notification artifact: %v", err)
 	}
-	if !strings.Contains(string(notificationData), `"replace_older"`) || !strings.Contains(string(notificationData), `"bundleCreated"`) {
-		t.Fatalf("notification artifact missing status report or created timestamp:\n%s", string(notificationData))
+	var notificationArtifact state.DistributorNotificationArtifact
+	if err := json.Unmarshal(notificationData, &notificationArtifact); err != nil {
+		t.Fatalf("decode notification artifact: %v", err)
+	}
+	if notificationArtifact.PipelineID != "weatherreporter.daily" || notificationArtifact.BundleCreated.IsZero() || notificationArtifact.RunStatus == nil || !strings.Contains(string(notificationArtifact.RunStatus.Report), "replace_older") {
+		t.Fatalf("notification artifact = %#v, want requested pipeline, status report, and created timestamp", notificationArtifact)
 	}
 	if len(notifier.requests) != 1 {
 		t.Fatalf("notification requests = %d, want 1", len(notifier.requests))
@@ -346,11 +351,14 @@ func TestGenerateReportNotifiesManagedReportPath(t *testing.T) {
 	if req.BundlePath != "daily.md" {
 		t.Fatalf("notification BundlePath = %q, want daily.md", req.BundlePath)
 	}
-	if req.BundleID != "weatherreporter.home.daily_today."+result.Metadata.RunID {
+	if req.PipelineID != "weatherreporter.daily" {
+		t.Fatalf("notification PipelineID = %q, want rendered pipeline", req.PipelineID)
+	}
+	if req.BundleID != "weatherreporter.home.daily_today" {
 		t.Fatalf("notification BundleID = %q, want default template", req.BundleID)
 	}
-	if req.IdempotencyKey != req.BundleID {
-		t.Fatalf("IdempotencyKey = %q, want bundle id %q", req.IdempotencyKey, req.BundleID)
+	if req.IdempotencyKey != req.BundleID+"."+result.Metadata.RunID {
+		t.Fatalf("IdempotencyKey = %q, want per-run key", req.IdempotencyKey)
 	}
 	if req.RunID != result.Metadata.RunID {
 		t.Fatalf("notification RunID = %q, want report run id %q", req.RunID, result.Metadata.RunID)
@@ -365,6 +373,7 @@ func TestGenerateReportNotificationFailureFailsReport(t *testing.T) {
 	cfg := dailyTestConfig(t, server)
 	cfg.Workspace.Root = t.TempDir()
 	cfg.Notify.Distributor.Enabled = true
+	cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
 	resolved, err := ResolveGenerate(GenerateRequest{
 		Config: cfg,
 		Report: ReportDaily,
@@ -441,6 +450,7 @@ func TestGenerateReportDoesNotNotifyAfterRenderOrRunFailure(t *testing.T) {
 			cfg := dailyTestConfig(t, server)
 			cfg.Workspace.Root = t.TempDir()
 			cfg.Notify.Distributor.Enabled = true
+			cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
 			resolved, err := ResolveGenerate(GenerateRequest{
 				Config: cfg,
 				Report: ReportDaily,
@@ -473,6 +483,7 @@ func TestGenerateReportDoesNotNotifyAfterFetchFailure(t *testing.T) {
 	cfg.WeatherAPI.Timezone = "America/Chicago"
 	cfg.Workspace.Root = t.TempDir()
 	cfg.Notify.Distributor.Enabled = true
+	cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
 	resolved, err := ResolveGenerate(GenerateRequest{
 		Config: cfg,
 		Report: ReportDaily,
@@ -1257,6 +1268,7 @@ func TestRunBatchContinuesAfterNotificationFailure(t *testing.T) {
 	cfg.WeatherAPI.Timezone = "America/Chicago"
 	cfg.Workspace.Root = t.TempDir()
 	cfg.Notify.Distributor.Enabled = true
+	cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
 	notifier := &recordingNotifier{
 		errByReport: map[report.ID]error{
 			report.ThreeDay: errors.New("distributor unavailable"),
@@ -1289,6 +1301,9 @@ func TestRunBatchContinuesAfterNotificationFailure(t *testing.T) {
 			if item.NotificationStatus != "failed" {
 				t.Fatalf("3-day notification status = %q, want failed", item.NotificationStatus)
 			}
+			if item.NotificationPipelineID != "weatherreporter.three-day" {
+				t.Fatalf("3-day notification pipeline = %q, want weatherreporter.three-day", item.NotificationPipelineID)
+			}
 			if !strings.Contains(item.NotificationError, "distributor unavailable") {
 				t.Fatalf("3-day notification error = %q, want distributor unavailable", item.NotificationError)
 			}
@@ -1300,6 +1315,9 @@ func TestRunBatchContinuesAfterNotificationFailure(t *testing.T) {
 		if item.NotificationStatus != "accepted" {
 			t.Fatalf("report %s notification status = %q, want accepted", item.ReportID, item.NotificationStatus)
 		}
+		if item.NotificationPipelineID == "" {
+			t.Fatalf("report %s notification pipeline is empty", item.ReportID)
+		}
 	}
 	if !failedThreeDay {
 		t.Fatalf("reports = %#v, want notification failure on 3-day item", result.Reports)
@@ -1500,9 +1518,13 @@ func (n *recordingNotifier) Notify(_ context.Context, req NotificationRequest) (
 		if result.IdempotencyKey == "" {
 			result.IdempotencyKey = req.IdempotencyKey
 		}
+		if result.PipelineID == "" {
+			result.PipelineID = req.PipelineID
+		}
 		return &result, nil
 	}
 	return &NotificationResult{
+		PipelineID:     req.PipelineID,
 		BundleID:       req.BundleID,
 		IdempotencyKey: req.IdempotencyKey,
 		Status:         "accepted",
diff --git a/internal/cli/root_test.go b/internal/cli/root_test.go
index 12d674c..dd38f88 100644
--- a/internal/cli/root_test.go
+++ b/internal/cli/root_test.go
@@ -469,7 +469,12 @@ func TestRunEveningUsesOutputDirectoryAndSummary(t *testing.T) {
 func TestRunEveningReportsNotificationSuccess(t *testing.T) {
 	server := dailyServer(t)
 	distributorServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/upload" {
+		if r.URL.Path == "/runs/distributor-run-1" {
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = w.Write([]byte(`{"run_id":"distributor-run-1","pipeline_id":"weatherreporter.daily","status":"succeeded","report":{"actions":[{"action":"replace_older"}]}}`))
+			return
+		}
+		if r.URL.Path != "/v1/pipelines/weatherreporter.daily/upload" {
 			http.NotFound(w, r)
 			return
 		}
@@ -481,7 +486,7 @@ func TestRunEveningReportsNotificationSuccess(t *testing.T) {
 	scriptoriumPath := writeFakeScriptorium(t, tempDir)
 	configPath := filepath.Join(tempDir, "config.yml")
 	workspaceRoot := filepath.Join(tempDir, "workspace")
-	configBody := "weather_api:\n  base_url: " + server.URL + "/\n  timezone: America/Chicago\nscriptorium:\n  binary: " + scriptoriumPath + "\nworkspace:\n  root: " + workspaceRoot + "\nnotify:\n  distributor:\n    enabled: true\n    endpoint: " + distributorServer.URL + "\n    token_env: CLI_DISTRIBUTOR_TOKEN\n"
+	configBody := "weather_api:\n  base_url: " + server.URL + "/\n  timezone: America/Chicago\nscriptorium:\n  binary: " + scriptoriumPath + "\nworkspace:\n  root: " + workspaceRoot + "\nnotify:\n  distributor:\n    enabled: true\n    endpoint: " + distributorServer.URL + "\n    token_env: CLI_DISTRIBUTOR_TOKEN\n    pipeline_id_template: weatherreporter.{artifact_group}\n"
 	if err := os.WriteFile(configPath, []byte(configBody), 0o600); err != nil {
 		t.Fatalf("write config: %v", err)
 	}
@@ -505,10 +510,10 @@ func TestRunEveningReportsNotificationSuccess(t *testing.T) {
 	if len(summary.Reports) != 1 {
 		t.Fatalf("reports = %#v, want one report", summary.Reports)
 	}
-	if summary.Reports[0].NotificationStatus != "accepted" || summary.Reports[0].NotificationRunID != "distributor-run-1" {
+	if summary.Reports[0].NotificationStatus != "succeeded" || summary.Reports[0].NotificationRunID != "distributor-run-1" || summary.Reports[0].NotificationPipelineID != "weatherreporter.daily" {
 		t.Fatalf("notification fields = %#v", summary.Reports[0])
 	}
-	if !strings.Contains(stderr.String(), `notificationStatus="accepted"`) || !strings.Contains(stderr.String(), `notificationRunId="distributor-run-1"`) {
+	if !strings.Contains(stderr.String(), `notificationStatus="succeeded"`) || !strings.Contains(stderr.String(), `notificationRunId="distributor-run-1"`) {
 		t.Fatalf("stderr missing notification fields:\n%s", stderr.String())
 	}
 	if strings.Contains(stdout.String(), "cli-secret-token") || strings.Contains(stderr.String(), "cli-secret-token") {
@@ -527,7 +532,7 @@ func TestRunEveningReportsNotificationFailureWithoutToken(t *testing.T) {
 	scriptoriumPath := writeFakeScriptorium(t, tempDir)
 	configPath := filepath.Join(tempDir, "config.yml")
 	workspaceRoot := filepath.Join(tempDir, "workspace")
-	configBody := "weather_api:\n  base_url: " + server.URL + "/\n  timezone: America/Chicago\nscriptorium:\n  binary: " + scriptoriumPath + "\nworkspace:\n  root: " + workspaceRoot + "\nnotify:\n  distributor:\n    enabled: true\n    endpoint: " + distributorServer.URL + "\n    token_env: CLI_DISTRIBUTOR_TOKEN\n"
+	configBody := "weather_api:\n  base_url: " + server.URL + "/\n  timezone: America/Chicago\nscriptorium:\n  binary: " + scriptoriumPath + "\nworkspace:\n  root: " + workspaceRoot + "\nnotify:\n  distributor:\n    enabled: true\n    endpoint: " + distributorServer.URL + "\n    token_env: CLI_DISTRIBUTOR_TOKEN\n    pipeline_id_template: weatherreporter.{artifact_group}\n"
 	if err := os.WriteFile(configPath, []byte(configBody), 0o600); err != nil {
 		t.Fatalf("write config: %v", err)
 	}
diff --git a/internal/config/config.go b/internal/config/config.go
index 72203c7..093cee0 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -56,6 +56,7 @@ type DistributorNotifyConfig struct {
 	TokenEnv               string              `yaml:"token_env"`
 	Timeout                time.Duration       `yaml:"timeout"`
 	FailurePolicy          NotifyFailurePolicy `yaml:"failure_policy"`
+	PipelineIDTemplate     string              `yaml:"pipeline_id_template"`
 	BundleIDTemplate       string              `yaml:"bundle_id_template"`
 	IdempotencyKeyTemplate string              `yaml:"idempotency_key_template"`
 	ReportPathTemplate     string              `yaml:"report_path_template"`
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index 9c345fe..f4f7f80 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -44,10 +44,13 @@ func TestDefaults(t *testing.T) {
 	if cfg.Notify.Distributor.FailurePolicy != NotifyFailureError {
 		t.Fatalf("Notify.Distributor.FailurePolicy = %q, want error", cfg.Notify.Distributor.FailurePolicy)
 	}
-	if cfg.Notify.Distributor.BundleIDTemplate != "weatherreporter.{location_id}.{report_id}.{run_id}" {
+	if cfg.Notify.Distributor.PipelineIDTemplate != "" {
+		t.Fatalf("Notify.Distributor.PipelineIDTemplate = %q, want empty", cfg.Notify.Distributor.PipelineIDTemplate)
+	}
+	if cfg.Notify.Distributor.BundleIDTemplate != "weatherreporter.{location_id}.{report_id}" {
 		t.Fatalf("Notify.Distributor.BundleIDTemplate = %q, want default", cfg.Notify.Distributor.BundleIDTemplate)
 	}
-	if cfg.Notify.Distributor.IdempotencyKeyTemplate != "{bundle_id}" {
+	if cfg.Notify.Distributor.IdempotencyKeyTemplate != "{bundle_id}.{run_id}" {
 		t.Fatalf("Notify.Distributor.IdempotencyKeyTemplate = %q, want default", cfg.Notify.Distributor.IdempotencyKeyTemplate)
 	}
 	if cfg.Notify.Distributor.ReportPathTemplate != "{batch_output_name}" {
@@ -76,6 +79,9 @@ func TestLoadExampleConfig(t *testing.T) {
 	if cfg.Location.ID != "home" || cfg.Location.Name != "Brentwood" || cfg.Location.Region != "St. Louis Metro" {
 		t.Fatalf("Location = %#v, want example location", cfg.Location)
 	}
+	if cfg.Notify.Distributor.PipelineIDTemplate != "weatherreporter.{artifact_group}" {
+		t.Fatalf("PipelineIDTemplate = %q, want example pipeline template", cfg.Notify.Distributor.PipelineIDTemplate)
+	}
 }
 
 func TestLoadMinimalExampleConfig(t *testing.T) {
@@ -200,6 +206,27 @@ func TestEnabledDistributorNotifyValidation(t *testing.T) {
 			},
 			wantErr: "notify.distributor.failure_policy",
 		},
+		{
+			name: "PipelineTemplateEmpty",
+			mutate: func(cfg *Config) {
+				cfg.Notify.Distributor.PipelineIDTemplate = ""
+			},
+			wantErr: "notify.distributor.pipeline_id_template",
+		},
+		{
+			name: "PipelineTemplateUnknown",
+			mutate: func(cfg *Config) {
+				cfg.Notify.Distributor.PipelineIDTemplate = "{unknown}"
+			},
+			wantErr: "notify.distributor.pipeline_id_template",
+		},
+		{
+			name: "PipelineTemplateRenderedEmpty",
+			mutate: func(cfg *Config) {
+				cfg.Notify.Distributor.PipelineIDTemplate = "   "
+			},
+			wantErr: "notify.distributor.pipeline_id_template",
+		},
 		{
 			name: "BundleTemplate",
 			mutate: func(cfg *Config) {
@@ -234,6 +261,7 @@ func TestEnabledDistributorNotifyValidation(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			cfg := Defaults()
 			cfg.Notify.Distributor.Enabled = true
+			cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}"
 			tt.mutate(&cfg)
 
 			err := Validate(cfg)
@@ -254,23 +282,31 @@ func TestDistributorTemplateRendering(t *testing.T) {
 		RunID:           "20260607T120000Z",
 		ArtifactGroup:   "daily",
 		BatchOutputName: "daily.md",
-		BundleID:        "weatherreporter.home.daily.20260607T120000Z",
+		BundleID:        "weatherreporter.home.daily",
 	}
 
-	bundleID, err := RenderDistributorBundleID("weatherreporter.{location_id}.{report_id}.{run_id}", values)
+	bundleID, err := RenderDistributorBundleID("weatherreporter.{location_id}.{report_id}", values)
 	if err != nil {
 		t.Fatalf("RenderDistributorBundleID() error = %v", err)
 	}
-	if bundleID != "weatherreporter.home.daily.20260607T120000Z" {
+	if bundleID != "weatherreporter.home.daily" {
 		t.Fatalf("bundleID = %q, want rendered value", bundleID)
 	}
 
-	idempotencyKey, err := RenderDistributorIdempotencyKey("{bundle_id}", values)
+	pipelineID, err := RenderDistributorPipelineID("weatherreporter.{artifact_group}.{bundle_id}", values)
+	if err != nil {
+		t.Fatalf("RenderDistributorPipelineID() error = %v", err)
+	}
+	if pipelineID != "weatherreporter.daily.weatherreporter.home.daily" {
+		t.Fatalf("pipelineID = %q, want rendered pipeline ID", pipelineID)
+	}
+
+	idempotencyKey, err := RenderDistributorIdempotencyKey("{bundle_id}.{run_id}", values)
 	if err != nil {
 		t.Fatalf("RenderDistributorIdempotencyKey() error = %v", err)
 	}
 	if idempotencyKey != "weatherreporter.home.daily.20260607T120000Z" {
-		t.Fatalf("idempotencyKey = %q, want rendered bundle ID", idempotencyKey)
+		t.Fatalf("idempotencyKey = %q, want rendered run key", idempotencyKey)
 	}
 
 	reportPath, err := RenderDistributorReportPath("reports/{batch_output_name}", values)
@@ -375,7 +411,8 @@ func TestLoadFileLoadsSecretsBeforeReturningNotifyConfig(t *testing.T) {
 		"  directory: " + secretsDir + "\n" +
 		"notify:\n" +
 		"  distributor:\n" +
-		"    enabled: true\n"
+		"    enabled: true\n" +
+		"    pipeline_id_template: weatherreporter.{artifact_group}\n"
 	if err := os.WriteFile(path, []byte(configYAML), 0o600); err != nil {
 		t.Fatalf("write config fixture: %v", err)
 	}
diff --git a/internal/config/defaults.go b/internal/config/defaults.go
index c4f1c8f..30f176e 100644
--- a/internal/config/defaults.go
+++ b/internal/config/defaults.go
@@ -28,8 +28,9 @@ func Defaults() Config {
 				TokenEnv:               "DISTRIBUTOR_UPLOAD_TOKEN",
 				Timeout:                30 * time.Second,
 				FailurePolicy:          NotifyFailureError,
-				BundleIDTemplate:       "weatherreporter.{location_id}.{report_id}.{run_id}",
-				IdempotencyKeyTemplate: "{bundle_id}",
+				PipelineIDTemplate:     "",
+				BundleIDTemplate:       "weatherreporter.{location_id}.{report_id}",
+				IdempotencyKeyTemplate: "{bundle_id}.{run_id}",
 				ReportPathTemplate:     "{batch_output_name}",
 			},
 		},
diff --git a/internal/config/notify_templates.go b/internal/config/notify_templates.go
index c5b6b17..3f701d5 100644
--- a/internal/config/notify_templates.go
+++ b/internal/config/notify_templates.go
@@ -32,10 +32,23 @@ var distributorIdempotencyTemplateVariables = map[string]struct{}{
 	"bundle_id":         {},
 }
 
+var distributorPipelineTemplateVariables = distributorIdempotencyTemplateVariables
+
 func RenderDistributorBundleID(template string, values DistributorTemplateValues) (string, error) {
 	return renderDistributorTemplate("notify.distributor.bundle_id_template", template, values, distributorTemplateVariables)
 }
 
+func RenderDistributorPipelineID(template string, values DistributorTemplateValues) (string, error) {
+	rendered, err := renderDistributorTemplate("notify.distributor.pipeline_id_template", template, values, distributorPipelineTemplateVariables)
+	if err != nil {
+		return "", err
+	}
+	if strings.TrimSpace(rendered) == "" {
+		return "", fmt.Errorf("notify.distributor.pipeline_id_template renders an empty pipeline id")
+	}
+	return rendered, nil
+}
+
 func RenderDistributorIdempotencyKey(template string, values DistributorTemplateValues) (string, error) {
 	return renderDistributorTemplate("notify.distributor.idempotency_key_template", template, values, distributorIdempotencyTemplateVariables)
 }
diff --git a/internal/config/validate.go b/internal/config/validate.go
index 7032454..2e9ba77 100644
--- a/internal/config/validate.go
+++ b/internal/config/validate.go
@@ -100,6 +100,12 @@ func validateDistributorNotify(cfg DistributorNotifyConfig) error {
 	if cfg.FailurePolicy != NotifyFailureError {
 		return fmt.Errorf("notify.distributor.failure_policy must be error when enabled")
 	}
+	if cfg.PipelineIDTemplate == "" {
+		return fmt.Errorf("notify.distributor.pipeline_id_template is required when enabled")
+	}
+	if err := validateDistributorTemplate("notify.distributor.pipeline_id_template", cfg.PipelineIDTemplate, distributorPipelineTemplateVariables); err != nil {
+		return err
+	}
 	if cfg.BundleIDTemplate == "" {
 		return fmt.Errorf("notify.distributor.bundle_id_template is required when enabled")
 	}
@@ -122,6 +128,14 @@ func validateDistributorNotify(cfg DistributorNotifyConfig) error {
 		ArtifactGroup:   "artifact",
 		BatchOutputName: "report.md",
 	}
+	bundleID, err := RenderDistributorBundleID(cfg.BundleIDTemplate, values)
+	if err != nil {
+		return err
+	}
+	values.BundleID = bundleID
+	if _, err := RenderDistributorPipelineID(cfg.PipelineIDTemplate, values); err != nil {
+		return err
+	}
 	if _, err := RenderDistributorReportPath(cfg.ReportPathTemplate, values); err != nil {
 		return err
 	}
diff --git a/internal/state/filesystem_test.go b/internal/state/filesystem_test.go
index 4380a8e..997e796 100644
--- a/internal/state/filesystem_test.go
+++ b/internal/state/filesystem_test.go
@@ -65,6 +65,7 @@ func TestSaveArtifactsAndMetadataRoundTrip(t *testing.T) {
 		ReportID:       resolved.Definition.ID,
 		AttemptedAt:    resolved.GeneratedAt,
 		Endpoint:       "https://distributor.example.test",
+		PipelineID:     "weatherreporter.daily",
 		BundleID:       "weatherreporter.home.daily.run",
 		IdempotencyKey: "weatherreporter.home.daily.run",
 		SourcePath:     "/tmp/report.md",
@@ -102,7 +103,7 @@ func TestSaveArtifactsAndMetadataRoundTrip(t *testing.T) {
 	if err := json.Unmarshal(notificationData, &notification); err != nil {
 		t.Fatalf("decode notification: %v", err)
 	}
-	if notification.SchemaVersion != DistributorNotificationSchemaVersion || notification.RunStatus == nil || notification.RunStatus.Status != "succeeded" {
+	if notification.SchemaVersion != DistributorNotificationSchemaVersion || notification.PipelineID != "weatherreporter.daily" || notification.RunStatus == nil || notification.RunStatus.Status != "succeeded" {
 		t.Fatalf("notification = %#v, want persisted distributor status", notification)
 	}
 	paths, err := store.Paths(resolved)
diff --git a/internal/state/store.go b/internal/state/store.go
index d1771c0..e293dc1 100644
--- a/internal/state/store.go
+++ b/internal/state/store.go
@@ -45,6 +45,7 @@ type DistributorNotificationArtifact struct {
 	ReportID       report.ID                `json:"reportId"`
 	AttemptedAt    time.Time                `json:"attemptedAt"`
 	Endpoint       string                   `json:"endpoint"`
+	PipelineID     string                   `json:"pipelineId,omitempty"`
 	BundleID       string                   `json:"bundleId,omitempty"`
 	IdempotencyKey string                   `json:"idempotencyKey,omitempty"`
 	SourcePath     string                   `json:"sourcePath,omitempty"`