import { describe, expect, it } from "vitest";

import type {
  OpfsPreparedObject,
  OpfsPhysicalGenerationId,
  OpfsStorageScope,
} from "../../src/application/ports/browser-file-storage/opfs-ports.ts";
import type {
  BrowserStoragePolicy,
} from "../../src/application/ports/browser-file-storage/shared.ts";
import {
  resolveOpfsRuntimePolicy,
} from "../../src/adapters/storage/opfs/opfs-policy.ts";
import {
  createOpfsWorkerGateway,
  type OpfsWorkerLike,
} from "../../src/adapters/storage/opfs/opfs-worker-client.ts";
import { OPFS_WORKER_PROTOCOL_VERSION } from "../../src/adapters/storage/opfs/opfs-worker-protocol.ts";
import type {
  OpfsWorkerRequest,
  OpfsWorkerResponse,
} from "../../src/adapters/storage/opfs/opfs-worker-protocol.ts";
import {
  createOpfsWorkerRuntime,
  startBrowserOpfsDedicatedWorker,
  type OpfsMutationLeaseManager,
  type OpfsWorkerMessageHost,
} from "../../src/adapters/storage/opfs/opfs-worker-runtime.ts";

const scopeA: OpfsStorageScope = Object.freeze({
  namespace: "durable-objects",
  authorityToken: "authority_12345678",
  namespaceToken: "namespace_12345678",
  partitionToken: "partition_12345678",
});
const scopeB: OpfsStorageScope = Object.freeze({
  ...scopeA,
  authorityToken: "authority_87654321",
});
const storagePolicy: BrowserStoragePolicy = Object.freeze({
  owner: "test-owner",
  namespace: scopeA.namespace,
  classification: "PERSONAL",
  authority: "LOCAL_FIRST",
  accountScope: "OPAQUE_PARTITION",
  retention: Object.freeze({ kind: "EXPLICIT_DELETE" }),
  softBudgetBytes: 1024 * 1024,
  hardBudgetBytes: 2 * 1024 * 1024,
  evictionPriority: "USER_AUTHORED",
  logoutAction: "EXPORT_THEN_PURGE",
  accountDeletionAction: "PURGE_PARTITION",
  pressureAction: "RETAIN",
  unavailableFallback: "EXPORT_REQUIRED",
});
const runtimePolicy = resolveOpfsRuntimePolicy({
  chunkSizeBytes: 64 * 1024,
  maxObjectBytes: 64 * 1024,
  maxChunkCount: 1,
  orphanGracePeriodMs: 60_000,
});

type MemoryNode = MemoryDirectory | MemoryFile;

class MemoryFile {
  readonly kind = "file";
  bytes = new Uint8Array();
  lastModified = 0;
}

class MemoryDirectory {
  readonly kind = "directory";
  readonly children = new Map<string, MemoryNode>();

  async getDirectoryHandle(
    name: string,
    options: FileSystemGetDirectoryOptions = {},
  ): Promise<FileSystemDirectoryHandle> {
    const current = this.children.get(name);
    if (current instanceof MemoryDirectory) {
      return current as unknown as FileSystemDirectoryHandle;
    }
    if (current || !options.create) throw notFound();
    const created = new MemoryDirectory();
    this.children.set(name, created);
    return created as unknown as FileSystemDirectoryHandle;
  }

  async getFileHandle(
    name: string,
    options: FileSystemGetFileOptions = {},
  ): Promise<FileSystemFileHandle> {
    const current = this.children.get(name);
    if (current instanceof MemoryFile) {
      return fileHandle(current);
    }
    if (current || !options.create) throw notFound();
    const created = new MemoryFile();
    this.children.set(name, created);
    return fileHandle(created);
  }

  async removeEntry(
    name: string,
    options: FileSystemRemoveOptions = {},
  ): Promise<void> {
    const current = this.children.get(name);
    if (!current) throw notFound();
    if (
      current instanceof MemoryDirectory &&
      current.children.size > 0 &&
      !options.recursive
    ) {
      throw new DOMException("Directory is not empty.", "InvalidModificationError");
    }
    this.children.delete(name);
  }

  async *entries(): AsyncIterableIterator<
    [string, FileSystemDirectoryHandle | FileSystemFileHandle]
  > {
    for (const [name, node] of this.children) {
      yield [
        name,
        node instanceof MemoryDirectory
          ? (node as unknown as FileSystemDirectoryHandle)
          : fileHandle(node),
      ];
    }
  }

  has(path: readonly string[]): boolean {
    let current: MemoryNode = this;
    for (const segment of path) {
      if (!(current instanceof MemoryDirectory)) return false;
      const next = current.children.get(segment);
      if (!next) return false;
      current = next;
    }
    return true;
  }
}

