Add wrap-the-lifetime hooks on client and worker plugins

`ClientPlugin.connectClient` wraps both `TemporalClient.connect` and
the `ServiceLifecycle` `run()`. `WorkerPlugin.runWorker` wraps
`TemporalWorker.run`, and `WorkerPlugin.runReplayer` wraps each
`replayWorkflow` call. Defaults forward to `next`, so existing
plugins are unaffected.

The chain helpers recurse over the plugin array rather than folding
with `reduce`, which keeps the wrapped body non-`@escaping` and
preserves the isolation semantics of `withGRPCClient`. First plugin
in the array is the outermost wrap, matching Ruby and C#.

`SimplePlugin` exposes the three wraps as optional `@Sendable`
before/after closures, so authors can bracket the connect,
worker-run, or replay lifetimes without a custom
`ClientPlugin`/`WorkerPlugin` conformance. Matches Ruby's
`connect_client`/`run_worker`/`with_workflow_replay_worker` hooks
on `SimplePlugin` and C#'s `SimplePlugin` virtual lifecycle methods.

Closes #111.

Author: RalucaP

Sources/Temporal/Client/Plugins/ClientPlugin.swift

@@ -40,6 +40,15 @@
 /// An interceptor wraps each outbound call at runtime, while a plugin runs once at configuration
 /// time. A plugin can contribute interceptors by appending them to
 /// ``TemporalClient/Configuration/interceptors`` during ``configure(_:)``.
+///
+/// ## Wrapping the connected client lifetime
+///
+/// A plugin can also observe and surround the lifetime of a connected client by overriding
+/// ``connectClient(configuration:next:)``. The method receives the merged configuration and a
+/// continuation closure, and is expected to invoke the continuation as part of its body. Wrapping
+/// is useful for emitting telemetry around connect/disconnect, mapping errors, or running setup
+/// and teardown that should bracket the client's connected state. When plugins compose, the first
+/// plugin in ``TemporalClient/Configuration/plugins`` is the outermost wrap.
 public protocol ClientPlugin: Sendable {
     /// A human-readable name used in diagnostics and ordering.
     ///
@@ -57,6 +66,38 @@ public protocol ClientPlugin: Sendable {
     ///
     /// - Parameter configuration: The configuration to customize.
     func configure(_ configuration: inout TemporalClient.Configuration)
+
+    /// Wraps the lifetime of a connected client.
+    ///
+    /// The default implementation forwards through to `next`, so plugins that do not need to
+    /// surround the connected lifetime can omit this method. Override to bracket the connected
+    /// client with custom logic, such as logging, metrics, or error mapping. Implementations must
+    /// invoke `next` exactly once and return whatever value it produces, so the value returned by
+    /// the public connect API is preserved. Calling `next` zero times or more than once is
+    /// undefined behavior; the SDK reserves the right to detect and reject it in a future release.
+    ///
+    /// When the configuration carries multiple plugins, the first plugin in
+    /// ``TemporalClient/Configuration/plugins`` is the outermost wrap and the last plugin is the
+    /// innermost. The continuation closure invokes the next plugin in the chain, and ultimately
+    /// the gRPC connection plus the body supplied to the public connect entry point.
+    ///
+    /// The wrap scope differs between the two public client entry points:
+    /// - ``TemporalClient/connect(transport:configuration:logger:_:)`` brackets the entire gRPC
+    ///   client lifetime, including transport open and close.
+    /// - For the two-phase ``TemporalClient`` `Service` API (init then `run()`), the wrap brackets
+    ///   the `run()` call only — the gRPC client is constructed by the init step before any plugin
+    ///   sees it.
+    ///
+    /// - Parameters:
+    ///   - configuration: The merged configuration the client is about to use. Plugins receive
+    ///     the configuration after every plugin's ``configure(_:)`` has run.
+    ///   - next: A continuation that, when invoked, runs the rest of the plugin chain and the
+    ///     underlying connected work. Returns the value produced by the public connect entry point.
+    /// - Returns: The value produced by `next`, optionally transformed by the plugin.
+    func connectClient<R: Sendable>(
+        configuration: TemporalClient.Configuration,
+        next: () async throws -> sending R
+    ) async throws -> sending R
 }
 
 extension ClientPlugin {
@@ -65,4 +106,11 @@ extension ClientPlugin {
     }
 
     public func configure(_ configuration: inout TemporalClient.Configuration) {}
+
+    public func connectClient<R: Sendable>(
+        configuration: TemporalClient.Configuration,
+        next: () async throws -> sending R
+    ) async throws -> sending R {
+        try await next()
+    }
 }

Sources/Temporal/Client/Plugins/TemporalClient+Plugins.swift

@@ -39,3 +39,26 @@ extension TemporalClient.Configuration {
         }
     }
 }
