diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7ec71d6..df244d2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,9 @@
 
 ### New
 
+- Add reserved cleanup worker tasks with bounded deletion chunks, retry/quarantine handling,
+  configuration-preserving continue-as-new, authenticated current-channel RPCs and filter/version isolation.
+  This completes the cleanup runtime; aggregate release/consumer validation follows in phase 6.
 - Add the DTS payload-purge protocol and `setLargePayloadAutoPurge()` client lifecycle,
   including setting-first enable/disable and reserved-instance identity verification.
   A compatible .NET worker or the later cleanup-worker phase is required; do not publish partial phases.
diff --git a/README.md b/README.md
index 3367f29..9cd5cd7 100644
--- a/README.md
+++ b/README.md
@@ -33,8 +33,8 @@ before conversion or execution. Empty payloads remain inline, and disabling the
 the existing transport behavior.
 
 Azure-managed builders expose `.useExternalizedPayloads({ store })` and require core `>=0.4.1`
-for the public payload types. This phase works with a custom/in-memory store and does not import
-an Azure Blob provider or register cleanup tasks. Complete the six-phase feature stack before release.
+for the public payload types. A custom/in-memory store or the optional Azure Blob provider can
+implement storage independently of the transport. Complete the six-phase feature stack before release.
 
 ### Payload cleanup client protocol
 
@@ -44,8 +44,11 @@ the reserved Running orchestration, then signals its batch size. Disabling only
 backend setting. These steps are not transactional; earlier successful steps are not rolled back.
 
 This client operation does not require blob credentials. It requires a backend implementing
-the DTS `LargePayloadPurge` service and a compatible worker (the existing .NET implementation
-or the later phase 5 JavaScript runner). This phase does not register worker cleanup tasks.
+the DTS `LargePayloadPurge` service and a compatible worker. Payload-enabled JavaScript workers
+now register the reserved cleanup orchestrator and activities automatically, preserve their
+wildcard filters and bypass business version policy without changing customer filters.
+Deletion verifies ownership and ETags, batches 50 tokens per activity and limits concurrency;
+the runner backs off empty/retry-only batches and preserves configuration through continue-as-new.
 Do not publish the feature until all six phases are complete.
 
 Single-instance purges now identify the instance kind even on management-only clients without
diff --git a/packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts b/packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts
index d797512..a9b191c 100644
--- a/packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts
+++ b/packages/durabletask-js-azuremanaged/test/unit/payload-builders.spec.ts
@@ -27,7 +27,7 @@ describe("Azure-managed payload configuration", () => {
     expect(client["_payloads"]).toBeDefined();
     await client.stop();
   });
-  it("preserves worker payload settings and business registrations", () => {
+  it("preserves worker payload settings, business registrations, and internal wildcard filters", () => {
     const worker = new DurableTaskAzureManagedWorkerBuilder()
       .useExternalizedPayloads({ store: new Store() })
       .addNamedOrchestrator("business", () => undefined)
@@ -41,6 +41,6 @@ describe("Azure-managed payload configuration", () => {
         .getWorkitemfilters()!
         .getOrchestrationsList()
         .map((f) => f.getName()),
-    ).toEqual(["business"]);
+    ).toEqual(["business", "BlobPurgeJobOrchestrator"]);
   });
 });