function fileHandle(file: MemoryFile): FileSystemFileHandle {
  return {
    kind: "file",
    name: "memory",
    async getFile() {
      const blob = new Blob([Uint8Array.from(file.bytes)]);
      Object.defineProperty(blob, "lastModified", {
        value: file.lastModified,
      });
      return blob as File;
    },
    async createWritable() {
      let pending = Uint8Array.from(file.bytes);
      return {
        async write(data: FileSystemWriteChunkType) {
          if (!(data instanceof Uint8Array)) {
            throw new TypeError("The test writer accepts Uint8Array only.");
          }
          pending = Uint8Array.from(data);
        },
        async close() {
          file.bytes = pending;
          file.lastModified = Date.now();
        },
        async abort() {},
      } as unknown as FileSystemWritableFileStream;
    },
  } as FileSystemFileHandle;
}

function notFound(): DOMException {
  return new DOMException("Entry was not found.", "NotFoundError");
}

const PHYSICAL_GENERATION_A = "a".repeat(32) as OpfsPhysicalGenerationId;

function beginRequest(
  requestId: string,
  transactionId: string,
  scope: OpfsStorageScope,
  physicalGenerationId: OpfsPhysicalGenerationId = PHYSICAL_GENERATION_A,
): OpfsWorkerRequest {
  return {
    requestId,
    protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
    kind: "BEGIN_PUT",
    transactionId,
    scope,
    objectId: "object_12345678",
    generation: 1,
    physicalGenerationId,
    declaredByteLength: 1,
    mediaType: "application/octet-stream",
    createdAtEpochMs: 100,
    storagePolicy,
    chunkSizeBytes: runtimePolicy.chunkSizeBytes,
  };
}

function immediateLeases(releases: { count: number }): OpfsMutationLeaseManager {
  return {
    async acquire() {
      let released = false;
      return {
        release() {
          if (released) return;
          released = true;
          releases.count += 1;
        },
      };
    },
  };
}

function expectFailureCode(
  response: OpfsWorkerResponse | null,
  code: string,
): void {
  expect(response).toMatchObject({ ok: false, failure: { code } });
}

function preparedValue(
  response: OpfsWorkerResponse | null,
): OpfsPreparedObject {
  if (
    !response?.ok ||
    !("value" in response) ||
    !response.value ||
    typeof response.value !== "object" ||
    response.value instanceof ArrayBuffer ||
    !("descriptor" in response.value)
  ) {
    throw new Error("Expected a prepared OPFS object.");
  }
  return response.value;
}

