Add worker-level orchestration and activity middleware

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### New
 
+- Add first-class worker-level orchestration and activity middleware with typed contexts, nonserialized host features, validation, and in-process host integration.
 - Add an optional `newVersion` parameter to `OrchestrationContext.continueAsNew()` for version migrations.
 - Implement entity support in the in-memory testing backend ([#341](https://github.com/microsoft/durabletask-js/pull/341))
 - Add the top-level `StartOrchestrationOptions.dedupeStatuses` option, `ValidDedupeStatuses`, and

README.md

@@ -18,11 +18,79 @@ worker.addActivity(myActivity);
 worker.addEntity(myEntity);
 
 const orchestrationResponseBytes = await worker.processOrchestratorRequest(orchestrationRequestBytes);
+const activityResponseBytes = await worker.processActivityRequest(activityRequestBytes);
 const entityResponseBytes = await worker.processEntityBatchRequest(entityBatchRequestBytes);
 ```
 
 `TaskHubGrpcClient` already exposes orchestration start/query/event/terminate/suspend/resume/purge APIs and entity signal/read/query/clean APIs through its existing `hostAddress` and `metadataGenerator` options. Host integrations that need task-hub routing metadata should provide it through `metadataGenerator`, keeping host-specific metadata policy outside the core client. Azure-managed scheduler connection strings remain in `@microsoft/durabletask-js-azuremanaged`.
 
+## Durable Task middleware
+
+Worker-level middleware runs around orchestration or activity execution. The first registered middleware is outermost, so middleware runs in registration order before user code and unwinds in reverse order afterward.
+
+```typescript
+import { ActivityMiddleware, OrchestrationMiddleware, TaskHubGrpcWorker } from "@microsoft/durabletask-js";
+
+const orchestrationLogging: OrchestrationMiddleware = async (context, next) => {
+  const logger = context.orchestrationContext.createReplaySafeLogger(appLogger);
+  logger.info(`Starting ${context.name} (${context.instanceId})`);
+
+  await next(context);
+
+  if (context.failure) {
+    logger.error(`Failed ${context.name}: ${context.failure.message}`);
+  } else if (context.result !== undefined) {
+    logger.info(`Completed ${context.name}`);
+  }
+};
+
+const activityCache: ActivityMiddleware = async (context, next) => {
+  const cached = await cache.get(context.name, context.input);
+  if (cached !== undefined) {
+    context.setResult(cached);
+    return;
+  }
+
+  await next(context);
+  await cache.set(context.name, context.input, context.result);
+};
+
+const worker = new TaskHubGrpcWorker();
+worker
+  .useOrchestrationMiddleware(orchestrationLogging)
+  .useActivityMiddleware(activityCache)
+  .addOrchestrator(myOrchestrator);
+worker.addActivity(myActivity);
+```
+
+Orchestration middleware participates in generator replay: code before `next(context)` re-executes on each replay pass, while `next` remains suspended with the orchestrator and code after it unwinds only when the orchestration reaches a terminal result or failure. Orchestration middleware must call `next(context)` exactly once and must not return before it completes. It cannot short-circuit, replace the orchestration result, or suppress a downstream failure by catching it. Activity middleware must either call `next(context)` once without returning before it completes, or explicitly short-circuit with `context.setResult(...)`; calling `next` after `setResult` is invalid, while calling `setResult` after `next` may replace a result or explicitly recover from a caught downstream failure. Missing, duplicate, and incomplete `next` calls are rejected. In normal `async` middleware, use `await next(context)` as shown above.
+
+Orchestration middleware follows the same determinism rules as orchestrator code. Do not await nondurable promises or use wall-clock time, random values, mutable process state, file/network I/O, or host state that can change across replay. Use durable context APIs and `createReplaySafeLogger()` instead. Put I/O in activities or activity middleware.
+
+### Host features
+
+`MiddlewareFeatures` is a typed, symbol-keyed map for per-invocation host objects. Feature values remain process-local and are never serialized into durable history.
+
+```typescript
+import { MiddlewareFeatures, createMiddlewareFeature } from "@microsoft/durabletask-js";
+
+interface HostInvocation {
+  invocationId: string;
+}
+
+const hostInvocation = createMiddlewareFeature<HostInvocation>("host invocation");
+const features = new MiddlewareFeatures().set(hostInvocation, currentInvocation);
+
+worker.useActivityMiddleware(async (context, next) => {
+  const invocation = context.features.get(hostInvocation);
+  await next(context);
+});
+
+await worker.processActivityRequest(activityRequestBytes, features);
+```
+
+Hosts can pass features to `processOrchestratorRequest()` and `processActivityRequest()`. Durable Task middleware is for durable-aware invocation data such as task name, instance ID, version, parent, tags, input, replay state, result, and failure. Host or framework middleware remains the right layer for concerns that apply to every host invocation, such as HTTP headers, authentication preprocessing, and nondurable bindings. Entity middleware is not supported in this version.
+
 ## npm packages
 
 The following npm packages are available for download.
@@ -176,10 +244,7 @@ Long-running orchestrations can restart with fresh history and optionally move t
 orchestration version:
 
 ```typescript
-const eternalOrchestrator: TOrchestrator = async function* (
-  ctx: OrchestrationContext,
-  iteration: number,
-): any {
+const eternalOrchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, iteration: number): any {
   yield ctx.callActivity(processIteration, iteration);
   ctx.continueAsNew(iteration + 1, true, "2.0.0");
 };

packages/durabletask-js-azuremanaged/CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ### New
 
+- Add orchestration and activity middleware registration to `DurableTaskAzureManagedWorkerBuilder`.
+
 ### Fixes
 
 ## v0.4.0 (2026-07-31)

packages/durabletask-js-azuremanaged/README.md

@@ -26,6 +26,15 @@ const client = createAzureManagedClient(
 const worker = createAzureManagedWorkerBuilder(
   "Endpoint=https://myservice.durabletask.io;Authentication=DefaultAzure;TaskHub=myTaskHub",
 )
+  .useOrchestrationMiddleware(async (context, next) => {
+    const logger = context.orchestrationContext.createReplaySafeLogger(appLogger);
+    logger.info(`Running ${context.name} (${context.instanceId})`);
+    await next(context);
+  })
+  .useActivityMiddleware(async (context, next) => {
+    await next(context);
+    appLogger.info(`Activity ${context.name} returned`, context.result);
+  })
   .addOrchestrator(myOrchestrator)
   .addActivity(myActivity)
   .build();

packages/durabletask-js-azuremanaged/src/worker-builder.ts

@@ -15,6 +15,8 @@ import {
   ConsoleLogger,
   VersioningOptions,
   WorkItemFilters,
+  ActivityMiddleware,
+  OrchestrationMiddleware,
 } from "@microsoft/durabletask-js";
 
 /**
@@ -31,6 +33,8 @@ export class DurableTaskAzureManagedWorkerBuilder {
   private _shutdownTimeoutMs?: number;
   private _versioning?: VersioningOptions;
   private _workItemFilters?: WorkItemFilters | "auto";
+  private _orchestrationMiddleware: OrchestrationMiddleware[] = [];
+  private _activityMiddleware: ActivityMiddleware[] = [];
 
   /**
    * Creates a new instance of DurableTaskAzureManagedWorkerBuilder.
@@ -187,6 +191,22 @@ export class DurableTaskAzureManagedWorkerBuilder {
     return this;
   }
 
+  /**
+   * Adds orchestration middleware to the worker.
+   */
+  useOrchestrationMiddleware(middleware: OrchestrationMiddleware): DurableTaskAzureManagedWorkerBuilder {
+    this._orchestrationMiddleware.push(middleware);
+    return this;
+  }
+
+  /**
+   * Adds activity middleware to the worker.
+   */
+  useActivityMiddleware(middleware: ActivityMiddleware): DurableTaskAzureManagedWorkerBuilder {
+    this._activityMiddleware.push(middleware);
+    return this;
+  }
+
   /**
    * Registers an entity factory with the worker.
    * The entity name is derived from the factory function name.
@@ -297,6 +317,14 @@ export class DurableTaskAzureManagedWorkerBuilder {
       workItemFilters: this._workItemFilters,
     });
 
+    for (const middleware of this._orchestrationMiddleware) {
+      worker.useOrchestrationMiddleware(middleware);
+    }
+
+    for (const middleware of this._activityMiddleware) {
+      worker.useActivityMiddleware(middleware);
+    }
+
     // Register all orchestrators
     for (const { name, fn } of this._orchestrators) {
       if (name) {

packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts

@@ -2,7 +2,13 @@
 // Licensed under the MIT License.
 
 import { DurableTaskAzureManagedWorkerBuilder, createAzureManagedWorkerBuilder } from "../../src/worker-builder";
-import { TaskEntity, ITaskEntity, TaskEntityOperation } from "@microsoft/durabletask-js";
+import {
+  ActivityMiddleware,
+  OrchestrationMiddleware,
+  TaskEntity,
+  ITaskEntity,
+  TaskEntityOperation,
+} from "@microsoft/durabletask-js";
 
 // Simple test entity for registration testing
 class CounterEntity extends TaskEntity<number> {
@@ -79,6 +85,22 @@ describe("DurableTaskAzureManagedWorkerBuilder", () => {
 
       expect(worker).toBeDefined();
     });
+
+    describe("middleware", () => {
+      it("supports orchestration and activity middleware registration through the builder", () => {
+        const orchestrationMiddleware: OrchestrationMiddleware = async (context, next) => next(context);
+        const activityMiddleware: ActivityMiddleware = async (context, next) => next(context);
+        const builder = new DurableTaskAzureManagedWorkerBuilder();
+
+        const result = builder
+          .endpoint(ENDPOINT, TASKHUB, null)
+          .useOrchestrationMiddleware(orchestrationMiddleware)
+          .useActivityMiddleware(activityMiddleware);
+
+        expect(result).toBe(builder);
+        expect(result.build()).toBeDefined();
+      });
+    });
   });
 
   describe("createAzureManagedWorkerBuilder", () => {

packages/durabletask-js/src/index.ts

@@ -5,6 +5,17 @@
 export { TaskHubGrpcClient, TaskHubGrpcClientOptions, MetadataGenerator } from "./client/client";
 export { TaskHubGrpcWorker, TaskHubGrpcWorkerOptions } from "./worker/task-hub-grpc-worker";
 export { VersioningOptions, VersionMatchStrategy, VersionFailureStrategy } from "./worker/versioning-options";
+export {
+  ActivityMiddleware,
+  ActivityMiddlewareContext,
+  ActivityMiddlewareNext,
+  MiddlewareFeature,
+  MiddlewareFeatures,
+  OrchestrationMiddleware,
+  OrchestrationMiddlewareContext,
+  OrchestrationMiddlewareNext,
+  createMiddlewareFeature,
+} from "./worker/middleware";
 export {
   WorkItemFilters,
   OrchestrationWorkItemFilter,

packages/durabletask-js/src/testing/in-memory-backend.ts

@@ -7,6 +7,7 @@ import { OrchestrationStatus as ClientOrchestrationStatus } from "../orchestrati
 import { ParentOrchestrationInstance } from "../types/parent-orchestration-instance.type";
 import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb";
 import { randomUUID } from "crypto";
+import { mapToRecord } from "../utils/tags.util";
 import { validateDedupeStatusesForReplacement } from "../orchestration/orchestration-id-reuse-policy";
 import { OrchestrationAlreadyExistsError } from "../orchestration/exception/orchestration-already-exists-error";
 
@@ -43,6 +44,8 @@ export interface ActivityWorkItem {
   name: string;
   taskId: number;
   input?: string;
+  version?: string;
+  tags?: Record<string, string>;
   completionToken: number;
 }
 
@@ -1054,9 +1057,7 @@ export class InMemoryOrchestrationBackend {
         ?.getExecutionstarted()
         ?.getVersion()
         ?.getValue();
-      const newVersion = completeAction.hasNewversion()
-        ? completeAction.getNewversion()?.getValue()
-        : currentVersion;
+      const newVersion = completeAction.hasNewversion() ? completeAction.getNewversion()?.getValue() : currentVersion;
       const carryoverEvents = completeAction.getCarryovereventsList();
 
       // Cancel timers still pending from the previous iteration. Their timer IDs are
@@ -1092,7 +1093,7 @@ export class InMemoryOrchestrationBackend {
         newInput,
         undefined,
         instance.executionId,
-        newVersion
+        newVersion,
       );
       instance.pendingEvents = [orchestratorStarted, executionStarted, ...carryoverEvents];
 
@@ -1105,6 +1106,8 @@ export class InMemoryOrchestrationBackend {
     const taskId = action.getId();
     const taskName = scheduleTask.getName();
     const input = scheduleTask.getInput()?.getValue();
+    const version = scheduleTask.getVersion()?.getValue() || undefined;
+    const tags = mapToRecord(scheduleTask.getTagsMap());
 
     // Add TaskScheduled event to history
     const event = pbh.newTaskScheduledEvent(taskId, taskName, input);
@@ -1122,6 +1125,8 @@ export class InMemoryOrchestrationBackend {
       name: taskName,
       taskId,
       input,
+      version,
+      tags,
       completionToken: instance.completionToken,
     });
   }

packages/durabletask-js/src/testing/test-worker.ts

@@ -20,10 +20,11 @@ import {
 import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb";
 import * as pb from "../proto/orchestrator_service_pb";
 import * as pbh from "../utils/pb-helper.util";
+import { ActivityMiddleware, OrchestrationMiddleware } from "../worker/middleware";
 
 /**
  * Worker that processes orchestrations and activities from the in-memory backend.
- * 
+ *
  * This worker runs in the same process as the test and processes work items
  * synchronously in the Node.js event loop, avoiding the need for a separate
  * sidecar process.
@@ -34,6 +35,8 @@ export class TestOrchestrationWorker {
   private isRunning: boolean = false;
   private processingPromise: Promise<void> | null = null;
   private stopRequested: boolean = false;
+  private readonly orchestrationMiddleware: OrchestrationMiddleware[] = [];
+  private readonly activityMiddleware: ActivityMiddleware[] = [];
 
   constructor(backend: InMemoryOrchestrationBackend) {
     this.registry = new Registry();
@@ -103,6 +106,28 @@ export class TestOrchestrationWorker {
     return name;
   }
 
+  /**
+   * Adds orchestration middleware to this worker.
+   */
+  useOrchestrationMiddleware(middleware: OrchestrationMiddleware): this {
+    if (this.isRunning) {
+      throw new Error("Cannot add orchestration middleware while worker is running.");
+    }
+    this.orchestrationMiddleware.push(middleware);
+    return this;
+  }
+
+  /**
+   * Adds activity middleware to this worker.
+   */
+  useActivityMiddleware(middleware: ActivityMiddleware): this {
+    if (this.isRunning) {
+      throw new Error("Cannot add activity middleware while worker is running.");
+    }
+    this.activityMiddleware.push(middleware);
+    return this;
+  }
+
   /**
    * Starts the worker processing loop.
    */
@@ -178,7 +203,7 @@ export class TestOrchestrationWorker {
     const completionToken = instance.completionToken;
 
     try {
-      const executor = new OrchestrationExecutor(this.registry);
+      const executor = new OrchestrationExecutor(this.registry, undefined, this.orchestrationMiddleware);
       const result = await executor.execute(instanceId, instance.history, instance.pendingEvents, instance.executionId);
 
       this.backend.completeOrchestration(instanceId, completionToken, result.actions, result.customStatus);
@@ -200,11 +225,11 @@ export class TestOrchestrationWorker {
    * Processes a single activity work item.
    */
   private async processActivity(workItem: ActivityWorkItem): Promise<void> {
-    const { instanceId, executionId, name, taskId, input } = workItem;
+    const { instanceId, executionId, name, taskId, input, version, tags } = workItem;
 
     try {
-      const executor = new ActivityExecutor(this.registry);
-      const result = await executor.execute(instanceId, name, taskId, input);
+      const executor = new ActivityExecutor(this.registry, undefined, this.activityMiddleware);
+      const result = await executor.execute(instanceId, name, taskId, input, { version, tags });
       this.backend.completeActivity(instanceId, executionId, taskId, result);
     } catch (error: unknown) {
       const err = error instanceof Error ? error : new Error(String(error));