diff --git a/.changeset/failed-fan-in-duplicates.md b/.changeset/failed-fan-in-duplicates.md
new file mode 100644
index 00000000..2f8654fd
--- /dev/null
+++ b/.changeset/failed-fan-in-duplicates.md
@@ -0,0 +1,11 @@
+---
+"@convex-dev/workflow": patch
+---
+
+A workflow that fails after parallel steps is no longer replayed by the extra
+handler runs queued by those steps. Those runs now return early instead of
+failing with "Workflow not running" and being counted as Workpool failures.
+(#327)
+
+Handler runs and handler failure notifications for a workflow that already
+finished, was restarted or was cleaned up now return without throwing. (#328)
diff --git a/.changeset/journal-load-short-circuit.md b/.changeset/journal-load-short-circuit.md
new file mode 100644
index 00000000..79c72c3d
--- /dev/null
+++ b/.changeset/journal-load-short-circuit.md
@@ -0,0 +1,6 @@
+---
+"@convex-dev/workflow": patch
+---
+
+A handler run for a workflow that already finished no longer reads its journal
+before returning.
diff --git a/example/convex/_generated/api.d.ts b/example/convex/_generated/api.d.ts
index d4da517a..91b96422 100644
--- a/example/convex/_generated/api.d.ts
+++ b/example/convex/_generated/api.d.ts
@@ -16,6 +16,7 @@ import type * as nestedWorkflow from "../nestedWorkflow.js";
 import type * as oversized from "../oversized.js";
 import type * as passingSignals from "../passingSignals.js";
 import type * as test_contextRoundtrip from "../test/contextRoundtrip.js";
+import type * as test_fanIn from "../test/fanIn.js";
 import type * as test_inline from "../test/inline.js";
 import type * as test_oldSyntax from "../test/oldSyntax.js";
 import type * as transcription from "../transcription.js";
@@ -36,6 +37,7 @@ declare const fullApi: ApiFromModules<{
   oversized: typeof oversized;
   passingSignals: typeof passingSignals;
   "test/contextRoundtrip": typeof test_contextRoundtrip;
+  "test/fanIn": typeof test_fanIn;
   "test/inline": typeof test_inline;
   "test/oldSyntax": typeof test_oldSyntax;
   transcription: typeof transcription;
diff --git a/example/convex/test/fanIn.test.ts b/example/convex/test/fanIn.test.ts
new file mode 100644
index 00000000..80544866
--- /dev/null
+++ b/example/convex/test/fanIn.test.ts
@@ -0,0 +1,84 @@
+/// <reference types="vite/client" />
+
+import { getStatus } from "@convex-dev/workflow";
+import { makeFunctionReference } from "convex/server";
+import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+import { handlerOnComplete } from "../../../src/component/pool.js";
+import { components, internal } from "../_generated/api";
+import { initConvexTest } from "../setup.test";
+import { handlerRuns, workflow } from "./fanIn";
+
+type WithHandler<T> = T & { _handler: (...args: unknown[]) => unknown };
+
+describe("fan-in of parallel steps", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    handlerRuns.count = 0;
+  });
+  afterEach(() => {
+    vi.restoreAllMocks();
+    vi.useRealTimers();
+  });
+
+  test.each([{ fail: true }, { fail: false }])(
+    "replays the handler once when all steps finish (fail: $fail)",
+    async ({ fail }) => {
+      const t = initConvexTest();
+      // Called once per handler work item that the workpool records as failed.
+      const failureCallback = vi.spyOn(
+        handlerOnComplete as WithHandler<typeof handlerOnComplete>,
+        "_handler",
+      );
+      const workflowId = await t.run((ctx) =>
+        workflow.start(ctx, internal.test.fanIn.fanIn, { width: 8, fail }),
+      );
+      await t.finishAllScheduledFunctions(vi.runAllTimers);
+      const status = await t.run((ctx) =>
+        getStatus(ctx, components.workflow, workflowId),
+      );
+      if (fail) {
+        expect(status).toMatchObject({
+          type: "failed",
+          error: expect.stringContaining("boom"),
+        });
+      } else {
+        expect(status).toEqual({ type: "completed", result: 8 });
+      }
+      // One run to start the steps and one replay once they have all finished.
+      // The other queued runs return without replaying.
+      expect(handlerRuns.count).toBe(2);
+      expect(failureCallback).not.toHaveBeenCalled();
+    },
+  );
+
+  test.each([{ fail: true }, { fail: false }])(
+    "a duplicate run of a finished workflow does not replay (fail: $fail)",
+    async ({ fail }) => {
+      const t = initConvexTest();
+      const workflowId = await t.run((ctx) =>
+        workflow.start(ctx, internal.test.fanIn.fanIn, { width: 3, fail }),
+      );
+      await t.finishAllScheduledFunctions(vi.runAllTimers);
+      const before = await t.run((ctx) =>
+        getStatus(ctx, components.workflow, workflowId),
+      );
+      const runsBefore = handlerRuns.count;
+
+      // What a second handler work item for the same generation would do.
+      const result = await t.mutation(
+        makeFunctionReference<
+          "mutation",
+          { workflowId: string; generationNumber: number }
+        >("test/fanIn:fanIn"),
+        { workflowId, generationNumber: 0 },
+      );
+
+      expect(result).toBe(workflowId);
+      expect(handlerRuns.count).toBe(runsBefore);
+      const after = await t.run((ctx) =>
+        getStatus(ctx, components.workflow, workflowId),
+      );
+      expect(after).toEqual(before);
+    },
+  );
+});
diff --git a/example/convex/test/fanIn.ts b/example/convex/test/fanIn.ts
new file mode 100644
index 00000000..9882d94f
--- /dev/null
+++ b/example/convex/test/fanIn.ts
@@ -0,0 +1,35 @@
+import { WorkflowManager } from "@convex-dev/workflow";
+import { v } from "convex/values";
+import { components, internal } from "../_generated/api.js";
+import { internalMutation } from "../_generated/server.js";
+
+export const workflow = new WorkflowManager(components.workflow);
+
+// Counts handler invocations (including replays), so tests can observe
+// redundant replays. Only meaningful in convex-test, where modules are shared.
+export const handlerRuns = { count: 0 };
+
+export const noop = internalMutation({
+  args: {},
+  returns: v.null(),
+  handler: async () => null,
+});
+
+// Many steps complete together, then the handler fails or succeeds.
+export const fanIn = workflow
+  .define({
+    args: { width: v.number(), fail: v.boolean() },
+    returns: v.number(),
+  })
+  .handler(async (step, args): Promise<number> => {
+    handlerRuns.count++;
+    await Promise.all(
+      Array.from({ length: args.width }, () =>
+        step.runMutation(internal.test.fanIn.noop, {}),
+      ),
+    );
+    if (args.fail) {
+      throw new Error("boom");
+    }
+    return args.width;
+  });
diff --git a/src/client/workflowMutation.ts b/src/client/workflowMutation.ts
index f0f4a69e..445736b4 100644
--- a/src/client/workflowMutation.ts
+++ b/src/client/workflowMutation.ts
@@ -190,6 +190,20 @@ export function workflowMutation<
       );
       const inProgress = journalEntries.filter(({ step }) => step.inProgress);
       const console = createLogger(logLevel);