describe("OPFS dedicated worker runtime", () => {
  it("cancels an exact BEGIN while waiting for a lock and releases a late lease", async () => {
    const root = new MemoryDirectory();
    let resolveLease:
      | ((lease: { release(): void }) => void)
      | undefined;
    let acquireSignal: AbortSignal | undefined;
    let releases = 0;
    const leaseManager: OpfsMutationLeaseManager = {
      acquire(signal) {
        acquireSignal = signal;
        return new Promise((resolve) => {
          resolveLease = resolve;
        });
      },
    };
    const runtime = createOpfsWorkerRuntime({
      root: root as unknown as FileSystemDirectoryHandle,
      crypto: globalThis.crypto,
      policy: runtimePolicy,
      leaseManager,
      dedicatedWorker: true,
      supportsSynchronousAccessHandles: false,
    });

    const beginning = runtime.handleRequest(
      beginRequest(
        "request_begin_1234",
        "transaction_12345678",
        scopeA,
      ),
    );
    await Promise.resolve();
    const aborted = await runtime.handleRequest({
      requestId: "request_abort_1234",
      protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
      kind: "ABORT_PUT",
      scope: scopeA,
      transactionId: "transaction_12345678",
    });
    expect(aborted).toMatchObject({ ok: true });
    expect(acquireSignal?.aborted).toBe(true);

    resolveLease?.({
      release() {
        releases += 1;
      },
    });
    expectFailureCode(await beginning, "ABORTED");
    expect(releases).toBe(1);
    expect(root.children.size).toBe(0);
  });

  it("serializes APPEND against ABORT so no receipt or generation is resurrected", async () => {
    const root = new MemoryDirectory();
    const releases = { count: 0 };
    let resolveDigest:
      | ((digest: ArrayBuffer) => void)
      | undefined;
    const delayedCrypto = {
      subtle: {
        digest: () =>
          new Promise<ArrayBuffer>((resolve) => {
            resolveDigest = resolve;
          }),
      },
    } as unknown as Crypto;
    const runtime = createOpfsWorkerRuntime({
      root: root as unknown as FileSystemDirectoryHandle,
      crypto: delayedCrypto,
      policy: runtimePolicy,
      leaseManager: immediateLeases(releases),
      dedicatedWorker: true,
      supportsSynchronousAccessHandles: false,
    });
    expect(
      await runtime.handleRequest(
        beginRequest(
          "request_begin_5678",
          "transaction_56785678",
          scopeA,
        ),
      ),
    ).toMatchObject({ ok: true });

    const appending = runtime.handleRequest({
      requestId: "request_append_5678",
      protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
      kind: "APPEND_CHUNK",
      scope: scopeA,
      transactionId: "transaction_56785678",
      sequence: 0,
      bytes: new Uint8Array([7]).buffer,
    });
    await Promise.resolve();
    const aborting = runtime.handleRequest({
      requestId: "request_abort_5678",
      protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
      kind: "ABORT_PUT",
      scope: scopeA,
      transactionId: "transaction_56785678",
    });
    resolveDigest?.(new Uint8Array(32).buffer);

    expectFailureCode(await appending, "ABORTED");
    expect(await aborting).toMatchObject({ ok: true });
    expect(releases.count).toBe(1);
    expect(
      root.has([
        "authorities",
        scopeA.authorityToken,
        scopeA.namespaceToken,
        scopeA.partitionToken,
        "staging",
        "transaction_56785678",
      ]),
    ).toBe(false);
  });

  it("isolates physical chunks and cancellation tombstones by opaque authority", async () => {
    const root = new MemoryDirectory();
    const releases = { count: 0 };
    const runtime = createOpfsWorkerRuntime({
      root: root as unknown as FileSystemDirectoryHandle,
      crypto: globalThis.crypto,
      policy: runtimePolicy,
      leaseManager: immediateLeases(releases),
      dedicatedWorker: true,
      supportsSynchronousAccessHandles: false,
    });
    const transactionId = "transaction_99999999";
    for (const [index, targetScope] of [scopeA, scopeB].entries()) {
      expect(
        await runtime.handleRequest(
          beginRequest(
            `request_begin_iso_${index}`,
            transactionId,
            targetScope,
          ),
        ),
      ).toMatchObject({ ok: true });
      expect(
        await runtime.handleRequest({
          requestId: `request_append_iso_${index}`,
          protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
          kind: "APPEND_CHUNK",
          scope: targetScope,
          transactionId,
          sequence: 0,
          bytes: new Uint8Array([42]).buffer,
        }),
      ).toMatchObject({ ok: true });
    }

    expect(
      await runtime.handleRequest({
        requestId: "request_abort_iso_a",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "ABORT_PUT",
        scope: scopeA,
        transactionId,
      }),
    ).toMatchObject({ ok: true });
    const finished = await runtime.handleRequest({
      requestId: "request_finish_iso_b",
      protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
      kind: "FINISH_PUT",
      scope: scopeB,
      transactionId,
    });
    expect(finished).toMatchObject({ ok: true });
    const preparedObject = preparedValue(finished);
    expect(
      await runtime.handleRequest({
        requestId: "request_verify_iso_b",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "VERIFY_OBJECT",
        preparedObject,
      }),
    ).toEqual({
      requestId: "request_verify_iso_b",
      protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
      kind: expect.any(String),
      ok: true,
      value: true,
    });
  });

  it("garbage-collects orphan chunks only inside the requested physical authority", async () => {
    const root = new MemoryDirectory();
    const releases = { count: 0 };
    const runtime = createOpfsWorkerRuntime({
      root: root as unknown as FileSystemDirectoryHandle,
      crypto: globalThis.crypto,
      policy: runtimePolicy,
      leaseManager: immediateLeases(releases),
      dedicatedWorker: true,
      supportsSynchronousAccessHandles: false,
    });
    const preparedByScope = new Map<string, OpfsPreparedObject>();
    for (const [index, targetScope] of [scopeA, scopeB].entries()) {
      const transactionId = `transaction_gc_${index}_1234`;
      await runtime.handleRequest(
        beginRequest(
          `request_gc_begin_${index}`,
          transactionId,
          targetScope,
        ),
      );
      await runtime.handleRequest({
        requestId: `request_gc_append_${index}`,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "APPEND_CHUNK",
        scope: targetScope,
        transactionId,
        sequence: 0,
        bytes: new Uint8Array([99]).buffer,
      });
      const object = preparedValue(
        await runtime.handleRequest({
          requestId: `request_gc_finish_${index}`,
          protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
          kind: "FINISH_PUT",
          scope: targetScope,
          transactionId,
        }),
      );
      preparedByScope.set(targetScope.authorityToken, object);
      await runtime.handleRequest({
        requestId: `request_gc_finalize_${index}`,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "FINALIZE_PUT",
        transactionId,
        preparedObject: object,
      });
    }
    const digestHex = preparedByScope.get(
      scopeA.authorityToken,
    )!.chunks[0]!.digestHex;
    const cutoff = Date.now() + 10_000;
    expect(
      await runtime.handleRequest({
        requestId: "request_gc_list_a",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "LIST_ORPHAN_CANDIDATES",
        scope: scopeA,
        olderThanEpochMs: cutoff,
        maxEntries: 10,
      }),
    ).toMatchObject({ ok: true, value: { digests: [digestHex] } });
    expect(
      await runtime.handleRequest({
        requestId: "request_gc_delete_a",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "DELETE_ORPHAN_CHUNK",
        scope: scopeA,
        digestHex,
        olderThanEpochMs: cutoff,
      }),
    ).toMatchObject({ ok: true, value: { deleted: true } });

    expect(
      await runtime.handleRequest({
        requestId: "request_gc_verify_a",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "VERIFY_OBJECT",
        preparedObject: preparedByScope.get(scopeA.authorityToken)!,
      }),
    ).toEqual({
      requestId: "request_gc_verify_a",
      protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
      kind: expect.any(String),
      ok: true,
      value: false,
    });
    expect(
      await runtime.handleRequest({
        requestId: "request_gc_verify_b",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "VERIFY_OBJECT",
        preparedObject: preparedByScope.get(scopeB.authorityToken)!,
      }),
    ).toEqual({
      requestId: "request_gc_verify_b",
      protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
      kind: expect.any(String),
      ok: true,
      value: true,
    });
  });

  it("reports every required capability and fails closed when the lock is absent", async () => {
    const runtime = createOpfsWorkerRuntime({
      root: new MemoryDirectory() as unknown as FileSystemDirectoryHandle,
      crypto: globalThis.crypto,
      policy: runtimePolicy,
      leaseManager: null,
      dedicatedWorker: true,
      supportsSynchronousAccessHandles: true,
    });
    expect(
      await runtime.handleRequest({
        requestId: "request_caps_1234",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "CAPABILITIES",
      }),
    ).toEqual({
      requestId: "request_caps_1234",
      protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
      kind: expect.any(String),
      ok: true,
      value: {
        available: false,
        dedicatedWorkerRequired: true,
        crossContextMutationLockAvailable: false,
        synchronousAccessHandleAvailable: true,
      },
    });
    expectFailureCode(
      await runtime.handleRequest(
        beginRequest(
          "request_begin_caps",
          "transaction_caps_1234",
          scopeA,
        ),
      ),
      "UNSUPPORTED",
    );
  });
});

