feat(openworkflow): pluggable logger

apps/docs/docs/workers.mdx

@@ -188,6 +188,32 @@ bunx @openworkflow/cli worker start
 
 </CodeGroup>
 
+## Logging
+
+Workers log operational errors, such as failed polls and heartbeats, to
+`console` by default. Pass a `logger` with `debug`, `info`, `warn`, and `error`
+methods to send them somewhere else, such as a structured logger:
+
+```ts
+import pino from "pino";
+
+const log = pino();
+
+const ow = new OpenWorkflow({
+  backend,
+  logger: {
+    debug: (message, ...args) => log.debug({ args }, message),
+    info: (message, ...args) => log.info({ args }, message),
+    warn: (message, ...args) => log.warn({ args }, message),
+    // workers pass the error after the message; `err` gets pino's serializer
+    error: (message, err, ...args) => log.error({ err, args }, message),
+  },
+});
+```
+
+Workers created with `ow.newWorker()` use the client's logger. You can also
+pass `logger` to `newWorker()` or to `new Worker()` directly.
+
 ## Workflow Discovery
 
 The CLI automatically discovers workflows from directories specified in your

packages/openworkflow/client/client.ts

@@ -1,6 +1,7 @@
 import type { Backend, SendSignalResult } from "../core/backend.js";
 import type { DurationString } from "../core/duration.js";
 import type { JsonValue } from "../core/json.js";