+      if (workflow.generationNumber !== generationNumber) {
+        console.error(
+          `Invalid generation number: ${generationNumber} running workflow ${workflow.name} (${workflowId})`,
+        );
+        return workflowId;
+      }
+      if (workflow.runResult !== undefined) {
+        // A duplicate run of a generation that already finished. Completing it
+        // again can never succeed, so don't replay the handler.
+        console.log(
+          `Workflow ${workflowId} already ${workflow.runResult.kind}, returning.`,
+        );
+        return workflowId;
+      }
       if (!ok) {
         console.error(`Failed to load journal for ${workflowId}`);
         await ctx.runMutation(component.workflow.complete, {
@@ -199,16 +213,6 @@ export function workflowMutation<
         });
         return workflowId;
       }
-      if (workflow.generationNumber !== generationNumber) {
-        console.error(
-          `Invalid generation number: ${generationNumber} running workflow ${workflow.name} (${workflowId})`,
-        );
-        return workflowId;
-      }
-      if (workflow.runResult?.kind === "success") {
-        console.log(`Workflow ${workflowId} completed, returning.`);
-        return workflowId;
-      }
       if (inProgress.length > 0) {
         console.log(
           `Workflow ${workflowId} blocked by ` +
diff --git a/src/component/journal.ts b/src/component/journal.ts
index aa48574c..8aeb3b25 100644
--- a/src/component/journal.ts
+++ b/src/component/journal.ts
@@ -40,6 +40,10 @@ export const load = query({
     const { logLevel } = await getDefaultLogger(ctx);
     const journalEntries: JournalEntry[] = [];
     let journalSize = 0;
+    if (shortCircuit && workflow.runResult !== undefined) {
+      // Nothing left to replay, so skip reading the journal.
+      return { journalEntries, workflow, logLevel, ok: true };
+    }
     if (shortCircuit) {
       const inProgress = await ctx.db
         .query("steps")
diff --git a/src/component/pool.ts b/src/component/pool.ts
index 06dc11b5..ab07ab10 100644
--- a/src/component/pool.ts
+++ b/src/component/pool.ts
@@ -245,6 +245,29 @@ export const handlerOnComplete = internalMutation({
       return;
     }
     const { workflowId, generationNumber } = args.context;
+    // The handler's mutation already rolled back. If its run can no longer be
+    // completed, this notification never can be either, so don't throw.
+    // A malformed id falls through, and fails loudly in completeHandler.
+    const id = ctx.db.normalizeId("workflows", workflowId);
+    const workflow = id && (await ctx.db.get("workflows", id));
+    if (id && !workflow) {
+      console.warn(
+        `Workflow ${workflowId} not found when its handler failed. It was likely cleaned up.`,
+      );
+      return;
+    }
+    if (workflow && workflow.generationNumber !== generationNumber) {
+      console.warn(
+        `Workflow ${workflowId} is at generation ${workflow.generationNumber} when its handler failed at generation ${generationNumber}`,
+      );
+      return;
+    }
+    if (workflow && workflow.runResult !== undefined) {
+      console.warn(
+        `Workflow ${workflowId} already ${workflow.runResult.kind} when its handler failed`,
+      );
+      return;
+    }
     await completeHandler(ctx, {
       workflowId,
       generationNumber,
diff --git a/src/component/workflow.test.ts b/src/component/workflow.test.ts
index 1515fdb3..5a93cd6a 100644
--- a/src/component/workflow.test.ts
+++ b/src/component/workflow.test.ts
@@ -1,12 +1,29 @@
 /// <reference types="vite/client" />
 
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
-import { api } from "./_generated/api.js";
+import { makeFunctionReference } from "convex/server";
+import { api, internal } from "./_generated/api.js";
 import { initConvexTest } from "./setup.test.js";
 import type { Id } from "./_generated/dataModel.js";
 import { internalMutation } from "./_generated/server.js";
 import { v } from "convex/values";
 import { enqueueWorkflow, getWorkpool, handlerOnComplete } from "./pool.js";
+import { workflowMutation } from "../client/workflowMutation.js";
+import type { WorkflowComponent } from "../client/types.js";
+import type { WorkId } from "@convex-dev/workpool";
+
+const completedStep = (name: string, args: Record<string, unknown> = {}) => ({
+  kind: "function" as const,
+  functionType: "mutation" as const,
+  handle: "function://;workflow.test:noop",
+  name,
+  inProgress: false,
+  argsSize: 0,
+  args,
+  runResult: { kind: "success" as const, returnValue: null },
+  startedAt: Date.now(),
+  completedAt: Date.now(),
+});
 
 describe("workflow", () => {
   beforeEach(async () => {
@@ -773,8 +790,190 @@ describe("workflow", () => {
   });
 });
 
+describe("stale handler runs and notifications", () => {
+  beforeEach(async () => {
+    vi.useFakeTimers();
+    handlerRuns.count = 0;
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+    vi.useRealTimers();
+  });
+
+  async function createWorkflow(
+    t: ReturnType<typeof initConvexTest>,
+    handler = "noop",
+  ) {
+    return await t.mutation(api.workflow.create, {
+      workflowName: "test",
+      workflowHandle: `function://;workflow.test:${handler}`,
+      workflowArgs: {},
+      createOnly: true,
+    });
+  }
+
+  const notify = (
+    t: ReturnType<typeof initConvexTest>,
+    workflowId: Id<"workflows">,
+    generationNumber: number,
+  ) =>
+    t.mutation(internal.pool.handlerOnComplete, {
+      workId: "work" as WorkId,
+      result: { kind: "failed", error: "handler failed" },
+      context: { workflowId, generationNumber },
+    });
+
+  test("handlerOnComplete fails a live workflow", async () => {
+    const t = initConvexTest();
+    const id = await createWorkflow(t);
+    await notify(t, id, 0);
+    const status = await t.query(api.workflow.getStatus, { workflowId: id });
+    expect(status.workflow.runResult).toEqual({
+      kind: "failed",
+      error: "handler failed",
+    });
+  });
+
+  test("handlerOnComplete ignores a workflow that already finished", async () => {
+    const t = initConvexTest();
+    const id = await createWorkflow(t);
+    const runResult = { kind: "failed" as const, error: "the real failure" };
+    await t.mutation(api.workflow.complete, {
+      workflowId: id,
+      generationNumber: 0,
+      runResult,
+    });
+    await notify(t, id, 0);
+    const status = await t.query(api.workflow.getStatus, { workflowId: id });
+    expect(status.workflow.runResult).toEqual(runResult);
+  });
+
+  test("handlerOnComplete ignores a restarted workflow", async () => {
+    const t = initConvexTest();
+    const id = await createWorkflow(t);
+    await t.mutation(api.workflow.complete, {
+      workflowId: id,
+      generationNumber: 0,
+      runResult: { kind: "failed", error: "previous attempt" },
+    });
+    await t.mutation(api.workflow.restart, { workflowId: id });
+    await notify(t, id, 0);
+    const status = await t.query(api.workflow.getStatus, { workflowId: id });
+    expect(status.workflow.generationNumber).toBe(1);
+    expect(status.workflow.runResult).toBeUndefined();
+  });
+
+  test("handlerOnComplete ignores a canceled workflow", async () => {
+    const t = initConvexTest();
+    const id = await createWorkflow(t);
+    await t.mutation(api.workflow.cancel, { workflowId: id });
+    await notify(t, id, 0);
+    const status = await t.query(api.workflow.getStatus, { workflowId: id });
+    expect(status.workflow.runResult).toEqual({ kind: "canceled" });
+  });
+
+  test("handlerOnComplete ignores a cleaned up workflow", async () => {
+    const t = initConvexTest();
+    const id = await createWorkflow(t);
+    await t.mutation(api.workflow.cleanup, { workflowId: id, force: true });
+    await notify(t, id, 0);
+    await t.run(async (ctx) => {
+      expect(await ctx.db.get("workflows", id)).toBeNull();
+    });
+  });
+
+  test("journal.load skips the journal of a finished workflow", async () => {
+    const t = initConvexTest();
+    const id = await createWorkflow(t);
+    // More than the journal size limit, so reading it at all is detectable.
+    await t.run(async (ctx) => {
+      const blob = "x".repeat(900_000);
+      for (let i = 0; i < 10; i++) {
+        await ctx.db.insert("steps", {
+          workflowId: id,
+          stepNumber: i,
+          step: completedStep(`step${i}`, { blob }),
+        });
+      }
+    });
+    const running = await t.query(api.journal.load, {
+      workflowId: id,
+      shortCircuit: true,
+    });
+    expect(running.ok).toBe(false);
+
+    await t.mutation(api.workflow.complete, {
+      workflowId: id,
+      generationNumber: 0,
+      runResult: { kind: "failed", error: "boom" },
+    });
+    const finished = await t.query(api.journal.load, {
+      workflowId: id,
+      shortCircuit: true,
+    });
+    expect(finished).toMatchObject({ ok: true, journalEntries: [] });
+    expect(finished.workflow.runResult).toMatchObject({ kind: "failed" });
+    // Without the flag it still reads the journal.
+    const full = await t.query(api.journal.load, { workflowId: id });
+    expect(full.ok).toBe(false);
+    expect(full.journalEntries.length).toBeGreaterThan(0);
+  });
+
+  test("a stale run does not fail on an oversized journal", async () => {
+    const t = initConvexTest();
+    const id = await createWorkflow(t, "throwingWorkflow");
+    await t.run(async (ctx) => {
+      // The workflow has since been restarted.
+      await ctx.db.patch("workflows", id, { generationNumber: 1 });
+      const blob = "x".repeat(900_000);
+      for (let i = 0; i < 10; i++) {
+        await ctx.db.insert("steps", {
+          workflowId: id,
+          stepNumber: i,
+          step: completedStep(`step${i}`, { blob }),
+        });
+      }
+    });
+    const loaded = await t.query(api.journal.load, { workflowId: id });
+    expect(loaded.ok).toBe(false);
+
+    const run = (generationNumber: number) =>
+      t.mutation(
+        makeFunctionReference<"mutation">("workflow.test:throwingWorkflow"),
+        { workflowId: id, generationNumber },
+      );
+    // A run left over from the previous generation leaves the workflow alone.
+    await run(0);
+    const stale = await t.query(api.workflow.getStatus, { workflowId: id });
+    expect(stale.workflow.runResult).toBeUndefined();
+
+    // The current generation still fails on the oversized journal.
+    await run(1);
+    const current = await t.query(api.workflow.getStatus, { workflowId: id });
+    expect(current.workflow.runResult).toEqual({
+      kind: "failed",
+      error: "Failed to load journal",
+    });
+    expect(handlerRuns.count).toBe(0);
+  });
+});
+
 export const noop = internalMutation({ args: v.any(), handler: () => {} });
 
+const handlerRuns = { count: 0 };
+
+// A real workflow mutation, running against this component's own API.
+export const throwingWorkflow = workflowMutation(
+  api as unknown as WorkflowComponent,
+  {
+    handler: async () => {
+      handlerRuns.count++;
+      throw new Error("boom");
+    },
+  },
+);
+
 export const failingHandler = internalMutation({
   args: { workflowId: v.id("workflows"), generationNumber: v.number() },
   returns: v.null(),
diff --git a/src/component/workflow.ts b/src/component/workflow.ts
index 8e181448..3b0adc71 100644
--- a/src/component/workflow.ts
+++ b/src/component/workflow.ts
@@ -386,7 +386,7 @@ export async function completeHandler(
   );
   const console = await getDefaultLogger(ctx);
   if (workflow.runResult) {
-    throw new Error(`Workflow not running: ${workflow}`);
+    throw new Error(`Workflow not running: ${workflow._id}`);
   }
   workflow.runResult = args.runResult;
   console.event("completed", {