describe("OPFS worker client lifecycle", () => {
  it("rejects a colliding concurrent request id without replacing the first RPC", async () => {
    let listener: ((event: MessageEvent<unknown>) => void) | undefined;
    const posted: OpfsWorkerRequest[] = [];
    const worker: OpfsWorkerLike = {
      postMessage(message) {
        posted.push(message);
      },
      addEventListener(_type, next) {
        listener = next;
      },
      removeEventListener() {
        listener = undefined;
      },
    };
    const gateway = createOpfsWorkerGateway({
      worker,
      policy: runtimePolicy,
      createRequestId: () => "request_collision_1234",
    });

    const first = gateway.capabilities();
    const second = gateway.capabilities();
    expect(posted).toHaveLength(1);

    listener?.({
      data: {
        requestId: "request_collision_1234",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "CAPABILITIES",
        ok: true,
        value: {
          available: true,
          dedicatedWorkerRequired: true,
          crossContextMutationLockAvailable: true,
          synchronousAccessHandleAvailable: true,
        },
      },
    } as MessageEvent<unknown>);
    const secondResult = await second;
    gateway.close();
    await expect(first).resolves.toMatchObject({ ok: true });
    expect(secondResult).toMatchObject({
      ok: false,
      error: { code: "UNAVAILABLE" },
    });
  });

  it("rejects pending RPCs immediately when the worker reports a fatal event", async () => {
    let failureListener: ((event: Event) => void) | undefined;
    const worker: OpfsWorkerLike = {
      postMessage() {},
      addEventListener() {},
      removeEventListener() {},
      addFailureEventListener(listener) {
        failureListener = listener;
      },
      removeFailureEventListener() {
        failureListener = undefined;
      },
    };
    const gateway = createOpfsWorkerGateway({
      worker,
      policy: runtimePolicy,
      createRequestId: () => "request_crash_1234",
    });
    const pending = gateway.capabilities();

    expect(failureListener).toBeTypeOf("function");
    if (!failureListener) {
      gateway.close();
      return;
    }
    failureListener(new Event("error"));
    await expect(pending).resolves.toMatchObject({
      ok: false,
      error: { code: "UNAVAILABLE" },
    });
    await expect(gateway.capabilities()).resolves.toMatchObject({
      ok: false,
      error: { code: "UNAVAILABLE" },
    });
  });

  it("scopes the open signal to verification and leaves the acquired source readable", async () => {
    let listener:
      | ((event: MessageEvent<unknown>) => void)
      | undefined;
    const worker: OpfsWorkerLike = {
      postMessage(message) {
        const response: OpfsWorkerResponse =
          message.kind === "VERIFY_OBJECT"
            ? {
                requestId: message.requestId,
                protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
                kind: message.kind,
                ok: true,
                value: true,
              }
            : {
                requestId: message.requestId,
                protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
                kind: message.kind,
                ok: true,
                value: new Uint8Array([4, 2]).buffer,
              };
        queueMicrotask(() =>
          listener?.({ data: response } as MessageEvent<unknown>),
        );
      },
      addEventListener(_type, nextListener) {
        listener = nextListener;
      },
      removeEventListener() {
        listener = undefined;
      },
    };
    let requestSequence = 0;
    const gateway = createOpfsWorkerGateway({
      worker,
      policy: runtimePolicy,
      createRequestId: () => `request_open_${++requestSequence}_1234`,
    });
    const preparedObject: OpfsPreparedObject = {
      physicalSchemaVersion: 1,
      descriptor: {
        objectId: "object_open_12345678",
        scope: scopeA,
        generation: 1,
        byteLength: 2,
        mediaType: "application/octet-stream",
        createdAtEpochMs: 100,
        integrity: {
          algorithm: "SHA-256-TREE-V1",
          rootDigestHex: "a".repeat(64),
          chunkSizeBytes: runtimePolicy.chunkSizeBytes,
        },
        storagePolicy,
      },
      chunks: [
        {
          sequence: 0,
          byteLength: 2,
          digestHex: "b".repeat(64),
        },
      ],
    };
    const openController = new AbortController();
    const opened = await gateway.openObject(
      preparedObject,
      openController.signal,
    );
    expect(opened.ok).toBe(true);
    openController.abort();

    const chunks: number[][] = [];
    if (opened.ok) {
      for await (const result of opened.value.stream(
        new AbortController().signal,
      )) {
        expect(result.ok).toBe(true);
        if (result.ok) chunks.push([...result.value]);
      }
    }
    expect(chunks).toEqual([[4, 2]]);
    gateway.close();
  });

  it("removes its listener and rejects all pending RPCs before Worker termination", async () => {
    const listeners = new Set<(event: MessageEvent<unknown>) => void>();
    const worker: OpfsWorkerLike = {
      postMessage() {},
      addEventListener(_type, listener) {
        listeners.add(listener);
      },
      removeEventListener(_type, listener) {
        listeners.delete(listener);
      },
    };
    const gateway = createOpfsWorkerGateway({
      worker,
      policy: runtimePolicy,
      createRequestId: () => "request_pending_1234",
    });
    const pending = gateway.capabilities();
    expect(listeners.size).toBe(1);
    gateway.close();
    expect(listeners.size).toBe(0);
    await expect(pending).resolves.toMatchObject({
      ok: false,
      error: { code: "UNAVAILABLE" },
    });
    await expect(gateway.capabilities()).resolves.toMatchObject({
      ok: false,
      error: { code: "UNAVAILABLE" },
    });
  });
});