+
+/// Runs `body` wrapped by each plugin's ``ClientPlugin/connectClient(configuration:next:)``.
+///
+/// The first plugin in `plugins` is the outermost wrap and the last plugin is the innermost.
+/// When `plugins` is empty, `body` runs directly. The helper is recursive so that `body` can be
+/// passed through as a non-escaping closure, which lets callers forward a non-`@escaping` body
+/// parameter from a public connect entry point.
+package func applyClientConnectChain<R: Sendable>(
+    plugins: ArraySlice<any ClientPlugin>,
+    configuration: TemporalClient.Configuration,
+    body: () async throws -> sending R
+) async throws -> sending R {
+    guard let plugin = plugins.first else {
+        return try await body()
+    }
+    return try await plugin.connectClient(configuration: configuration) {
+        try await applyClientConnectChain(
+            plugins: plugins.dropFirst(),
+            configuration: configuration,
+            body: body
+        )
+    }
+}

Sources/Temporal/Client/TemporalClient+ServiceLifecycle.swift

@@ -50,11 +50,20 @@ extension TemporalClient: Service {
     /// Starts the Temporal client and keeps it running until shutdown is initiated.
     ///
     /// This method starts the underlying gRPC client and keeps it running to handle requests.
+    /// Plugins set on ``Configuration/plugins`` wrap the run via
+    /// ``ClientPlugin/connectClient(configuration:next:)``, with the first plugin in the array as
+    /// the outermost wrap.
     ///
     /// - Throws: A runtime error if the client cannot be started or is in an invalid state.
     public func run() async throws {
         // Graceful shutdown is handled by `GRPCServiceLifecycle`
-        try await self.workflowService.client.client.run()
+        let configuration = self.configuration
+        try await applyClientConnectChain(
+            plugins: configuration.plugins[...],
+            configuration: configuration
+        ) {
+            try await self.workflowService.client.client.run()
+        }
     }
 
     /// Initiates graceful shutdown of the Temporal client.

Sources/Temporal/Client/TemporalClient.swift

@@ -173,13 +173,19 @@ public final class TemporalClient: Sendable {
     ) async throws -> sending Result {
         var configuration = configuration
         configuration.applyPlugins(logger: logger)
-        return try await withGRPCClient(
-            transport: transport,
-            interceptors: Self.grpcClientInterceptors(serverHostname: configuration.instrumentation.serverHostname, logger: logger),
-            isolation: #isolation
-        ) { client in
-            let temporalClient = Self.init(client: client, configuration: configuration)
-            return try await body(temporalClient)
+        let plugins = configuration.plugins
+        return try await applyClientConnectChain(
+            plugins: plugins[...],
+            configuration: configuration
+        ) {
+            try await withGRPCClient(
+                transport: transport,
+                interceptors: Self.grpcClientInterceptors(serverHostname: configuration.instrumentation.serverHostname, logger: logger),
+                isolation: #isolation
+            ) { client in
+                let temporalClient = Self.init(client: client, configuration: configuration)
+                return try await body(temporalClient)
+            }
         }
     }
 }

Sources/Temporal/Documentation.docc/Temporal.md

@@ -71,6 +71,14 @@ Configure workers and clients for different environments.
 - ``TemporalWorker/Configuration``
 - ``TemporalClient/Configuration``
 
+### Extending with plugins
+
+Customize clients, workers, and replayers with reusable plugins.
+
+- ``ClientPlugin``
+- ``WorkerPlugin``
+- ``SimplePlugin``
+
 ## See Also
 
 - [Temporal Documentation - Complete platform documentation](https://docs.temporal.io)

Sources/Temporal/Plugins/SimplePlugin.swift

@@ -49,6 +49,17 @@
 ///     dataConverter: { existing in EncryptingDataConverter(wrapping: existing) }
 /// )
 /// ```
+///
+/// ## Bracketing the connect, run, and replay lifetimes
+///
+/// To run setup or teardown around the connected-client, worker-run, or replay lifetimes, pass the
+/// matching `before`/`after` closures. The before closure runs immediately before the wrapped
+/// work, the after closure immediately after it returns successfully; if the wrapped work throws,
+/// the after closure does not run. If the before closure itself throws, the error propagates and
+/// neither the wrapped work nor the after closure runs. For full wrap semantics, such as
+/// short-circuiting or mapping errors, conform to ``ClientPlugin`` or ``WorkerPlugin`` directly
+/// and override the `connectClient(configuration:next:)`, `runWorker(configuration:next:)`, or
+/// `runReplayer(configuration:next:)` method.
 public struct SimplePlugin: ClientPlugin, WorkerPlugin {
     /// A human-readable name used in diagnostics and ordering.
     public let name: String
@@ -68,6 +79,30 @@ public struct SimplePlugin: ClientPlugin, WorkerPlugin {
     /// Workflows the plugin registers with the worker and replayer.
     public let workflows: [any WorkflowDefinition.Type]
 
+    /// A closure invoked immediately before the inner work inside ``connectClient(configuration:next:)``.
+    public let beforeConnect: (@Sendable (TemporalClient.Configuration) async throws -> Void)?
+
+    /// A closure invoked immediately after the inner work inside ``connectClient(configuration:next:)``.
+    ///
+    /// Does not run if the inner work throws.
+    public let afterConnect: (@Sendable (TemporalClient.Configuration) async throws -> Void)?
+
+    /// A closure invoked immediately before the inner work inside ``runWorker(configuration:next:)``.
+    public let beforeRunWorker: (@Sendable (TemporalWorker.Configuration) async throws -> Void)?
+
+    /// A closure invoked immediately after the inner work inside ``runWorker(configuration:next:)``.
+    ///
+    /// Does not run if the inner work throws.
+    public let afterRunWorker: (@Sendable (TemporalWorker.Configuration) async throws -> Void)?
+
+    /// A closure invoked immediately before the inner work inside ``runReplayer(configuration:next:)``.
+    public let beforeRunReplayer: (@Sendable (WorkflowReplayer.Configuration) async throws -> Void)?
+
+    /// A closure invoked immediately after the inner work inside ``runReplayer(configuration:next:)``.
+    ///
+    /// Does not run if the inner work throws.
+    public let afterRunReplayer: (@Sendable (WorkflowReplayer.Configuration) async throws -> Void)?
+
     /// Creates a new simple plugin.
     ///
     /// - Parameters:
@@ -80,20 +115,41 @@ public struct SimplePlugin: ClientPlugin, WorkerPlugin {
     ///     Defaults to empty.
     ///   - activities: Activities registered with the worker. Defaults to empty.
     ///   - workflows: Workflows registered with the worker and replayer. Defaults to empty.
+    ///   - beforeConnect: An optional closure run immediately before the connected-client work.
+    ///   - afterConnect: An optional closure run immediately after the connected-client work
+    ///     returns successfully.
+    ///   - beforeRunWorker: An optional closure run immediately before the worker's run loop.
+    ///   - afterRunWorker: An optional closure run immediately after the worker's run loop returns
+    ///     successfully.
+    ///   - beforeRunReplayer: An optional closure run immediately before a single replay.
+    ///   - afterRunReplayer: An optional closure run immediately after a single replay returns
+    ///     successfully.
     public init(
         name: String,
         dataConverter: (@Sendable (DataConverter) -> DataConverter)? = nil,
         clientInterceptors: [any ClientInterceptor] = [],
         workerInterceptors: [any WorkerInterceptor] = [],
         activities: [any ActivityDefinition] = [],
-        workflows: [any WorkflowDefinition.Type] = []
+        workflows: [any WorkflowDefinition.Type] = [],
+        beforeConnect: (@Sendable (TemporalClient.Configuration) async throws -> Void)? = nil,
+        afterConnect: (@Sendable (TemporalClient.Configuration) async throws -> Void)? = nil,
+        beforeRunWorker: (@Sendable (TemporalWorker.Configuration) async throws -> Void)? = nil,
+        afterRunWorker: (@Sendable (TemporalWorker.Configuration) async throws -> Void)? = nil,
+        beforeRunReplayer: (@Sendable (WorkflowReplayer.Configuration) async throws -> Void)? = nil,
+        afterRunReplayer: (@Sendable (WorkflowReplayer.Configuration) async throws -> Void)? = nil
     ) {
         self.name = name
         self.dataConverter = dataConverter
         self.clientInterceptors = clientInterceptors
         self.workerInterceptors = workerInterceptors
         self.activities = activities
         self.workflows = workflows
+        self.beforeConnect = beforeConnect
+        self.afterConnect = afterConnect
+        self.beforeRunWorker = beforeRunWorker
+        self.afterRunWorker = afterRunWorker
+        self.beforeRunReplayer = beforeRunReplayer
+        self.afterRunReplayer = afterRunReplayer
     }
 
     public func configure(_ configuration: inout TemporalClient.Configuration) {
@@ -116,4 +172,33 @@ public struct SimplePlugin: ClientPlugin, WorkerPlugin {
         }
         configuration.interceptors.append(contentsOf: workerInterceptors)
     }
+
+    public func connectClient<R: Sendable>(
+        configuration: TemporalClient.Configuration,
+        next: () async throws -> sending R
+    ) async throws -> sending R {
+        try await beforeConnect?(configuration)
+        let result = try await next()
+        try await afterConnect?(configuration)
+        return result
+    }
+
+    public func runWorker(
+        configuration: TemporalWorker.Configuration,
+        next: () async throws -> Void
+    ) async throws {
+        try await beforeRunWorker?(configuration)
+        try await next()
+        try await afterRunWorker?(configuration)
+    }
+
+    public func runReplayer<R: Sendable>(
+        configuration: WorkflowReplayer.Configuration,
+        next: () async throws -> sending R
+    ) async throws -> sending R {
+        try await beforeRunReplayer?(configuration)
+        let result = try await next()
+        try await afterRunReplayer?(configuration)
+        return result
+    }
 }

Sources/Temporal/Replayer/WorkflowReplayer.swift

@@ -62,6 +62,10 @@ public final class WorkflowReplayer: Sendable {
 
     /// Replays a single workflow history.
     ///
+    /// Plugins set on ``Configuration/plugins`` wrap each replay invocation via
+    /// ``WorkerPlugin/runReplayer(configuration:next:)``, with the first plugin in the array as the
+    /// outermost wrap.
+    ///
     /// - Parameters:
     ///   - history: The workflow history to replay.
     ///   - throwOnReplayFailure: If `true` (the default), throws an error when replay
@@ -75,15 +79,25 @@ public final class WorkflowReplayer: Sendable {
         history: WorkflowHistory,
         throwOnReplayFailure: Bool = true
     ) async throws -> WorkflowReplayResult {
-        let runner = try WorkflowHistoryRunner(configuration: self.configuration)
-        return try await runner.replayWorkflow(
-            history: history,
-            throwOnReplayFailure: throwOnReplayFailure
-        )
+        try await applyReplayerRunChain(
+            plugins: self.configuration.plugins[...],
+            configuration: self.configuration
+        ) {
+            let runner = try WorkflowHistoryRunner(configuration: self.configuration)
+            return try await runner.replayWorkflow(
+                history: history,
+                throwOnReplayFailure: throwOnReplayFailure
+            )
+        }
     }
 
     /// Replays multiple workflow histories.
     ///
+    /// Each history is replayed by delegating to ``replayWorkflow(history:throwOnReplayFailure:)``,
+    /// so the plugin chain wraps each replay individually rather than the batch as a whole. A
+    /// plugin that records latency in ``WorkerPlugin/runReplayer(configuration:next:)`` therefore
+    /// observes one sample per history.
+    ///
     /// - Parameters:
     ///   - histories: The workflow histories to replay.
     ///   - throwOnReplayFailure: If `true` (the default), throws on the first replay

Sources/Temporal/Worker/Plugins/TemporalWorker+Plugins.swift

@@ -72,3 +72,46 @@ extension WorkflowReplayer.Configuration {
         self.workflows += plugins.flatMap(\.workflows)
     }
 }
+
+/// Runs `body` wrapped by each plugin's ``WorkerPlugin/runWorker(configuration:next:)``.
+///
+/// The first plugin in `plugins` is the outermost wrap and the last plugin is the innermost.
+/// When `plugins` is empty, `body` runs directly.
+package func applyWorkerRunChain(
+    plugins: ArraySlice<any WorkerPlugin>,
+    configuration: TemporalWorker.Configuration,
+    body: () async throws -> Void
+) async throws {
+    guard let plugin = plugins.first else {
+        try await body()
+        return
+    }
+    try await plugin.runWorker(configuration: configuration) {
+        try await applyWorkerRunChain(
+            plugins: plugins.dropFirst(),
+            configuration: configuration,
+            body: body
+        )
+    }
+}
+
+/// Runs `body` wrapped by each plugin's ``WorkerPlugin/runReplayer(configuration:next:)``.
+///
+/// The first plugin in `plugins` is the outermost wrap and the last plugin is the innermost.
+/// When `plugins` is empty, `body` runs directly.
+package func applyReplayerRunChain<R: Sendable>(
+    plugins: ArraySlice<any WorkerPlugin>,
+    configuration: WorkflowReplayer.Configuration,
+    body: () async throws -> sending R
+) async throws -> sending R {
+    guard let plugin = plugins.first else {
+        return try await body()
+    }
+    return try await plugin.runReplayer(configuration: configuration) {
+        try await applyReplayerRunChain(
+            plugins: plugins.dropFirst(),
+            configuration: configuration,
+            body: body
+        )
+    }
+}