+import type { Logger } from "../core/logger.js";
 import type { StandardSchemaV1 } from "../core/standard-schema.js";
 import { calculateDateFromDuration } from "../core/step-attempt.js";
 import {
@@ -44,34 +45,43 @@ type WorkflowRunInput<TSchema, Input> = SchemaInput<TSchema, Input>;
  */
 export interface OpenWorkflowOptions {
   backend: Backend;
+  /**
+   * Default logger for workers created with `newWorker()`.
+   * @default console
+   */
+  logger?: Logger | undefined;
 }
 
 /**
  * Client used to register workflows and start runs.
  */
 export class OpenWorkflow {
   private backend: Backend;
+  private logger: Logger | undefined;
   private registry = new WorkflowRegistry();
 
   constructor(options: OpenWorkflowOptions) {
     this.backend = options.backend;
+    this.logger = options.logger;
   }
 
   /**
    * Create a new Worker with this client's backend and workflows.
    * @param options - Worker options
    * @param options.concurrency - Max concurrent workflow runs
    * @param options.leaseDurationMs - Workflow run lease duration
+   * @param options.logger - Logger, overriding the client's logger
    * @returns Worker instance
    */
   newWorker(
-    options?: Pick<WorkerOptions, "concurrency" | "leaseDurationMs">,
+    options?: Pick<WorkerOptions, "concurrency" | "leaseDurationMs" | "logger">,
   ): Worker {
     return new Worker({
       backend: this.backend,
       workflows: this.registry.getAll(),
       concurrency: options?.concurrency,
       leaseDurationMs: options?.leaseDurationMs,
+      logger: options?.logger ?? this.logger,
     });
   }
 

packages/openworkflow/core/logger.ts

@@ -0,0 +1,16 @@
+/**
+ * Receives operational log output from OpenWorkflow, such as worker loop and
+ * heartbeat errors. The method signatures match `console`, so `console` itself
+ * is a valid logger, and structured loggers can be adapted with a thin wrapper.
+ */
+export interface Logger {
+  debug(message: string, ...args: unknown[]): void;
+  info(message: string, ...args: unknown[]): void;
+  warn(message: string, ...args: unknown[]): void;
+  error(message: string, ...args: unknown[]): void;
+}
+
+/**
+ * The default logger, which writes to the global `console`.
+ */
+export const consoleLogger: Logger = console;

packages/openworkflow/index.ts

@@ -3,6 +3,7 @@ export type { OpenWorkflowOptions } from "./client/client.js";
 export { OpenWorkflow } from "./client/client.js";
 
 // core
+export type { Logger } from "./core/logger.js";
 export type { RetryPolicy, Workflow } from "./core/workflow-definition.js";
 export type { WorkflowRunMetadata } from "./core/workflow-function.js";
 export {

packages/openworkflow/worker/worker.test.ts

@@ -1701,6 +1701,175 @@ describe("Worker", () => {
     }
   });
 
+  describe("logger", () => {
+    test("logs tick failures to console by default", async () => {
+      const error = new Error("boom");
+      const consoleErrorSpy = vi.spyOn(console, "error").mockReturnValue();
+      const worker = new Worker({
+        backend: createStubBackend({
+          claimWorkflowRun: vi.fn().mockRejectedValue(error),
+        }),
+        workflows: [],
+      });
+
+      try {
+        await worker.start();
+        await worker.stop();
+
+        expect(consoleErrorSpy).toHaveBeenCalledWith(
+          "Worker tick failed:",
+          error,
+        );
+      } finally {
+        consoleErrorSpy.mockRestore();
+      }
+    });
+
+    test("logs tick failures to a custom logger", async () => {
+      const error = new Error("boom");
+      const consoleErrorSpy = vi.spyOn(console, "error").mockReturnValue();
+      const logger = createTestLogger();
+      const worker = new Worker({
+        backend: createStubBackend({
+          claimWorkflowRun: vi.fn().mockRejectedValue(error),
+        }),
+        workflows: [],
+        logger,
+      });
+
+      try {
+        await worker.start();
+        await worker.stop();
+
+        expect(logger.error).toHaveBeenCalledWith("Worker tick failed:", error);
+        expect(consoleErrorSpy).not.toHaveBeenCalled();
+      } finally {
+        consoleErrorSpy.mockRestore();
+      }
+    });
+
+    test("logs heartbeat failures to a custom logger", async () => {
+      vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
+
+      const backend = await createTestBackend();
+      const error = new Error("heartbeat boom");
+      vi.spyOn(backend, "extendWorkflowRunLease").mockRejectedValue(error);
+      const logger = createTestLogger();
+      const client = new OpenWorkflow({ backend });
+      const stepStarted = Promise.withResolvers<null>();
+      const releaseStep = Promise.withResolvers<null>();
+
+      const workflow = client.defineWorkflow(
+        { name: "heartbeat-logger" },
+        async ({ step }) => {
+          await step.run({ name: "wait" }, async () => {
+            stepStarted.resolve(null);
+            await releaseStep.promise;
+          });
+          return "done";
+        },
+      );
+      const worker = client.newWorker({ logger });
+
+      try {
+        const handle = await workflow.run();
+        await worker.tick();
+        await stepStarted.promise;
+
+        await vi.advanceTimersByTimeAsync(15_000);
+        expect(logger.error).toHaveBeenCalledWith("Heartbeat failed:", error);
+
+        releaseStep.resolve(null);
+        await expect(handle.result()).resolves.toBe("done");
+      } finally {
+        releaseStep.resolve(null);
+        await worker.stop();
+        vi.useRealTimers();
+      }
+    });
+
+    test("logs critical execution failures to a custom logger", async () => {
+      const backend = await createTestBackend();
+      const error = new Error("backend boom");
+      vi.spyOn(backend, "failWorkflowRun").mockRejectedValue(error);
+      const logger = createTestLogger();
+      const client = new OpenWorkflow({ backend });
+      const workflow = client.defineWorkflow(
+        { name: "critical-execution-logger" },
+        () => {
+          throw new Error("workflow boom");
+        },
+      );
+      const worker = client.newWorker({ logger });
+      const handle = await workflow.run();
+
+      await worker.tick();
+      await expect(worker.stop()).resolves.toBeUndefined();
+
+      expect(logger.error).toHaveBeenCalledWith(
+        `Critical error during workflow execution for run ${handle.workflowRun.id}:`,
+        error,
+      );
+      expect(logger.error.mock.calls[0]?.[1]).toBe(error);
+    });
+
+    test("keeps polling when the logger throws", async () => {
+      vi.useFakeTimers();
+
+      const claimWorkflowRun = vi.fn().mockRejectedValue(new Error("boom"));
+      const logger = createTestLogger();
+      logger.error.mockImplementation(() => {
+        throw new Error("logger boom");
+      });
+      const worker = new Worker({
+        backend: createStubBackend({ claimWorkflowRun }),
+        workflows: [],
+        logger,
+      });
+
+      try {
+        await worker.start();
+        await vi.advanceTimersByTimeAsync(100);
+
+        expect(claimWorkflowRun).toHaveBeenCalledTimes(2);
+        expect(logger.error).toHaveBeenCalledTimes(2);
+      } finally {
+        const stopPromise = worker.stop();
+        await vi.runOnlyPendingTimersAsync();
+        await stopPromise;
+        vi.useRealTimers();
+      }
+    });
+
+    test("uses the client logger unless the worker overrides it", async () => {
+      const error = new Error("boom");
+      const backend = createStubBackend({
+        claimWorkflowRun: vi.fn().mockRejectedValue(error),
+      });
+      const clientLogger = createTestLogger();
+      const workerLogger = createTestLogger();
+      const client = new OpenWorkflow({ backend, logger: clientLogger });
+
+      const defaultWorker = client.newWorker();
+      await defaultWorker.start();
+      await defaultWorker.stop();
+      expect(clientLogger.error).toHaveBeenCalledWith(
+        "Worker tick failed:",
+        error,
+      );
+
+      clientLogger.error.mockClear();
+      const overriddenWorker = client.newWorker({ logger: workerLogger });
+      await overriddenWorker.start();
+      await overriddenWorker.stop();
+      expect(workerLogger.error).toHaveBeenCalledWith(
+        "Worker tick failed:",
+        error,
+      );
+      expect(clientLogger.error).not.toHaveBeenCalled();
+    });
+  });
+
   test("respects custom retry policy from spec", async () => {
     const backend = await createTestBackend();
     const client = new OpenWorkflow({ backend });
@@ -2288,6 +2457,10 @@ function retryDelayMs(run: WorkflowRun, fallbackStartMs: number): number {
   return run.availableAt.getTime() - fallbackStartMs;
 }
 
+function createTestLogger() {
+  return { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() };
+}
+
 function sleep(ms: number): Promise<void> {
   return new Promise((resolve) => {
     setTimeout(resolve, ms);