/**
 * STO-RR-01. A Web Lock is not reentrant. Any path that re-acquires the origin
 * mutation lease while already holding it stops making progress forever, and a
 * lock the runtime waits on cannot be observed by a fake that hands out an
 * unlimited number of leases.
 */
function strictNonReentrantLeases(
  counters: { acquires: number; releases: number },
): OpfsMutationLeaseManager {
  let held = false;
  return {
    async acquire() {
      if (held) {
        // A second holder waits for the first to release. Nothing here ever
        // does, which is exactly what a deadlock looks like.
        return await new Promise<never>(() => {});
      }
      held = true;
      counters.acquires += 1;
      let released = false;
      return {
        release() {
          if (released) return;
          released = true;
          held = false;
          counters.releases += 1;
        },
      };
    },
  };
}

function withTimeout<Value>(
  operation: Promise<Value>,
  label: string,
  ms = 200,
): Promise<Value> {
  return Promise.race([
    operation,
    new Promise<never>((_resolve, reject) => {
      setTimeout(() => reject(new Error(`${label} did not settle`)), ms);
    }),
  ]);
}

describe("STO-RR-01 OPFS finalization under a non-reentrant lock", () => {
  async function completedPut(
    runtime: ReturnType<typeof createOpfsWorkerRuntime>,
    transactionId: string,
  ): Promise<OpfsPreparedObject> {
    expect(
      await runtime.handleRequest(
        beginRequest(`request_begin_${transactionId}`, transactionId, scopeA),
      ),
    ).toMatchObject({ ok: true });
    expect(
      await runtime.handleRequest({
        requestId: `request_append_${transactionId}`,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "APPEND_CHUNK",
        scope: scopeA,
        transactionId,
        sequence: 0,
        bytes: new Uint8Array([9]).buffer,
      }),
    ).toMatchObject({ ok: true });
    const finished = await runtime.handleRequest({
      requestId: `request_finish_${transactionId}`,
      protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
      kind: "FINISH_PUT",
      scope: scopeA,
      transactionId,
    });
    expect(finished).toMatchObject({ ok: true });
    return preparedValue(finished);
  }

  it("finalizes a normal PUT with exactly one lock acquisition", async () => {
    const root = new MemoryDirectory();
    const counters = { acquires: 0, releases: 0 };
    const runtime = createOpfsWorkerRuntime({
      root: root as unknown as FileSystemDirectoryHandle,
      crypto: globalThis.crypto,
      policy: runtimePolicy,
      leaseManager: strictNonReentrantLeases(counters),
      dedicatedWorker: true,
      supportsSynchronousAccessHandles: false,
    });
    const transactionId = "transaction_final_0001";
    const prepared = await completedPut(runtime, transactionId);
    const before = counters.acquires;

    const finalized = await withTimeout(
      runtime.handleRequest({
        requestId: "request_finalize_0001",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "FINALIZE_PUT",
        transactionId,
        preparedObject: prepared,
      }),
      "FINALIZE_PUT",
    );

    expect(finalized).toMatchObject({ ok: true });
    expect(counters.acquires - before).toBe(1);
    expect(counters.acquires).toBe(counters.releases);
    expect(
      root.has([
        "authorities",
        scopeA.authorityToken,
        scopeA.namespaceToken,
        scopeA.partitionToken,
        "staging",
        transactionId,
      ]),
    ).toBe(false);
  });

  it("leaves the lock free for the next mutation after a finalized PUT", async () => {
    const root = new MemoryDirectory();
    const counters = { acquires: 0, releases: 0 };
    const runtime = createOpfsWorkerRuntime({
      root: root as unknown as FileSystemDirectoryHandle,
      crypto: globalThis.crypto,
      policy: runtimePolicy,
      leaseManager: strictNonReentrantLeases(counters),
      dedicatedWorker: true,
      supportsSynchronousAccessHandles: false,
    });
    const first = "transaction_final_0002";
    const prepared = await completedPut(runtime, first);
    await withTimeout(
      runtime.handleRequest({
        requestId: "request_finalize_0002",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "FINALIZE_PUT",
        transactionId: first,
        preparedObject: prepared,
      }),
      "first FINALIZE_PUT",
    );

    const second = "transaction_final_0003";
    await expect(
      withTimeout(completedPut(runtime, second), "second PUT"),
    ).resolves.toMatchObject({ descriptor: { objectId: "object_12345678" } });
    expect(counters.acquires).toBe(counters.releases);
  });
});

