import { afterEach, describe, expect, it, vi } from "vitest";
import { createResumableStreamContext } from "./ResumableStreamContext";
import { ResumableStreamError } from "./errors";
import { createInMemoryResumableStreamStore } from "./stores/InMemoryResumableStreamStore";

import type { ResumableStreamStore } from "./types";

const enc = new TextEncoder();
const dec = new TextDecoder();

const bytes = (s: string): Uint8Array => enc.encode(s);

afterEach(() => {
  vi.restoreAllMocks();
});

async function collect(stream: ReadableStream<Uint8Array>): Promise<string> {
  const reader = stream.getReader();
  let out = "";
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) return out;
      out += dec.decode(value, { stream: true });
    }
  } finally {
    reader.releaseLock();
  }
}

function makeStringStream(parts: string[]): ReadableStream<Uint8Array> {
  return new ReadableStream<Uint8Array>({
    async start(controller) {
      for (const part of parts) {
        controller.enqueue(bytes(part));
        await Promise.resolve();
      }
      controller.close();
    },
  });
}

describe("createResumableStreamContext", () => {
  it.each(["done", "error"] as const)(
    "passes the acquisition lease to append and %s finalize",
    async (status) => {
      const lease = { token: "producer-token" };
      const append = vi.fn<ResumableStreamStore["append"]>(async () => {
        if (status === "error") throw new Error("append failed");
      });
      const finalize = vi.fn<ResumableStreamStore["finalize"]>(async () => {});
      const acquire = vi.fn<ResumableStreamStore["acquire"]>();
      const acquireLease = vi.fn<
        NonNullable<ResumableStreamStore["acquireLease"]>
      >(async () => ({ role: "producer", lease }));
      const store: ResumableStreamStore = {
        acquire,
        acquireLease,
        append,
        finalize,
        async *read() {},
        async status() {
          return "streaming";
        },
        async delete() {},
      };
      const tasks: Promise<unknown>[] = [];
      const ctx = createResumableStreamContext({
        store,
        ttlMs: 123,
        waitUntil: (task) => tasks.push(task),
      });
      await ctx.run("a", () => makeStringStream(["x"]));
      await Promise.all(tasks);
      expect(acquire).not.toHaveBeenCalled();
      expect(acquireLease).toHaveBeenCalledWith("a", { ttlMs: 123 });
      expect(append).toHaveBeenCalledWith("a", bytes("x"), lease);
      expect(finalize).toHaveBeenCalledWith(
        "a",
        status,
        status === "error" ? "append failed" : undefined,
        lease,
      );
    },
  );

  it("runs a producer through acquire when acquireLease is absent", async () => {
    const backing = createInMemoryResumableStreamStore();
    const store: ResumableStreamStore = {
      acquire: vi.fn(backing.acquire),
      append: backing.append,
      finalize: backing.finalize,
      read: backing.read,
      status: backing.status,
      delete: backing.delete,
    };
    const ctx = createResumableStreamContext({ store });
    expect(
      await collect(await ctx.run("a", () => makeStringStream(["legacy"]))),
    ).toBe("legacy");
    expect(store.acquire).toHaveBeenCalledWith("a", undefined);
  });
  it("keeps a producer that outlived its TTL out of the stream a second run reacquired", async () => {
    let now = 0;
    const store = createInMemoryResumableStreamStore({
      now: () => now,
      defaultTtlMs: 10,
    });
    const errors: unknown[] = [];
    const ctx = createResumableStreamContext({
      store,
      onError: (_id, err) => errors.push(err),
    });
    let releaseStale!: () => void;
    const stale = new ReadableStream<Uint8Array>({
      async start(controller) {
        controller.enqueue(bytes("old"));
        await new Promise<void>((resolve) => (releaseStale = resolve));
        controller.enqueue(bytes("late"));
        controller.close();
      },
    });
    await ctx.run("a", () => stale);
    await vi.waitFor(() => expect(releaseStale).toBeDefined());

    now = 11;
    const fresh = await ctx.run("a", () => makeStringStream(["new"]));
    releaseStale();
    expect(await collect(fresh)).toBe("new");
    await vi.waitFor(() => expect(errors).toHaveLength(1));
    expect(errors[0]).toMatchObject({ code: "missing" });
  });

  it("producer caller receives full byte stream", async () => {
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
    });
    const stream = await ctx.run("a", () =>
      makeStringStream(["hello ", "world"]),
    );
    expect(await collect(stream)).toBe("hello world");
  });

  it("second caller becomes consumer and receives identical bytes", async () => {
    const store = createInMemoryResumableStreamStore();
    const ctx = createResumableStreamContext({ store });

    const producerStream = await ctx.run("a", () =>
      makeStringStream(["one ", "two ", "three"]),
    );
    const consumerStream = await ctx.run("a", () =>
      makeStringStream(["should-not-run"]),
    );

    const [a, b] = await Promise.all([
      collect(producerStream),
      collect(consumerStream),
    ]);
    expect(a).toBe("one two three");
    expect(b).toBe("one two three");
  });

  it("late consumer after done replays via resume", async () => {
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
    });
    const producer = await ctx.run("a", () =>
      makeStringStream(["alpha", "beta", "gamma"]),
    );
    expect(await collect(producer)).toBe("alphabetagamma");

    const replay = await ctx.resume("a");
    expect(replay).not.toBeNull();
    expect(await collect(replay!)).toBe("alphabetagamma");
  });

  it("resume returns null for missing streams", async () => {
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
    });
    expect(await ctx.resume("nope")).toBeNull();
  });

  it("requireResume throws ResumableStreamError for missing streams", async () => {
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
    });
    await expect(ctx.requireResume("nope")).rejects.toBeInstanceOf(
      ResumableStreamError,
    );
    await expect(ctx.requireResume("nope")).rejects.toMatchObject({
      code: "missing",
    });
  });

  it("requireResume returns the replay stream when it exists", async () => {
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
    });
    const producer = await ctx.run("a", () => makeStringStream(["hi"]));
    expect(await collect(producer)).toBe("hi");

    const replay = await ctx.requireResume("a");
    expect(await collect(replay)).toBe("hi");
  });

  it("status tracks lifecycle", async () => {
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
    });
    expect(await ctx.status("a")).toBe("missing");
    const stream = await ctx.run("a", () => makeStringStream(["x"]));
    await collect(stream);
    expect(await ctx.status("a")).toBe("done");
  });

  it("delete removes stream state", async () => {
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
    });
    const stream = await ctx.run("a", () => makeStringStream(["x"]));
    await collect(stream);
    expect(await ctx.status("a")).toBe("done");
    await ctx.delete("a");
    expect(await ctx.status("a")).toBe("missing");
  });

  it("producer keeps writing after the local consumer cancels", async () => {
    const store = createInMemoryResumableStreamStore();
    const ctx = createResumableStreamContext({ store });

    let producerEmitted = 0;
    const slowStream = new ReadableStream<Uint8Array>({
      async start(controller) {
        for (let i = 0; i < 5; i++) {
          await new Promise((r) => setTimeout(r, 5));
          controller.enqueue(bytes(`chunk${i};`));
          producerEmitted += 1;
        }
        controller.close();
      },
    });

    const stream = await ctx.run("a", () => slowStream);
    const reader = stream.getReader();
    const first = await reader.read();
    expect(first.done).toBe(false);
    await reader.cancel();

    while (producerEmitted < 5) {
      await new Promise((r) => setTimeout(r, 5));
    }
    while ((await ctx.status("a")) === "streaming") {
      await new Promise((r) => setTimeout(r, 5));
    }

    const replay = await ctx.resume("a");
    expect(replay).not.toBeNull();
    expect(await collect(replay!)).toBe("chunk0;chunk1;chunk2;chunk3;chunk4;");
  });

  it("waits for store iterator cleanup when a consumer cancels", async () => {
    let releaseCleanup!: () => void;
    const cleanupGate = new Promise<void>((resolve) => {
      releaseCleanup = resolve;
    });
    const iterator = {
      next: vi.fn(
        () =>
          new Promise<IteratorResult<{ cursor: string; chunk: Uint8Array }>>(
            () => undefined,
          ),
      ),
      return: vi.fn(async () => {
        await cleanupGate;
        return { done: true as const, value: undefined };
      }),
    };
    const ctx = createResumableStreamContext({
      store: {
        async acquire() {
          return "consumer";
        },
        async append() {},
        async finalize() {},
        read() {
          return {
            [Symbol.asyncIterator]: () => iterator,
          };
        },
        async status() {
          return "streaming";
        },
        async delete() {},
      },
    });

    const stream = await ctx.run("a", () => {
      throw new Error("consumer must not create a producer stream");
    });
    let cancelSettled = false;
    const cancelPromise = stream.cancel().then(() => {
      cancelSettled = true;
    });

    await vi.waitFor(() => expect(iterator.return).toHaveBeenCalledOnce());
    expect(cancelSettled).toBe(false);

    releaseCleanup();
    await cancelPromise;
    expect(cancelSettled).toBe(true);
  });

  it("cancels a parked in-memory store reader", async () => {
    const store = createInMemoryResumableStreamStore();
    await store.acquire("a");
    const read = store.read.bind(store);
    let markReadStarted!: () => void;
    const readStarted = new Promise<void>((resolve) => {
      markReadStarted = resolve;
    });
    vi.spyOn(store, "read").mockImplementation(async function* (...args) {
      markReadStarted();
      yield* read(...args);
    });
    const ctx = createResumableStreamContext({ store });

    const stream = await ctx.run("a", () => {
      throw new Error("consumer must not create a producer stream");
    });
    const reader = stream.getReader();
    const readPromise = reader.read();
    await readStarted;

    await expect(reader.cancel()).resolves.toBeUndefined();
    await expect(readPromise).resolves.toEqual({
      done: true,
      value: undefined,
    });
  });

  it("propagates producer errors and releases the source reader", async () => {
    const tasks: Promise<unknown>[] = [];
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
      waitUntil: (task) => tasks.push(task),
    });
    const failing = new ReadableStream<Uint8Array>({
      start(controller) {
        controller.enqueue(bytes("partial;"));
        controller.error(new Error("oops"));
      },
    });
    const stream = await ctx.run("a", () => failing);
    await expect(collect(stream)).rejects.toThrow("oops");
    await Promise.all(tasks);
    expect(await ctx.status("a")).toBe("error");
    expect(failing.locked).toBe(false);
  });

  it.each(["resolve", "reject"] as const)(
    "publishes append errors before source cancellation settles with %s",
    async (cleanupOutcome) => {
      const cleanupError = new Error("cleanup failed");
      const consoleError = vi
        .spyOn(console, "error")
        .mockImplementation(() => {});
      let finishCleanup!: () => void;
      const cleanup = new Promise<void>((resolve, reject) => {
        finishCleanup = () =>
          cleanupOutcome === "resolve" ? resolve() : reject(cleanupError);
      });
      let markCancelStarted!: () => void;
      const cancelStarted = new Promise<void>((resolve) => {
        markCancelStarted = resolve;
      });
      const cancel = vi.fn(() => {
        markCancelStarted();
        return cleanup;
      });
      const source = new ReadableStream<Uint8Array>({
        start(controller) {
          controller.enqueue(bytes("ab"));
        },
        cancel,
      });
      const store = createInMemoryResumableStreamStore({ maxChunkBytes: 1 });
      const onError = vi.fn();
      const onFinalize = vi.fn();
      let task!: Promise<unknown>;
      let taskSettled = false;
      const ctx = createResumableStreamContext({
        store,
        onError,
        onFinalize,
        waitUntil: (value) => {
          task = value.then(() => {
            taskSettled = true;
          });
        },
      });
      const response = await ctx.run("a", () => source);
      const result = collect(response).catch((error: unknown) => error);

      try {
        await cancelStarted;
        expect(await ctx.status("a")).toBe("error");
        expect(await result).toEqual(
          new Error("Chunk exceeds maxChunkBytes: 2"),
        );
        expect(onError).toHaveBeenCalledExactlyOnceWith(
          "a",
          new Error("Chunk exceeds maxChunkBytes: 2"),
        );
        expect(onFinalize).toHaveBeenCalledExactlyOnceWith(
          "a",
          "error",
          "Chunk exceeds maxChunkBytes: 2",
        );
        expect(taskSettled).toBe(false);
        expect(source.locked).toBe(true);
      } finally {
        finishCleanup();
        await task;
        await result;
        store.dispose();
      }

      expect(taskSettled).toBe(true);
      expect(source.locked).toBe(false);
      expect(cancel).toHaveBeenCalledExactlyOnceWith(
        new Error("Chunk exceeds maxChunkBytes: 2"),
      );
      if (cleanupOutcome === "reject") {
        expect(consoleError).toHaveBeenCalledExactlyOnceWith(
          "resumable stream reader cancel failed:",
          cleanupError,
        );
      } else {
        expect(consoleError).not.toHaveBeenCalled();
      }
    },
  );

  it.each(["fenced", "failed"] as const)(
    "awaits source cleanup when error finalization is %s",
    async (finalizeOutcome) => {
      const finalizeError = new Error("finalize failed");
      const consoleError = vi
        .spyOn(console, "error")
        .mockImplementation(() => {});
      let finishCleanup!: () => void;
      const cleanup = new Promise<void>((resolve) => {
        finishCleanup = resolve;
      });
      const source = new ReadableStream<Uint8Array>({
        start(controller) {
          controller.enqueue(bytes("ab"));
        },
        cancel: () => cleanup,
      });
      const store = createInMemoryResumableStreamStore({ maxChunkBytes: 1 });
      const finalize = vi
        .spyOn(store, "finalize")
        .mockImplementation(async () => {
          if (finalizeOutcome === "failed") throw finalizeError;
          return false;
        });
      const onFinalize = vi.fn();
      let task!: Promise<unknown>;
      let taskSettled = false;
      const ctx = createResumableStreamContext({
        store,
        onFinalize,
        waitUntil: (value) => {
          task = value.then(() => {
            taskSettled = true;
          });
        },
      });
      const response = await ctx.run("a", () => source);

      try {
        await vi.waitFor(() => expect(finalize).toHaveBeenCalledOnce());
        expect(taskSettled).toBe(false);
        expect(source.locked).toBe(true);
        expect(onFinalize).not.toHaveBeenCalled();
      } finally {
        finishCleanup();
        await task;
        await response.cancel();
        store.dispose();
      }

      expect(taskSettled).toBe(true);
      expect(source.locked).toBe(false);
      expect(onFinalize).not.toHaveBeenCalled();
      if (finalizeOutcome === "failed") {
        expect(consoleError).toHaveBeenCalledExactlyOnceWith(
          "resumable stream finalize failed:",
          finalizeError,
        );
      } else {
        expect(consoleError).not.toHaveBeenCalled();
      }
    },
  );

  it("waitUntil receives the producer task promise", async () => {
    const promises: Promise<unknown>[] = [];
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
      waitUntil: (p) => promises.push(p),
    });
    const stream = await ctx.run("a", () => makeStringStream(["x"]));
    expect(await collect(stream)).toBe("x");
    expect(promises.length).toBe(1);
    await Promise.all(promises);
  });

  it("onAcquire fires for both producer and consumer roles", async () => {
    const calls: Array<{ id: string; role: string }> = [];
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
      onAcquire: (id, role) => calls.push({ id, role }),
    });
    const producer = await ctx.run("a", () => makeStringStream(["x"]));
    const consumer = await ctx.run("a", () => makeStringStream(["unused"]));
    await Promise.all([collect(producer), collect(consumer)]);
    expect(calls).toEqual([
      { id: "a", role: "producer" },
      { id: "a", role: "consumer" },
    ]);
  });

  it("onAppend fires per appended chunk with byteLength", async () => {
    const calls: Array<{ id: string; byteLength: number }> = [];
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
      onAppend: (id, byteLength) => calls.push({ id, byteLength }),
    });
    const stream = await ctx.run("a", () => makeStringStream(["ab", "cde"]));
    expect(await collect(stream)).toBe("abcde");
    expect(calls).toEqual([
      { id: "a", byteLength: 2 },
      { id: "a", byteLength: 3 },
    ]);
  });

  it("onFinalize fires on successful completion", async () => {
    const calls: Array<{
      id: string;
      status: "done" | "error";
      error: string | undefined;
    }> = [];
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
      onFinalize: (id, status, error) => calls.push({ id, status, error }),
    });
    const stream = await ctx.run("a", () => makeStringStream(["x"]));
    await collect(stream);
    expect(calls).toEqual([{ id: "a", status: "done", error: undefined }]);
  });

  it("onFinalize fires with error status when producer fails", async () => {
    const calls: Array<{
      id: string;
      status: "done" | "error";
      error: string | undefined;
    }> = [];
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
      onFinalize: (id, status, error) => calls.push({ id, status, error }),
    });
    const failing = new ReadableStream<Uint8Array>({
      start(controller) {
        controller.error(new Error("boom"));
      },
    });
    const stream = await ctx.run("a", () => failing);
    await expect(collect(stream)).rejects.toThrow("boom");
    expect(calls).toEqual([{ id: "a", status: "error", error: "boom" }]);
  });

  describe("a producer fenced out by a reacquisition", () => {
    type Ending = "chunk" | "close" | "error";

    const gatedStream = (first: string, ending: Ending) => {
      let release!: () => void;
      const stream = new ReadableStream<Uint8Array>({
        async start(controller) {
          controller.enqueue(bytes(first));
          await new Promise<void>((resolve) => (release = resolve));
          if (ending === "chunk") {
            controller.enqueue(bytes("late"));
            controller.close();
          } else if (ending === "close") {
            controller.close();
          } else {
            controller.error(new Error("boom"));
          }
        },
      });
      return { stream, release: () => release() };
    };

    const runStaleProducer = async (
      ending: Ending,
      wrap: (store: ResumableStreamStore) => ResumableStreamStore = (s) => s,
    ) => {
      let now = 0;
      const store = wrap(
        createInMemoryResumableStreamStore({
          now: () => now,
          defaultTtlMs: 10,
        }),
      );
      const errors: unknown[] = [];
      const finalizes: unknown[][] = [];
      const tasks: Promise<unknown>[] = [];
      const ctx = createResumableStreamContext({
        store,
        waitUntil: (task) => tasks.push(task),
        onError: (_id, error) => errors.push(error),
        onFinalize: (...args) => finalizes.push(args),
      });
      const stale = gatedStream("old", ending);
      await ctx.run("a", () => stale.stream);
      await vi.waitFor(() => expect(tasks).toHaveLength(1));
      now = 11;
      const fresh = gatedStream("new", "close");
      await ctx.run("a", () => fresh.stream);
      await vi.waitFor(() => expect(tasks).toHaveLength(2));
      stale.release();
      await tasks[0];
      const observed = {
        errors,
        finalizes: [...finalizes],
        status: await ctx.status("a"),
      };
      fresh.release();
      await tasks[1];
      return observed;
    };

    it("reports the superseded append through onError only", async () => {
      const { errors, finalizes, status } = await runStaleProducer("chunk");
      expect(errors).toEqual([
        expect.objectContaining({
          code: "missing",
          message: "Stream superseded by a new acquisition: a",
        }),
      ]);
      expect(finalizes).toEqual([]);
      expect(status).toBe("streaming");
    });

    it("reports a fenced done finalize through onError only", async () => {
      const { errors, finalizes, status } = await runStaleProducer("close");
      expect(errors).toEqual([
        expect.objectContaining({
          code: "missing",
          message: "Stream no longer owned by this producer: a",
        }),
      ]);
      expect(errors[0]).toBeInstanceOf(ResumableStreamError);
      expect(finalizes).toEqual([]);
      expect(status).toBe("streaming");
    });

    it("does not report a fenced error finalize", async () => {
      const { errors, finalizes, status } = await runStaleProducer("error");
      expect(errors).toEqual([expect.objectContaining({ message: "boom" })]);
      expect(finalizes).toEqual([]);
      expect(status).toBe("streaming");
    });

    it("needs no particular error message from the store", async () => {
      const { errors, finalizes, status } = await runStaleProducer(
        "chunk",
        (inner) => ({
          ...inner,
          acquireLease: (id, options) => inner.acquireLease!(id, options),
          async append(id, chunk, lease) {
            try {
              await inner.append(id, chunk, lease);
            } catch (error) {
              if (
                error instanceof ResumableStreamError &&
                error.code === "missing"
              ) {
                throw new ResumableStreamError("missing", `lost ${id}`);
              }
              throw error;
            }
          },
        }),
      );
      expect(errors).toEqual([expect.objectContaining({ message: "lost a" })]);
      expect(finalizes).toEqual([]);
      expect(status).toBe("streaming");
    });
  });

  it("treats a finalize that resolves without a value as finalized", async () => {
    const calls: unknown[][] = [];
    const backing = createInMemoryResumableStreamStore();
    const store: ResumableStreamStore = {
      acquire: backing.acquire,
      append: backing.append,
      async finalize(id, status, error) {
        await backing.finalize(id, status, error);
      },
      read: backing.read,
      status: backing.status,
      delete: backing.delete,
    };
    const ctx = createResumableStreamContext({
      store,
      onFinalize: (...args) => calls.push(args),
    });
    await collect(await ctx.run("a", () => makeStringStream(["x"])));
    expect(calls).toEqual([["a", "done"]]);
  });

  it("onError fires when the producer task throws", async () => {
    const errors: Array<{ id: string; error: unknown }> = [];
    const ctx = createResumableStreamContext({
      store: createInMemoryResumableStreamStore(),
      onError: (id, error) => errors.push({ id, error }),
    });
    const failing = new ReadableStream<Uint8Array>({
      start(controller) {
        controller.error(new Error("boom"));
      },
    });
    const stream = await ctx.run("a", () => failing);
    await expect(collect(stream)).rejects.toThrow("boom");
    expect(errors).toHaveLength(1);
    expect(errors[0]!.id).toBe("a");
    expect((errors[0]!.error as Error).message).toBe("boom");
  });

  it("continues acquisition when onAcquire throws", async () => {
    const consoleError = vi
      .spyOn(console, "error")
      .mockImplementation(() => {});
    const hookError = new Error("acquire observer failed");
    const store = createInMemoryResumableStreamStore();
    const ctx = createResumableStreamContext({
      store,
      onAcquire: () => {
        throw hookError;
      },
    });

    const stream = await ctx.run("a", () => makeStringStream(["hello"]));

    expect(await collect(stream)).toBe("hello");
    expect(await ctx.status("a")).toBe("done");
    expect(consoleError).toHaveBeenCalledWith(
      "resumable stream onAcquire hook failed:",
      hookError,
    );
  });

  it("continues streaming when onAppend throws", async () => {
    vi.spyOn(console, "error").mockImplementation(() => {});
    const store = createInMemoryResumableStreamStore();
    const ctx = createResumableStreamContext({
      store,
      onAppend: () => {
        throw new Error("append observer failed");
      },
    });

    const stream = await ctx.run("a", () =>
      makeStringStream(["hello ", "world"]),
    );

    expect(await collect(stream)).toBe("hello world");
    expect(await ctx.status("a")).toBe("done");
  });

  it("continues streaming when async onAppend rejects", async () => {
    const consoleError = vi
      .spyOn(console, "error")
      .mockImplementation(() => {});
    const hookError = new Error("async append observer failed");
    const store = createInMemoryResumableStreamStore();
    const ctx = createResumableStreamContext({
      store,
      onAppend: async () => {
        throw hookError;
      },
    });

    const stream = await ctx.run("a", () => makeStringStream(["hello"]));

    expect(await collect(stream)).toBe("hello");
    expect(await ctx.status("a")).toBe("done");
    await vi.waitFor(() => {
      expect(consoleError).toHaveBeenCalledWith(
        "resumable stream onAppend hook failed:",
        hookError,
      );
    });
  });

  it("keeps successful completion when onFinalize throws", async () => {
    vi.spyOn(console, "error").mockImplementation(() => {});
    const onError = vi.fn();
    const onFinalize = vi.fn(() => {
      throw new Error("finalize observer failed");
    });
    const store = createInMemoryResumableStreamStore();
    const ctx = createResumableStreamContext({ store, onError, onFinalize });

    const stream = await ctx.run("a", () => makeStringStream(["hello"]));

    expect(await collect(stream)).toBe("hello");
    expect(await ctx.status("a")).toBe("done");
    expect(onFinalize).toHaveBeenCalledOnce();
    expect(onError).not.toHaveBeenCalled();
  });

  it("finalizes producer errors when onError throws", async () => {
    vi.spyOn(console, "error").mockImplementation(() => {});
    const tasks: Promise<unknown>[] = [];
    const store = createInMemoryResumableStreamStore();
    const ctx = createResumableStreamContext({
      store,
      waitUntil: (task) => tasks.push(task),
      onError: () => {
        throw new Error("error observer failed");
      },
    });
    const producerError = new Error("producer failed");

    const stream = await ctx.run(
      "a",
      () =>
        new ReadableStream<Uint8Array>({
          start(controller) {
            controller.error(producerError);
          },
        }),
    );
    await Promise.allSettled(tasks);

    expect(await ctx.status("a")).toBe("error");
    await expect(collect(stream)).rejects.toThrow("producer failed");
  });
});