diff --git a/packages/durabletask-js/src/payloads/purge-orchestrator.ts b/packages/durabletask-js/src/payloads/purge-orchestrator.ts
new file mode 100644
index 0000000..0484c62
--- /dev/null
+++ b/packages/durabletask-js/src/payloads/purge-orchestrator.ts
@@ -0,0 +1,123 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import { OrchestrationContext } from "../task/context/orchestration-context";
+import { Task } from "../task/task";
+import { whenAll, whenAny } from "../task";
+import { RetryPolicy } from "../task/retry/retry-policy";
+import { TaskFailureDetails } from "../task/failure-details";
+import { TaskFailedError } from "../task/exception/task-failed-error";
+import { Logger, ConsoleLogger } from "../types/logger.type";
+
+import {
+  BlobPurgeConstants,
+  BlobPurgeJobRunRequest,
+  BlobPurgeOutcome,
+  LargePayloadTombstone,
+  LargePayloadPurgeResult,
+  LargePayloadPurgeDisposition,
+  validatePurgeBatchSize,
+} from "./purge";
+
+export function isUnsupportedPurgeFailure(failure: TaskFailureDetails): boolean {
+  for (let current: TaskFailureDetails | undefined = failure; current; current = current.innerFailure) {
+    if (current.errorType === "System.NotImplementedException" || current.errorType === "NotImplementedException")
+      return true;
+  }
+  return false;
+}
+
+const retry = new RetryPolicy({
+  maxNumberOfAttempts: 3,
+  firstRetryIntervalInMilliseconds: 15000,
+  backoffCoefficient: 2,
+  maxRetryIntervalInMilliseconds: 60000,
+  handleFailure: (failure) => !isUnsupportedPurgeFailure(failure),
+});
+const taskOptions = { retry, version: "" };
+
+export async function* blobPurgeJobOrchestrator(
+  context: OrchestrationContext,
+  input: BlobPurgeJobRunRequest,
+  logger: Logger = new ConsoleLogger(),
+): AsyncGenerator<Task<any>, void, any> {
+  validatePurgeBatchSize(input.PurgeBatchSize);
+  const replayLogger = context.createReplaySafeLogger(logger);
+  let batchSize = input.PurgeBatchSize;
+  let purgedCount = input.PurgedCount ?? 0;
+  let cycles = 0;
+  let configuration = context.waitForExternalEvent(BlobPurgeConstants.configurationEvent);
+  for (;;) {
+    // CAN cannot preserve a configuration already delivered to this waiter.
+    while (configuration.isComplete) {
+      batchSize = configuration.getResult();
+      validatePurgeBatchSize(batchSize);
+      configuration = context.waitForExternalEvent(BlobPurgeConstants.configurationEvent);
+    }
+    if (cycles >= 5) {
+      context.continueAsNew({ PurgeBatchSize: batchSize, PurgedCount: purgedCount }, true);
+      return;
+    }
+    context.setCustomStatus({ Status: "Running", BatchSize: batchSize, PurgedCount: purgedCount });
+    cycles++;
+    let wait = false;
+    try {
+      const tombstones: LargePayloadTombstone[] = yield context.callActivity(
+        BlobPurgeConstants.fetch,
+        batchSize,
+        taskOptions,
+      );
+      if (!tombstones?.length) {
+        wait = true;
+      } else {
+        const results: LargePayloadPurgeResult[] = [];
+        for (let start = 0; start < tombstones.length; start += 50 * 4) {
+          const chunks: LargePayloadTombstone[][] = [];
+          const tasks: Task<BlobPurgeOutcome[]>[] = [];
+          for (let index = start; index < Math.min(start + 50 * 4, tombstones.length); index += 50) {
+            const chunk = tombstones.slice(index, index + 50);
+            chunks.push(chunk);
+            tasks.push(
+              context.callActivity(
+                BlobPurgeConstants.delete,
+                chunk.map((t) => t.PayloadToken),
+                taskOptions,
+              ),
+            );
+          }
+          const outcomes: BlobPurgeOutcome[][] = yield whenAll(tasks);
+          for (let index = 0; index < chunks.length; index++) {
+            const chunk = chunks[index];
+            if (!outcomes[index] || outcomes[index].length !== chunk.length) {
+              throw new Error("The blob delete activity must return exactly one outcome per token.");
+            }
+            results.push(
+              ...chunk.map((tombstone, position) => ({
+                TombstoneToken: tombstone.TombstoneToken,
+                Disposition: outcomes[index][position].Disposition,
+              })),
+            );
+          }
+        }
+        yield context.callActivity(BlobPurgeConstants.report, results, taskOptions);
+        purgedCount += results.filter((r) => r.Disposition === LargePayloadPurgeDisposition.Deleted).length;
+        wait = results.every((r) => r.Disposition === LargePayloadPurgeDisposition.Retry);
+      }
+    } catch (error) {
+      if (error instanceof TaskFailedError && isUnsupportedPurgeFailure(error.details)) {
+        replayLogger.warn("Large payload auto-purge backend unsupported; waiting for explicit configuration.");
+        context.setCustomStatus({ Status: "BackendUnsupported", BatchSize: batchSize, PurgedCount: purgedCount });
+        yield configuration;
+      } else {
+        replayLogger.error("Large payload auto-purge cycle failed; backing off before another attempt.");
+        wait = true;
+      }
+    }
+    if (wait && !configuration.isComplete) {
+      context.setCustomStatus({ Status: "Waiting", BatchSize: batchSize, PurgedCount: purgedCount });
+      const timer = context.createTimer(new Date(context.currentUtcDateTime.getTime() + 60000));
+      yield whenAny([timer, configuration]);
+      if (!timer.isComplete) timer.cancel();
+    }
+  }
+}
diff --git a/packages/durabletask-js/src/payloads/purge-worker.ts b/packages/durabletask-js/src/payloads/purge-worker.ts
new file mode 100644
index 0000000..33db347
--- /dev/null
+++ b/packages/durabletask-js/src/payloads/purge-worker.ts
@@ -0,0 +1,191 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import * as grpc from "@grpc/grpc-js";
+import * as lp from "../proto/large_payload_purge_pb";
+import { LargePayloadPurgeService } from "../proto/large_payload_purge_grpc_pb";
+import { Registry } from "../worker/registry";
+import { WorkItemFilters } from "../worker/work-item-filters";
+import { Logger } from "../types/logger.type";
+import { MetadataGenerator } from "../utils/grpc-helper.util";
+import { PayloadStore, PayloadDeleteOutcome, PayloadStorageException } from "./payload-store";
+import { runBounded } from "./run-bounded";
+import { callPayloadPurge } from "./purge-transport";
+import {
+  BlobPurgeConstants,
+  BlobPurgeOutcome,
+  LargePayloadPurgeDisposition,
+  LargePayloadPurgeResult,
+  validatePurgeBatchSize,
+} from "./purge";
+import { blobPurgeJobOrchestrator } from "./purge-orchestrator";
+
+const wireDispositions: Record<LargePayloadPurgeDisposition, lp.LargePayloadPurgeDisposition> = {
+  [LargePayloadPurgeDisposition.Unspecified]:
+    lp.LargePayloadPurgeDisposition.LARGE_PAYLOAD_PURGE_DISPOSITION_UNSPECIFIED,
+  [LargePayloadPurgeDisposition.Deleted]: lp.LargePayloadPurgeDisposition.LARGE_PAYLOAD_PURGE_DISPOSITION_DELETED,
+  [LargePayloadPurgeDisposition.Retry]: lp.LargePayloadPurgeDisposition.LARGE_PAYLOAD_PURGE_DISPOSITION_RETRY,
+  [LargePayloadPurgeDisposition.Quarantined]:
+    lp.LargePayloadPurgeDisposition.LARGE_PAYLOAD_PURGE_DISPOSITION_QUARANTINED,
+};
+
+export function registerPayloadPurge(
+  registry: Registry,
+  store: PayloadStore,
+  getClient: () => grpc.Client,
+  metadataGenerator: MetadataGenerator | undefined,
+  getSignal: () => AbortSignal | undefined,
+  logger: Logger,
+): void {
+  registry.addNamedOrchestrator(BlobPurgeConstants.orchestrator, (ctx, input) =>
+    blobPurgeJobOrchestrator(ctx, input, logger),
+  );
+  registry.addNamedActivity(BlobPurgeConstants.fetch, async (_ctx, limit: number) => {
+    validatePurgeBatchSize(limit);
+    try {
+      const response = await callPayloadPurge(
+        getClient(),
+        LargePayloadPurgeService.getLargePayloadTombstones,
+        new lp.GetLargePayloadTombstonesRequest().setLimit(limit),
+        metadataGenerator,
+        getSignal(),
+        new Date(Date.now() + 60000),
+      );
+      return response.getTombstonesList().map((t) => ({
+        TombstoneToken: t.getTombstoneToken(),
+        PayloadToken: t.getPayloadToken(),
+      }));
+    } catch (error) {
+      if (grpcStatus(error) === grpc.status.FAILED_PRECONDITION) {
+        logger.info("Large payload purge fetch precondition not met; idling.");
+        return [];
+      }
+      if (grpcStatus(error) === grpc.status.UNIMPLEMENTED) throw unsupportedBackend(error);
+      throw error;
+    }
+  });
+  registry.addNamedActivity(BlobPurgeConstants.delete, async (_ctx, tokens: string[]) =>
+    deletePayloadChunk(store, tokens, logger, getSignal()),
+  );
+  registry.addNamedActivity(BlobPurgeConstants.report, async (_ctx, results: LargePayloadPurgeResult[]) => {
+    if (!results?.length) return null;
+    const request = new lp.ReportLargePayloadPurgeResultsRequest().setResultsList(
+      results.map((r) => {
+        const disposition = wireDispositions[r.Disposition];
+        if (disposition === undefined) throw new TypeError("Unknown large-payload purge disposition.");
+        return new lp.LargePayloadPurgeResult().setTombstoneToken(r.TombstoneToken).setDisposition(disposition);
+      }),
+    );
+    try {
+      await callPayloadPurge(
+        getClient(),
+        LargePayloadPurgeService.reportLargePayloadPurgeResults,
+        request,
+        metadataGenerator,
+        getSignal(),
+        new Date(Date.now() + 60000),
+      );
+    } catch (error) {
+      if (grpcStatus(error) === grpc.status.UNIMPLEMENTED) throw unsupportedBackend(error);
+      throw error;
+    }
+    return null;
+  });
+}
+
+export async function deletePayloadChunk(
+  store: PayloadStore,
+  tokens: string[],
+  logger: Logger,
+  signal?: AbortSignal,
+): Promise<BlobPurgeOutcome[]> {
+  const outcomes: BlobPurgeOutcome[] = new Array(tokens.length);
+  await runBounded(
+    tokens.map((token, index) => async () => {
+      outcomes[index] = { Disposition: await deletePayload(store, token, logger, signal) };
+    }),
+  );
+  return outcomes;
+}
+
+async function deletePayload(
+  store: PayloadStore,
+  token: string,
+  logger: Logger,
+  signal?: AbortSignal,
+): Promise<LargePayloadPurgeDisposition> {
+  const classify = (disposition: LargePayloadPurgeDisposition, reason: string) => {
+    logger.warn(`Large payload purge: ${reason}; disposition=${disposition}.`);
+    return disposition;
+  };
+  if (token.startsWith("blob:v1:")) return classify(LargePayloadPurgeDisposition.Quarantined, "LegacyV1Token");
+  if (!token.startsWith("blob:v2:")) return classify(LargePayloadPurgeDisposition.Retry, "UnsupportedTokenVersion");
+  const timeout = new AbortController();
+  const timer = setTimeout(() => timeout.abort(), 60000);
+  timer.unref();
+  try {
+    const deleteSignal = signal ? AbortSignal.any([signal, timeout.signal]) : timeout.signal;
+    const outcome = await store.delete(token, deleteSignal);
+    switch (outcome) {
+      case PayloadDeleteOutcome.Deleted:
+      case PayloadDeleteOutcome.AlreadyAbsent:
+        return LargePayloadPurgeDisposition.Deleted;
+      case PayloadDeleteOutcome.NotStoreOwned:
+        logger.info("Large payload purge: NotStoreOwned; blob left untouched.");
+        return LargePayloadPurgeDisposition.Deleted;
+      default:
+        return classify(LargePayloadPurgeDisposition.Retry, "UnexpectedDeleteOutcome");
+    }
+  } catch (error) {
+    if (error instanceof TypeError) return classify(LargePayloadPurgeDisposition.Quarantined, "MalformedToken");
+    if (error instanceof PayloadStorageException)
+      return classify(LargePayloadPurgeDisposition.Retry, "StorageAccountUnreachable");
+    const status = error !== null && typeof error === "object" && "statusCode" in error ? error.statusCode : undefined;
+    if (status === 400) return classify(LargePayloadPurgeDisposition.Quarantined, "InvalidStorageRequest");
+    if (status === 401 || status === 403)
+      return classify(LargePayloadPurgeDisposition.Retry, "StorageAuthorizationFailed");
+    return classify(LargePayloadPurgeDisposition.Retry, "UnconfirmedDelete");
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
+export function isPayloadPurgeTask(name?: string): boolean {
+  return (
+    name === BlobPurgeConstants.orchestrator ||
+    name === BlobPurgeConstants.fetch ||
+    name === BlobPurgeConstants.delete ||
+    name === BlobPurgeConstants.report
+  );
+}
+
+export function withPayloadPurgeFilters(filters: WorkItemFilters): WorkItemFilters {
+  if (!(filters.orchestrations?.length || filters.activities?.length || filters.entities?.length)) return filters;
+  return {
+    ...filters,
+    orchestrations: [
+      ...(filters.orchestrations ?? []).filter((f) => f.name !== BlobPurgeConstants.orchestrator),
+      { name: BlobPurgeConstants.orchestrator, versions: [] },
+    ],
+    activities: [
+      ...(filters.activities ?? []).filter((f) => !isPayloadPurgeTask(f.name)),
+      ...[BlobPurgeConstants.fetch, BlobPurgeConstants.delete, BlobPurgeConstants.report].map((name) => ({
+        name,
+        versions: [],
+      })),
+    ],
+  };
+}
+
+function grpcStatus(error: unknown): unknown {
+  return error !== null && typeof error === "object" && "code" in error ? error.code : undefined;
+}
+
+function unsupportedBackend(cause: unknown): Error {
+  const error = new Error(
+    "The backend does not implement large-payload auto-purge. Upgrade the backend or DTS emulator, then enable auto-purge again.",
+    { cause },
+  );
+  error.name = "System.NotImplementedException";
+  return error;
+}
diff --git a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts
index 46284b6..c5a124d 100644
--- a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts
+++ b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts
@@ -29,6 +29,7 @@ import { compareVersions } from "../utils/versioning.util";
 import { PayloadStorageOptions } from "../payloads/payload-store";
 import { PayloadProcessor } from "../payloads/payload-processor";
 import { readPayloadHistory, PayloadHydrationError } from "../payloads/history-stream";
+import { registerPayloadPurge, withPayloadPurgeFilters, isPayloadPurgeTask } from "../payloads/purge-worker";
 import * as WorkerLogs from "./logs";
 import { ConcurrencyOptions, ResolvedConcurrencyOptions, resolveConcurrencyOptions } from "./concurrency-options";
 import {
@@ -266,6 +267,19 @@ export class TaskHubGrpcWorker {
     this._abortController = null;
     this._workerLoopPromise = null;
     this._deferredStubCloseTimers = new Map();
+    if (typeof hostAddressOrOptions === "object" && hostAddressOrOptions?.payloads) {
+      registerPayloadPurge(
+        this._registry,
+        hostAddressOrOptions.payloads.store,
+        () => {
+          if (!this._stub) throw new Error("The payload purge transport is not connected.");
+          return this._stub;
+        },
+        this._metadataGenerator,
+        () => this._abortController?.signal,
+        this._logger,
+      );
+    }
   }
 
   /** The configured maximum backend timer interval in milliseconds. */
@@ -880,7 +894,7 @@ export class TaskHubGrpcWorker {
         this._workItemFilters === "auto"
           ? generateWorkItemFiltersFromRegistry(this._registry, this._versioning)
           : this._workItemFilters;
-      request.setWorkitemfilters(toGrpcWorkItemFilters(filters));
+      request.setWorkitemfilters(toGrpcWorkItemFilters(this._payloads ? withPayloadPurgeFilters(filters) : filters));
     }
 
     return request;
@@ -889,13 +903,17 @@ export class TaskHubGrpcWorker {
   /**
    * Result of version compatibility check.
    */
-  private _checkVersionCompatibility(orchestrationVersion?: string): {
+  private _checkVersionCompatibility(
+    orchestrationVersion?: string,
+    taskName?: string,
+  ): {
     compatible: boolean;
     shouldFail: boolean;
     orchestrationVersion?: string;
     errorType?: string;
     errorMessage?: string;
   } {
+    if (this._payloads && isPayloadPurgeTask(taskName)) return { compatible: true, shouldFail: false };
     // If no versioning options configured or match strategy is None, always compatible
     if (
       !this._versioning ||
@@ -1139,7 +1157,11 @@ export class TaskHubGrpcWorker {
     }
 
     // Check version compatibility if versioning is enabled
-    const versionCheckResult = this._checkVersionCompatibility(this._getOrchestrationVersion(req));
+    const taskName = [...req.getPasteventsList(), ...req.getNeweventsList()]
+      .find((event) => event.hasExecutionstarted())
+      ?.getExecutionstarted()
+      ?.getName();
+    const versionCheckResult = this._checkVersionCompatibility(this._getOrchestrationVersion(req), taskName);
     if (!versionCheckResult.compatible) {
       if (versionCheckResult.shouldFail) {
         // Fail the orchestration with version mismatch error
@@ -1353,7 +1375,7 @@ export class TaskHubGrpcWorker {
       throw new Error("Activity request does not contain an orchestration instance id");
     }
 
-    const versionCheck = this._checkVersionCompatibility(req.getVersion()?.getValue());
+    const versionCheck = this._checkVersionCompatibility(req.getVersion()?.getValue(), req.getName());
     if (!versionCheck.compatible && !versionCheck.shouldFail) {
       WorkerLogs.versionMismatchAbandon(this._logger, instanceId, versionCheck.errorType!, versionCheck.errorMessage!);
       try {
diff --git a/packages/durabletask-js/test/payload-purge-worker-transport.spec.ts b/packages/durabletask-js/test/payload-purge-worker-transport.spec.ts
new file mode 100644
index 0000000..8dff54c
--- /dev/null
+++ b/packages/durabletask-js/test/payload-purge-worker-transport.spec.ts
@@ -0,0 +1,94 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import * as grpc from "@grpc/grpc-js";
+import { TaskHubGrpcWorker, NoOpLogger, PayloadStore, ActivityContext } from "../src";
+import * as stubs from "../src/proto/orchestrator_service_grpc_pb";
+import * as lp from "../src/proto/large_payload_purge_pb";
+import { LargePayloadPurgeService, ILargePayloadPurgeServer } from "../src/proto/large_payload_purge_grpc_pb";
+import { BlobPurgeConstants } from "../src/payloads/purge";
+
+class Store extends PayloadStore {
+  async upload(): Promise<string> {
+    return "blob:v2:https://account.invalid/c/blob";
+  }
+  async download(): Promise<string> {
+    return "null";
+  }
+  isKnownPayloadToken(value: string): boolean {
+    return value.startsWith("blob:");
+  }
+}
+
+it("uses the current authenticated sidecar transport for reserved fetch/report activities", async () => {
+  const calls: string[] = [];
+  let rpcError: grpc.status | undefined;
+  const server = new grpc.Server();
+  const service = {
+    getLargePayloadTombstones: (call, callback) => {
+      calls.push("fetch");
+      expect(call.metadata.get("taskhub")).toEqual(["payload-test"]);
+      expect(Number(call.getDeadline()) - Date.now()).toBeLessThanOrEqual(61000);
+      if (rpcError !== undefined) callback({ code: rpcError, message: "backend" });
+      else
+        callback(
+          null,
+          new lp.GetLargePayloadTombstonesResponse().setTombstonesList([
+            new lp.LargePayloadTombstone()
+              .setTombstoneToken("opaque:\u00e9:do-not-parse")
+              .setPayloadToken("blob:v2:https://a.invalid/c/b"),
+          ]),
+        );
+    },
+    reportLargePayloadPurgeResults: (call, callback) => {
+      calls.push("report");
+      expect(call.metadata.get("taskhub")).toEqual(["payload-test"]);
+      expect(Number(call.getDeadline()) - Date.now()).toBeLessThanOrEqual(61000);
+      expect(call.request.getResultsList()[0].getTombstoneToken()).toBe("opaque:\u00e9:do-not-parse");
+      callback(null, new lp.ReportLargePayloadPurgeResultsResponse());
+    },
+  } satisfies Pick<ILargePayloadPurgeServer, "getLargePayloadTombstones" | "reportLargePayloadPurgeResults">;
+  server.addService(LargePayloadPurgeService, service);
+  const port = await new Promise<number>((resolve, reject) =>
+    server.bindAsync("127.0.0.1:0", grpc.ServerCredentials.createInsecure(), (error, port) =>
+      error ? reject(error) : resolve(port),
+    ),
+  );
+  const first = new stubs.TaskHubSidecarServiceClient(`127.0.0.1:${port}`, grpc.credentials.createInsecure());
+  const replacement = new stubs.TaskHubSidecarServiceClient(`127.0.0.1:${port}`, grpc.credentials.createInsecure());
+  const firstCall = jest.spyOn(first, "makeUnaryRequest");
+  const replacementCall = jest.spyOn(replacement, "makeUnaryRequest");
+  try {
+    const worker = new TaskHubGrpcWorker({
+      payloads: { store: new Store() },
+      logger: new NoOpLogger(),
+      metadataGenerator: async () => {
+        const metadata = new grpc.Metadata();
+        metadata.set("taskhub", "payload-test");
+        return metadata;
+      },
+    });
+    worker["_stub"] = first;
+    const fetch = worker["_registry"].getActivity(BlobPurgeConstants.fetch)!;
+    const report = worker["_registry"].getActivity(BlobPurgeConstants.report)!;
+    const tombstones = await fetch(new ActivityContext("job", 0), 10);
+    expect(tombstones).toEqual([
+      { TombstoneToken: "opaque:\u00e9:do-not-parse", PayloadToken: "blob:v2:https://a.invalid/c/b" },
+    ]);
+    worker["_stub"] = replacement;
+    await report(new ActivityContext("job", 1), [{ TombstoneToken: tombstones[0].TombstoneToken, Disposition: 1 }]);
+    expect(firstCall).toHaveBeenCalledTimes(1);
+    expect(replacementCall).toHaveBeenCalledTimes(1);
+    expect(calls).toEqual(["fetch", "report"]);
+    rpcError = grpc.status.FAILED_PRECONDITION;
+    expect(await fetch(new ActivityContext("job", 2), 10)).toEqual([]);
+    rpcError = grpc.status.UNIMPLEMENTED;
+    await expect(fetch(new ActivityContext("job", 3), 10)).rejects.toMatchObject({
+      name: "System.NotImplementedException",
+    });
+  } finally {
+    first.close();
+    replacement.close();
+    server.forceShutdown();
+  }
+});
diff --git a/packages/durabletask-js/test/payload-purge-worker.spec.ts b/packages/durabletask-js/test/payload-purge-worker.spec.ts
new file mode 100644
index 0000000..2086524
--- /dev/null
+++ b/packages/durabletask-js/test/payload-purge-worker.spec.ts
@@ -0,0 +1,230 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import {
+  TaskHubGrpcWorker,
+  NoOpLogger,
+  PayloadStore,
+  PayloadDeleteOutcome,
+  VersionMatchStrategy,
+  WorkItemFilters,
+  TaskFailedError,
+} from "../src";
+import * as pb from "../src/proto/orchestrator_service_pb";
+import { BlobPurgeConstants, BlobPurgeOutcome, LargePayloadPurgeDisposition } from "../src/payloads/purge";
+import { blobPurgeJobOrchestrator } from "../src/payloads/purge-orchestrator";
+import { deletePayloadChunk } from "../src/payloads/purge-worker";
+import { RuntimeOrchestrationContext } from "../src/worker/runtime-orchestration-context";
+import { RetryPolicy } from "../src/task/retry/retry-policy";
+
+class Store extends PayloadStore {
+  async upload(): Promise<string> {
+    return "blob:v2:https://account.invalid/payloads/blob";
+  }
+  async download(): Promise<string> {
+    return "null";
+  }
+  isKnownPayloadToken(value: string): boolean {
+    return value.startsWith("blob:");
+  }
+  async delete(_token: string): Promise<PayloadDeleteOutcome> {
+    return PayloadDeleteOutcome.Deleted;
+  }
+}
+const logger = new NoOpLogger();
+
+describe("payload purge worker registrations", () => {
+  it.each<WorkItemFilters>([{}, { orchestrations: [], activities: [], entities: [] }])(
+    "leaves all-empty filters unfiltered",
+    (filters) => {
+      const worker = new TaskHubGrpcWorker({ payloads: { store: new Store() }, workItemFilters: filters });
+      expect(worker["_buildGetWorkItemsRequest"]().getWorkitemfilters()!.toObject()).toEqual({
+        orchestrationsList: [],
+        activitiesList: [],
+        entitiesList: [],
+      });
+    },
+  );
+  it("augments entity-only filters, bypasses business versions, and does not mutate caller lists", () => {
+    const filters: WorkItemFilters = { entities: [{ name: "counter" }] };
+    const worker = new TaskHubGrpcWorker({
+      payloads: { store: new Store() },
+      workItemFilters: filters,
+      versioning: { version: "business", matchStrategy: VersionMatchStrategy.Strict },
+    });
+    const wire = worker["_buildGetWorkItemsRequest"]().getWorkitemfilters()!;
+    expect(wire.getOrchestrationsList().map((f) => [f.getName(), f.getVersionsList()])).toEqual([
+      [BlobPurgeConstants.orchestrator, []],
+    ]);
+    expect(wire.getActivitiesList()).toHaveLength(3);
+    expect(filters).toEqual({ entities: [{ name: "counter" }] });
+    expect(worker["_checkVersionCompatibility"]("", BlobPurgeConstants.orchestrator).compatible).toBe(true);
+    expect(worker["_checkVersionCompatibility"]("", "business").compatible).toBe(false);
+  });
+});
+
+describe("purge delete classification", () => {
+  it("bounds deletes at eight and preserves positional dispositions without logging tokens", async () => {
+    const store = new Store();
+    let active = 0;
+    let max = 0;
+    store.delete = async (token) => {
+      max = Math.max(max, ++active);
+      await new Promise((resolve) => setImmediate(resolve));
+      active--;
+      if (token.endsWith("error")) throw Object.assign(new Error(token), { statusCode: 403 });
+      return PayloadDeleteOutcome.Deleted;
+    };
+    const tokens = Array.from({ length: 50 }, (_, i) => `blob:v2:https://a.invalid/c/${i === 3 ? "error" : i}`);
+    const warn = jest.fn();
+    const outcomes = await deletePayloadChunk(store, tokens, {
+      ...logger,
+      warn,
+      error: jest.fn(),
+      info: jest.fn(),
+      debug: jest.fn(),
+    });
+    expect(max).toBe(8);
+    expect(outcomes.map((o) => o.Disposition)).toEqual(tokens.map((_, i) => (i === 3 ? 2 : 1)));
+    expect(JSON.stringify(warn.mock.calls)).not.toContain("a.invalid");
+  });
+  it.each([
+    ["blob:v1:c:b", 3],
+    ["blob:v3:future", 2],
+  ])("does not delete %s", async (token, disposition) => {
+    const store = new Store();
+    store.delete = jest.fn();
+    expect(await deletePayloadChunk(store, [token], logger)).toEqual([{ Disposition: disposition }]);
+    expect(store.delete).not.toHaveBeenCalled();
+  });
+  it.each([
+    [new TypeError("bad token"), 3],
+    [Object.assign(new Error("bad"), { statusCode: 400 }), 3],
+    [Object.assign(new Error("bad"), { statusCode: 401 }), 2],
+    [Object.assign(new Error("bad"), { statusCode: 412 }), 2],
+    [new Error("network"), 2],
+  ])("classifies storage errors without losing tombstones (%#)", async (error, disposition) => {
+    const store = new Store();
+    store.delete = async () => {
+      throw error;
+    };
+    expect(await deletePayloadChunk(store, ["blob:v2:https://a.invalid/c/b"], logger)).toEqual([
+      { Disposition: disposition },
+    ]);
+  });
+  it.each([PayloadDeleteOutcome.Deleted, PayloadDeleteOutcome.AlreadyAbsent, PayloadDeleteOutcome.NotStoreOwned])(
+    "maps confirmed delete outcome %s to terminal success",
+    async (outcome) => {
+      const store = new Store();
+      store.delete = async () => outcome;
+      expect(await deletePayloadChunk(store, ["blob:v2:https://a.invalid/c/b"], logger)).toEqual([{ Disposition: 1 }]);
+    },
+  );
+});
+
+describe("replay-safe purge runner", () => {
+  it("chunks 50 tokens, schedules four chunks per wave, and echoes opaque tombstones unchanged", async () => {
+    const context = new RuntimeOrchestrationContext("job");
+    const run = blobPurgeJobOrchestrator(context, { PurgeBatchSize: 500, PurgedCount: 2 });
+    await run.next();
+    const tombstones = Array.from({ length: 251 }, (_, i) => ({
+      TombstoneToken: `opaque:${i}`,
+      PayloadToken: `blob:v2:https://a/c/${i}`,
+    }));
+    await run.next(tombstones);
+    let actions = Object.values(context._pendingActions);
+    expect(actions.filter((a) => a.getScheduletask()?.getName() === BlobPurgeConstants.delete)).toHaveLength(4);
+    await run.next(Array.from({ length: 4 }, () => Array.from({ length: 50 }, () => ({ Disposition: 1 }))));
+    actions = Object.values(context._pendingActions);
+    expect(actions.filter((a) => a.getScheduletask()?.getName() === BlobPurgeConstants.delete)).toHaveLength(6);
+    await run.next([Array.from({ length: 50 }, () => ({ Disposition: 1 })), [{ Disposition: 1 }]]);
+    const report = Object.values(context._pendingActions).find(
+      (a) => a.getScheduletask()?.getName() === BlobPurgeConstants.report,
+    )!;
+    expect(JSON.parse(report.getScheduletask()!.getInput()!.getValue())).toEqual(
+      tombstones.map((t) => ({ TombstoneToken: t.TombstoneToken, Disposition: 1 })),
+    );
+    await run.return();
+  });
+
+  it("drains delivered and buffered configuration before continue-as-new on cycle five", async () => {
+    const context = new RuntimeOrchestrationContext("job");
+    const run = blobPurgeJobOrchestrator(context, { PurgeBatchSize: 500, PurgedCount: 3 });
+    await run.next();
+    for (let cycle = 0; cycle < 5; cycle++) {
+      await run.next([{ TombstoneToken: "opaque", PayloadToken: "blob:v2:https://a/c/b" }]);
+      await run.next([[{ Disposition: 1 }]]);
+      if (cycle === 4) {
+        context._pendingEvents.setbatchsize[0].complete(60);
+        context._receivedEvents.setbatchsize = [70, 80];
+      }
+      await run.next(null);
+    }
+    expect(context._newInput).toEqual({ PurgeBatchSize: 80, PurgedCount: 8 });
+    expect(context._saveEvents).toBe(true);
+  });
+
+  it("waits for explicit configuration on unsupported backends without scheduling a retry timer", async () => {
+    const context = new RuntimeOrchestrationContext("job");
+    const call = jest.spyOn(context, "callActivity");
+    const run = blobPurgeJobOrchestrator(context, { PurgeBatchSize: 500, PurgedCount: 0 });
+    await run.next();
+    const policy = call.mock.calls[0][2]!.retry;
+    expect(policy).toBeInstanceOf(RetryPolicy);
+    expect(
+      (policy as RetryPolicy).shouldRetry({ errorType: "System.NotImplementedException", message: "unsupported" }),
+    ).toBe(false);
+    const failure = new TaskFailedError(
+      "unsupported",
+      new pb.TaskFailureDetails().setErrortype("System.NotImplementedException"),
+    );
+    const waiting = await run.throw(failure);
+    expect(waiting.value).toBe(context._pendingEvents.setbatchsize[0]);
+    expect(Object.values(context._pendingActions).some((a) => a.hasCreatetimer())).toBe(false);
+    expect(JSON.parse(context.getCustomStatus()!).Status).toBe("BackendUnsupported");
+    context._pendingEvents.setbatchsize[0].complete(100);
+    await run.next(100);
+    expect(call).toHaveBeenCalledTimes(2);
+    await run.return();
+  });
+
+  it("backs off all-Retry results and cancels its idle timer when configuration arrives", async () => {
+    const context = new RuntimeOrchestrationContext("job");
+    const timer = jest.spyOn(context, "createTimer");
+    const run = blobPurgeJobOrchestrator(context, { PurgeBatchSize: 10, PurgedCount: 0 });
+    await run.next();
+    await run.next([{ TombstoneToken: "opaque", PayloadToken: "blob:v2:https://a/c/b" }]);
+    await run.next([[{ Disposition: LargePayloadPurgeDisposition.Retry } satisfies BlobPurgeOutcome]]);
+    await run.next(null);
+    expect(timer).toHaveBeenCalledTimes(1);
+    expect((timer.mock.calls[0][0] as Date).getTime() - context.currentUtcDateTime.getTime()).toBe(60000);
+    context._pendingEvents.setbatchsize[0].complete(20);
+    await run.next(context._pendingEvents.setbatchsize[0]);
+    expect(timer.mock.results[0].value.isCanceled).toBe(true);
+    await run.return();
+  });
+
+  it("waits for configuration before rolling over an unsupported fifth cycle", async () => {
+    const context = new RuntimeOrchestrationContext("job");
+    const run = blobPurgeJobOrchestrator(context, { PurgeBatchSize: 500, PurgedCount: 0 }, logger);
+    await run.next();
+    for (let cycle = 0; cycle < 4; cycle++) {
+      await run.next([{ TombstoneToken: "opaque", PayloadToken: "blob:v2:https://a/c/b" }]);
+      await run.next([[{ Disposition: 1 }]]);
+      await run.next(null);
+    }
+    const failure = new TaskFailedError(
+      "unsupported",
+      new pb.TaskFailureDetails().setErrortype("System.NotImplementedException"),
+    );
+    const waiting = await run.throw(failure);
+    expect(waiting.done).toBe(false);
+    expect(context._newInput).toBeUndefined();
+    expect(Object.values(context._pendingActions).some((action) => action.hasCreatetimer())).toBe(false);
+    context._pendingEvents.setbatchsize[0].complete(60);
+    context._receivedEvents.setbatchsize = [70];
+    expect((await run.next(60)).done).toBe(true);
+    expect(context._newInput).toEqual({ PurgeBatchSize: 70, PurgedCount: 4 });
+    expect(context._saveEvents).toBe(true);
+  });
+});