/**
 * STO-RR-02. A failure raised while serving a validated request must answer
 * that request. Defaulting the response kind to `CAPABILITIES` made the client's
 * own expected-kind check reject it as a protocol breach, so a quota or
 * integrity failure reached the caller as `UNSUPPORTED`.
 */
describe("STO-RR-02 worker failure responses echo the request kind", () => {
  const failingRoot = {
    async getDirectoryHandle(): Promise<FileSystemDirectoryHandle> {
      throw new DOMException("Out of room", "QuotaExceededError");
    },
    async getFileHandle(): Promise<FileSystemFileHandle> {
      throw new DOMException("Out of room", "QuotaExceededError");
    },
    async removeEntry(): Promise<void> {
      throw new DOMException("Out of room", "QuotaExceededError");
    },
    async *entries(): AsyncIterableIterator<never> {},
  } as unknown as FileSystemDirectoryHandle;

  it("keeps the validated kind on every failure path", async () => {
    const counters = { acquires: 0, releases: 0 };
    const runtime = createOpfsWorkerRuntime({
      root: failingRoot,
      crypto: globalThis.crypto,
      policy: runtimePolicy,
      leaseManager: strictNonReentrantLeases(counters),
      dedicatedWorker: true,
      supportsSynchronousAccessHandles: false,
    });

    const requests: readonly OpfsWorkerRequest[] = [
      beginRequest("request_kind_begin", "transaction_kind_0001", scopeA),
      {
        requestId: "request_kind_remove",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "REMOVE_OBJECT",
        scope: scopeA,
        objectId: "object_12345678",
        generation: 1,
      },
      {
        requestId: "request_kind_cleanup",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "CLEANUP_TRANSACTION",
        scope: scopeA,
        transactionId: "transaction_kind_0001",
      },
      {
        requestId: "request_kind_orphans",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "LIST_ORPHAN_CANDIDATES",
        scope: scopeA,
        olderThanEpochMs: 1,
        maxEntries: 1,
      },
    ];

    for (const request of requests) {
      const response = await runtime.handleRequest(request);
      expect(response).toMatchObject({
        ok: false,
        kind: request.kind,
        requestId: request.requestId,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
      });
    }
  });

  it("still reports a protocol-level failure for an unreadable envelope", async () => {
    const counters = { acquires: 0, releases: 0 };
    const runtime = createOpfsWorkerRuntime({
      root: failingRoot,
      crypto: globalThis.crypto,
      policy: runtimePolicy,
      leaseManager: strictNonReentrantLeases(counters),
      dedicatedWorker: true,
      supportsSynchronousAccessHandles: false,
    });

    expect(
      await runtime.handleRequest({
        requestId: "request_kind_broken",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "NOT_A_KIND",
      }),
    ).toMatchObject({ ok: false, kind: "CAPABILITIES" });
  });
});

/**
 * STO-RR-03. The client decoder is the trust boundary for anything a worker
 * says. A `code` that is merely a string lets an arbitrary value escape the
 * closed `BrowserDataFailure` taxonomy into application code.
 */
describe("STO-RR-03 worker responses are decoded against closed sets", () => {
  function respondingWorker(
    reply: (request: OpfsWorkerRequest) => unknown,
  ): OpfsWorkerLike {
    const listeners = new Set<(event: MessageEvent<unknown>) => void>();
    return {
      postMessage(message: unknown) {
        const response = reply(message as OpfsWorkerRequest);
        queueMicrotask(() => {
          for (const listener of listeners) {
            listener({ data: response } as MessageEvent<unknown>);
          }
        });
      },
      addEventListener(_type: "message", listener: (event: MessageEvent<unknown>) => void) {
        listeners.add(listener);
      },
      removeEventListener(_type: "message", listener: (event: MessageEvent<unknown>) => void) {
        listeners.delete(listener);
      },
    } as unknown as OpfsWorkerLike;
  }

  const hostileReplies: readonly (readonly [string, (request: OpfsWorkerRequest) => unknown])[] = [
    [
      "unknown failure code",
      (request) => ({
        requestId: request.requestId,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: request.kind,
        ok: false,
        failure: { code: "EVIL", retryable: false },
      }),
    ],
    [
      "unknown request kind",
      (request) => ({
        requestId: request.requestId,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: "NOT_A_KIND",
        ok: false,
        failure: { code: "UNAVAILABLE", retryable: false },
      }),
    ],
    [
      "non-boolean retryable",
      (request) => ({
        requestId: request.requestId,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: request.kind,
        ok: false,
        failure: { code: "UNAVAILABLE", retryable: "yes" },
      }),
    ],
    [
      "inherited failure fields",
      (request) => ({
        requestId: request.requestId,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: request.kind,
        ok: false,
        failure: Object.create({ code: "UNAVAILABLE", retryable: false }) as object,
      }),
    ],
    [
      "throwing getter",
      (request) => {
        const response: Record<string, unknown> = {
          requestId: request.requestId,
          protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
          ok: false,
          failure: { code: "UNAVAILABLE", retryable: false },
        };
        Object.defineProperty(response, "kind", {
          enumerable: true,
          get: () => {
            throw new TypeError("hostile getter");
          },
        });
        return response;
      },
    ],
    [
      "extra own field",
      (request) => ({
        requestId: request.requestId,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: request.kind,
        ok: false,
        failure: { code: "UNAVAILABLE", retryable: false, injected: 1 },
      }),
    ],
  ];

  for (const [label, reply] of hostileReplies) {
    it(`closes a ${label} as UNSUPPORTED without rejecting`, async () => {
      const gateway = createOpfsWorkerGateway({
        worker: respondingWorker(reply),
        policy: runtimePolicy,
        createRequestId: () => `request_hostile_${label.replace(/\W/gu, "")}`,
      });
      const result = await withTimeout(gateway.capabilities(), label);
      expect(result.ok).toBe(false);
      expect(result.ok ? null : result.error.code).toBe("UNSUPPORTED");
    });
  }

  /**
   * NS-06. Correlation was read separately and the pending row and its timer
   * were removed before the reply was decoded. A trap that threw inside the
   * decoder therefore left the public promise pending forever, and a throwing
   * `requestId` getter produced a timeout instead of a prompt protocol failure.
   */
  const uncorrelatableReplies: readonly (readonly [
    string,
    (request: OpfsWorkerRequest) => unknown,
  ])[] = [
    [
      "throwing requestId getter",
      (request) =>
        Object.defineProperty(
          {
            protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
            kind: request.kind,
            ok: true,
            value: {},
          },
          "requestId",
          {
            enumerable: true,
            get: () => {
              throw new TypeError("hostile requestId getter");
            },
          },
        ),
    ],
    [
      "throwing ownKeys trap",
      (request) =>
        new Proxy(
          {
            requestId: request.requestId,
            protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
            kind: request.kind,
            ok: true,
            value: {},
          },
          {
            ownKeys() {
              throw new TypeError("hostile ownKeys trap");
            },
          },
        ),
    ],
    [
      "descriptor trap that throws after correlation",
      (request) => {
        let reads = 0;
        return new Proxy(
          {
            requestId: request.requestId,
            protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
            kind: request.kind,
            ok: true,
            value: {},
          },
          {
            getOwnPropertyDescriptor(target, key) {
              reads += 1;
              if (reads > 1) {
                throw new TypeError("stateful descriptor trap");
              }
              return Reflect.getOwnPropertyDescriptor(target, key);
            },
          },
        );
      },
    ],
    [
      "symbol-keyed field",
      (request) => ({
        requestId: request.requestId,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: request.kind,
        ok: true,
        value: {},
        [Symbol.for("injected")]: true,
      }),
    ],
    [
      "non-enumerable own field",
      (request) =>
        Object.defineProperty(
          {
            requestId: request.requestId,
            protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
            kind: request.kind,
            ok: true,
            value: {},
          },
          "injected",
          { enumerable: false, value: true },
        ),
    ],
  ];

  for (const [label, reply] of uncorrelatableReplies) {
    it(`closes a ${label} promptly as UNSUPPORTED`, async () => {
      const gateway = createOpfsWorkerGateway({
        worker: respondingWorker(reply),
        policy: { ...runtimePolicy, rpcTimeoutMs: 60_000 },
        createRequestId: () => `request_uncorr_${label.replace(/\W/gu, "")}`,
      });
      // The RPC timeout is far beyond the test budget, so a pass here means the
      // reply itself closed the request rather than the timer.
      const result = await withTimeout(gateway.capabilities(), label);
      expect(result.ok).toBe(false);
      expect(result.ok ? null : result.error.code).toBe("UNSUPPORTED");
    });
  }

  it("keeps serving requests after a malformed reply", async () => {
    let replies = 0;
    const gateway = createOpfsWorkerGateway({
      worker: respondingWorker((request) => {
        replies += 1;
        if (replies === 1) return { requestId: request.requestId };
        return {
          requestId: request.requestId,
          protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
          kind: request.kind,
          ok: false,
          failure: { code: "QUOTA_EXCEEDED", retryable: true },
        };
      }),
      policy: { ...runtimePolicy, rpcTimeoutMs: 60_000 },
      createRequestId: () => `request_sequence_${replies}`,
    });

    const first = await withTimeout(gateway.capabilities(), "first");
    expect(first.ok ? null : first.error.code).toBe("UNSUPPORTED");
    const second = await withTimeout(gateway.capabilities(), "second");
    expect(second.ok ? null : second.error.code).toBe("QUOTA_EXCEEDED");
  });

  it("still admits a well-formed closed failure", async () => {
    const gateway = createOpfsWorkerGateway({
      worker: respondingWorker((request) => ({
        requestId: request.requestId,
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind: request.kind,
        ok: false,
        failure: { code: "QUOTA_EXCEEDED", retryable: true },
      })),
      policy: runtimePolicy,
      createRequestId: () => "request_wellformed_1234",
    });
    const result = await withTimeout(gateway.capabilities(), "well-formed");
    expect(result.ok).toBe(false);
    expect(result.ok ? null : result.error.code).toBe("QUOTA_EXCEEDED");
  });
});

/**
 * NS-05. When the runtime never came up, the message host answered with a
 * default `CAPABILITIES` kind. The gateway saw that as an expected-kind
 * mismatch and replaced the real cause — `BLOCKED`, `QUOTA_EXCEEDED` — with a
 * generic `UNSUPPORTED` protocol breach, so the outage was misreported.
 */
describe("NS-05 a bootstrap failure answers the request it belongs to", () => {
  function hostFor(): Readonly<{
    host: OpfsWorkerMessageHost;
    posted: OpfsWorkerResponse[];
    deliver(message: unknown): void;
  }> {
    const listeners: ((event: MessageEvent<unknown>) => void)[] = [];
    const posted: OpfsWorkerResponse[] = [];
    return {
      host: {
        addEventListener(_type, listener) {
          listeners.push(listener);
        },
        postMessage(message) {
          posted.push(message);
        },
      },
      posted,
      deliver(message: unknown) {
        for (const listener of listeners) {
          listener({ data: message } as MessageEvent<unknown>);
        }
      },
    };
  }

  const requestKinds = [
    "CAPABILITIES",
    "BEGIN_PUT",
    "APPEND_CHUNK",
    "FINISH_PUT",
    "ABORT_PUT",
    "VERIFY_OBJECT",
    "READ_CHUNK",
    "REMOVE_OBJECT",
    "CLEANUP_TRANSACTION",
    "FINALIZE_PUT",
    "LIST_ORPHAN_CANDIDATES",
    "DELETE_ORPHAN_CHUNK",
  ] as const;

  for (const kind of requestKinds) {
    it(`preserves the ${kind} correlation when bootstrap fails`, async () => {
      const { host, posted, deliver } = hostFor();
      const start = startBrowserOpfsDedicatedWorker(host, {
        storageManager: {
          getDirectory: () =>
            Promise.reject(
              new DOMException("blocked", "SecurityError"),
            ),
        } as unknown as StorageManager,
        crypto: globalThis.crypto,
      });
      // The bootstrap rejection must not escape the worker entry point either.
      await expect(start).rejects.toBeInstanceOf(Error);

      deliver({
        requestId: "request_bootstrap_failure",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind,
      });
      await new Promise((resolve) => setTimeout(resolve, 0));

      expect(posted).toHaveLength(1);
      expect(posted[0]).toMatchObject({
        requestId: "request_bootstrap_failure",
        protocolVersion: OPFS_WORKER_PROTOCOL_VERSION,
        kind,
        ok: false,
      });
    });
  }
});