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

import {
  createBrowserRpcRuntime,
  defineBrowserRpcTransport,
  type BrowserRpcServerStreamLease,
  type BrowserRpcStreamFrame,
  type BrowserRpcTransport,
} from "../../../src/adapters/browser-rpc/index.ts";
import { decodeServerStreamLease } from "../../../src/adapters/browser-rpc/transport.ts";
import { installBrowserRpcContractBindings } from "../../../src/contracts/browser-rpc.ts";
import {
  MAPPERS,
  SCHEMA_CODECS,
  STREAM_ENCODER,
  UNARY_ENCODER,
  isResourceView,
  streamOperation,
  streamProfile,
  unaryOperation,
  unaryProfile,
} from "./fixture.ts";

function unaryRuntime(
  transport: BrowserRpcTransport,
  extra: Record<string, unknown> = {},
) {
  return createBrowserRpcRuntime({
    operations: { GET_RPC_RESOURCE: unaryOperation() },
    profiles: { CONNECT_REFERENCE_UNARY: unaryProfile() },
    schemaCodecs: SCHEMA_CODECS,
    mappers: MAPPERS,
    requestEncoders: { RpcResourceRequestEncoder: UNARY_ENCODER },
    transports: { CONNECT_REFERENCE_UNARY: transport },
    ...extra,
  });
}

function streamingRuntime(
  transport: BrowserRpcTransport,
  extra: Record<string, unknown> = {},
) {
  return createBrowserRpcRuntime({
    operations: { WATCH_RPC_RESOURCES: streamOperation() },
    profiles: { CONNECT_REFERENCE_STREAM: streamProfile() },
    schemaCodecs: SCHEMA_CODECS,
    mappers: MAPPERS,
    requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER },
    transports: { CONNECT_REFERENCE_STREAM: transport },
    ...extra,
  });
}

function unaryTransport(
  invokeUnary: BrowserRpcTransport["invokeUnary"],
): BrowserRpcTransport {
  return defineBrowserRpcTransport({
    runtimeProfileId: "CONNECT_REFERENCE_UNARY",
    providerId: "REFERENCE_RPC",
    protocol: "CONNECT_HTTP",
    rpcKind: "UNARY",
    invokeUnary,
  });
}

type LeaseProbe = Readonly<{
  transport: BrowserRpcTransport;
  cancels: string[];
  readonly opened: number;
  close(): void;
}>;

/**
 * RPC-RR-01. A transport whose physical stream does not close on its own. The
 * runtime must cancel it exactly once and must not admit a second stream for
 * the same operation until `waitClosed()` settles.
 */
function nonCooperativeStreamTransport(): LeaseProbe {
  const cancels: string[] = [];
  const counter = { opened: 0 };
  let release: (() => void) | undefined;
  const closed = new Promise<void>((resolve) => {
    release = resolve;
  });
  const openServerStream = (): BrowserRpcServerStreamLease => {
    counter.opened += 1;
    return {
      streamId: `physical-${counter.opened}`,
      frames: {
        [Symbol.asyncIterator]: () =>
          ({
            // Never yields and never settles: the runtime's own deadline is the
            // only thing that can end the call.
            next: () => new Promise<never>(() => {}),
          }) as AsyncIterator<BrowserRpcStreamFrame>,
      },
      cancel(reason: string) {
        cancels.push(reason);
      },
      waitClosed: () => closed,
    };
  };
  const transport = defineBrowserRpcTransport({
    runtimeProfileId: "CONNECT_REFERENCE_STREAM",
    providerId: "REFERENCE_RPC",
    protocol: "CONNECT_HTTP",
    rpcKind: "SERVER_STREAM",
    openServerStream,
  });
  return Object.freeze({
    transport,
    cancels,
    get opened() {
      return counter.opened;
    },
    close: () => release?.(),
  }) as LeaseProbe;
}

function streamTransport(
  frames: readonly BrowserRpcStreamFrame[],
): BrowserRpcTransport {
  return defineBrowserRpcTransport({
    runtimeProfileId: "CONNECT_REFERENCE_STREAM",
    providerId: "REFERENCE_RPC",
    protocol: "CONNECT_HTTP",
    rpcKind: "SERVER_STREAM",
    openServerStream: () => ({
      streamId: "remediation-stream",
      frames: {
        async *[Symbol.asyncIterator]() {
          for (const frame of frames) yield frame;
        },
      },
      cancel() {},
      async waitClosed() {},
    }),
  });
}

async function collect(
  source: AsyncIterable<unknown>,
): Promise<readonly unknown[]> {
  const seen: unknown[] = [];
  for await (const value of source) seen.push(value);
  return seen;
}

/**
 * RPC-RR-02. A collaborator that throws synchronously must not escape the
 * Result contract. A generation fence captured outside the protected boundary
 * and a `clock.sleep` invoked outside a promise boundary both did exactly that,
 * and the second also skipped the listener and timer release.
 */
describe("RPC-RR-02 synchronous collaborator throws stay inside Result", () => {
  const throwingFence = {
    capture: () => {
      throw new TypeError("fence exploded");
    },
    isCurrent: () => true,
  };

  it("closes a unary call whose fence throws", async () => {
    const runtime = unaryRuntime(
      unaryTransport(async () => ({
        ok: true,
        message: { id: "a", name: "A" },
        encodedBytes: 8,
      })),
      { generationFence: throwingFence },
    );

    const result = await runtime
      .bindUnary("GET_RPC_RESOURCE", isResourceView)
      .execute({ resourceId: "resource-1" });
    expect(result.ok).toBe(false);
  });

  it("closes a stream whose fence throws", async () => {
    const runtime = streamingRuntime(
      streamTransport([{ kind: "TERMINAL", ok: true }]),
      { generationFence: throwingFence },
    );

    const results = await collect(
      runtime
        .bindServerStream("WATCH_RPC_RESOURCES", isResourceView)
        .open({ resourceId: "scope-1" }),
    );
    expect(results.length).toBeGreaterThan(0);
    expect(results.every((value) => (value as { ok: boolean }).ok === false)).toBe(
      true,
    );
  });

  it("closes a unary call whose clock throws synchronously", async () => {
    const runtime = unaryRuntime(
      unaryTransport(() => new Promise(() => {})),
      {
        clock: {
          now: () => 0,
          sleep: () => {
            throw new TypeError("clock exploded");
          },
        },
      },
    );

    const result = await runtime
      .bindUnary("GET_RPC_RESOURCE", isResourceView)
      .execute({ resourceId: "resource-1" });
    expect(result.ok).toBe(false);
  });
});

/**
 * RPC-RR-03. The runtime reads a validated snapshot, never the caller's
 * objects: an accessor is refused without being invoked, and no installed
 * registry exposes a mutator.
 */
describe("RPC-RR-03 transport and binding registries are snapshots", () => {
  it("never invokes a transport accessor", () => {
    let getterCalls = 0;
    const hostile = {} as Record<string, unknown>;
    Object.defineProperties(hostile, {
      runtimeProfileId: {
        enumerable: true,
        get: () => {
          getterCalls += 1;
          return "CONNECT_REFERENCE_UNARY";
        },
      },
      providerId: { enumerable: true, value: "REFERENCE_RPC" },
      protocol: { enumerable: true, value: "CONNECT_HTTP" },
      rpcKind: { enumerable: true, value: "UNARY" },
      invokeUnary: {
        enumerable: true,
        value: async () => ({ ok: true, message: {}, encodedBytes: 1 }),
      },
    });

    expect(() =>
      unaryRuntime(hostile as unknown as BrowserRpcTransport),
    ).toThrow(TypeError);
    expect(getterCalls).toBe(0);
  });

  it("rejects an unexpected own field on a transport row", () => {
    const transport = unaryTransport(async () => ({
      ok: true,
      message: { id: "a", name: "A" },
      encodedBytes: 8,
    }));
    const widened = { ...transport, injected: true };
    expect(() =>
      unaryRuntime(widened as unknown as BrowserRpcTransport),
    ).toThrow(TypeError);
  });

  it("exposes no mutation API on any installed binding registry", () => {
    const installed = installBrowserRpcContractBindings({
      operations: { GET_RPC_RESOURCE: unaryOperation() },
      profiles: { CONNECT_REFERENCE_UNARY: unaryProfile() },
      schemaCodecs: SCHEMA_CODECS,
      mappers: MAPPERS,
      requestEncoders: { RpcResourceRequestEncoder: UNARY_ENCODER },
    });
    for (const registry of Object.values(installed)) {
      const record = registry as unknown as Record<string, unknown>;
      for (const mutator of ["set", "delete", "clear"]) {
        expect(record[mutator]).toBeUndefined();
      }
      expect(() =>
        Map.prototype.clear.call(registry as never),
      ).toThrow();
    }
    expect(installed.operations.get("GET_RPC_RESOURCE")).toBeDefined();
  });
});

/**
 * RPC-RR-04. Transport values are decoded, not adopted. `in` and a direct
 * property read run accessors and admit inherited or extra fields, and keeping
 * the caller's object lets it change after validation.
 */
describe("RPC-RR-04 transport results and frames are exactly decoded", () => {
  const hostileResults: readonly (readonly [string, () => unknown])[] = [
    ["extra own field", () => ({ ok: true, message: {}, encodedBytes: 1, injected: 1 })],
    [
      "inherited fields",
      () =>
        Object.create({ ok: true, message: {}, encodedBytes: 1 }) as object,
    ],
    [
      "throwing getter",
      () => {
        const value: Record<string, unknown> = { message: {}, encodedBytes: 1 };
        Object.defineProperty(value, "ok", {
          enumerable: true,
          get: () => {
            throw new TypeError("hostile getter");
          },
        });
        return value;
      },
    ],
    [
      "symbol key",
      () => ({
        ok: true,
        message: {},
        encodedBytes: 1,
        [Symbol("injected")]: 1,
      }),
    ],
  ];

  for (const [label, build] of hostileResults) {
    it(`refuses a unary result with ${label}`, async () => {
      const runtime = unaryRuntime(
        unaryTransport(async () => build() as never),
      );
      const result = await runtime
        .bindUnary("GET_RPC_RESOURCE", isResourceView)
        .execute({ resourceId: "resource-1" });
      expect(result.ok).toBe(false);
    });
  }

  it("refuses an unknown transport failure code", async () => {
    const runtime = unaryRuntime(
      unaryTransport(async () => ({
        ok: false,
        failure: { code: "MADE_UP_CODE" },
      }) as never),
    );
    const result = await runtime
      .bindUnary("GET_RPC_RESOURCE", isResourceView)
      .execute({ resourceId: "resource-1" });
    expect(result.ok).toBe(false);
  });

  it("refuses a retryAfterMs outside the hard ceiling", async () => {
    const runtime = unaryRuntime(
      unaryTransport(async () => ({
        ok: false,
        failure: { code: "NETWORK_UNREACHABLE", retryAfterMs: -1 },
      }) as never),
    );
    const result = await runtime
      .bindUnary("GET_RPC_RESOURCE", isResourceView)
      .execute({ resourceId: "resource-1" });
    expect(result.ok).toBe(false);
  });

  it("refuses a stream frame with an extra own field", async () => {
    const runtime = streamingRuntime(
      streamTransport([
        {
          kind: "MESSAGE",
          message: { id: "a", name: "A" },
          encodedBytes: 4,
          injected: 1,
        } as never,
      ]),
    );
    const results = await collect(
      runtime
        .bindServerStream("WATCH_RPC_RESOURCES", isResourceView)
        .open({ resourceId: "scope-1" }),
    );
    expect(results.every((value) => (value as { ok: boolean }).ok === false)).toBe(
      true,
    );
  });
});

/**
 * RPC-RR-01. A bare `AsyncIterable` gave the runtime no way to cancel the
 * physical stream or to learn when it actually closed, so a timed-out call left
 * the first stream running against the server while a second was admitted.
 */
describe("RPC-RR-01 server stream leases and the DRAINING fence", () => {
  const shortDeadline = () =>
    streamOperation({ totalDeadlineMs: 25, idleDeadlineMs: 25 });

  it("cancels the physical stream exactly once after a timeout", async () => {
    const probe = nonCooperativeStreamTransport();
    const runtime = createBrowserRpcRuntime({
      operations: { WATCH_RPC_RESOURCES: shortDeadline() },
      profiles: { CONNECT_REFERENCE_STREAM: streamProfile() },
      schemaCodecs: SCHEMA_CODECS,
      mappers: MAPPERS,
      requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER },
      transports: { CONNECT_REFERENCE_STREAM: probe.transport },
    });

    const results = await collect(
      runtime
        .bindServerStream("WATCH_RPC_RESOURCES", isResourceView)
        .open({ resourceId: "scope-1" }),
    );

    expect(results.every((value) => (value as { ok: boolean }).ok === false)).toBe(
      true,
    );
    expect(probe.cancels.length).toBe(1);
  });

  it("refuses a second stream while the first has not confirmed closure", async () => {
    const probe = nonCooperativeStreamTransport();
    const runtime = createBrowserRpcRuntime({
      operations: { WATCH_RPC_RESOURCES: shortDeadline() },
      profiles: { CONNECT_REFERENCE_STREAM: streamProfile() },
      schemaCodecs: SCHEMA_CODECS,
      mappers: MAPPERS,
      requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER },
      transports: { CONNECT_REFERENCE_STREAM: probe.transport },
    });
    const stream = runtime.bindServerStream(
      "WATCH_RPC_RESOURCES",
      isResourceView,
    );

    await collect(stream.open({ resourceId: "scope-1" }));
    expect(probe.opened).toBe(1);

    const second = await collect(stream.open({ resourceId: "scope-1" }));
    expect(second).toHaveLength(1);
    expect(second[0]).toMatchObject({
      ok: false,
      error: { code: "RPC_STREAM_DRAINING" },
    });
    // The refused call never reached the transport.
    expect(probe.opened).toBe(1);

    // Once the transport confirms closure the operation admits work again.
    probe.close();
    await new Promise((resolve) => setTimeout(resolve, 5));
    await collect(stream.open({ resourceId: "scope-1" }));
    expect(probe.opened).toBe(2);
  });

  it("refuses a malformed lease as a protocol failure", async () => {
    for (const malformed of [
      { frames: { [Symbol.asyncIterator]: () => ({ next: async () => ({ done: true, value: undefined }) }) }, cancel() {}, async waitClosed() {} },
      { streamId: "", frames: { [Symbol.asyncIterator]: () => ({ next: async () => ({ done: true, value: undefined }) }) }, cancel() {}, async waitClosed() {} },
      { streamId: "s1", frames: {}, cancel() {}, async waitClosed() {} },
      { streamId: "s1", frames: { [Symbol.asyncIterator]: () => ({ next: async () => ({ done: true, value: undefined }) }) }, cancel: 1, async waitClosed() {} },
    ]) {
      const runtime = createBrowserRpcRuntime({
        operations: { WATCH_RPC_RESOURCES: streamOperation() },
        profiles: { CONNECT_REFERENCE_STREAM: streamProfile() },
        schemaCodecs: SCHEMA_CODECS,
        mappers: MAPPERS,
        requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER },
        transports: {
          CONNECT_REFERENCE_STREAM: defineBrowserRpcTransport({
            runtimeProfileId: "CONNECT_REFERENCE_STREAM",
            providerId: "REFERENCE_RPC",
            protocol: "CONNECT_HTTP",
            rpcKind: "SERVER_STREAM",
            openServerStream: () => malformed as never,
          }),
        },
      });
      const results = await collect(
        runtime
          .bindServerStream("WATCH_RPC_RESOURCES", isResourceView)
          .open({ resourceId: "scope-1" }),
      );
      expect(
        results.every((value) => (value as { ok: boolean }).ok === false),
      ).toBe(true);
    }
  });
});

/**
 * RPC-01. A `waitClosed()` that rejects, throws, or does not even return a
 * promise is not evidence that the physical stream closed. Absorbing all three
 * into a fulfilled `undefined` let the registry prune a live stream and admit a
 * second one for the same operation against the same server.
 */
describe("RPC-01 only a positive receipt confirms physical closure", () => {
  const shortDeadline = () =>
    streamOperation({ totalDeadlineMs: 25, idleDeadlineMs: 25 });

  function leaseProbeWith(
    waitClosed: () => unknown,
  ): Readonly<{
    transport: BrowserRpcTransport;
    cancels: string[];
    opened(): number;
  }> {
    const cancels: string[] = [];
    const counter = { opened: 0 };
    const transport = defineBrowserRpcTransport({
      runtimeProfileId: "CONNECT_REFERENCE_STREAM",
      providerId: "REFERENCE_RPC",
      protocol: "CONNECT_HTTP",
      rpcKind: "SERVER_STREAM",
      openServerStream: () => {
        counter.opened += 1;
        return {
          streamId: `physical-${counter.opened}`,
          frames: {
            [Symbol.asyncIterator]: () =>
              ({
                next: () => new Promise<never>(() => {}),
              }) as AsyncIterator<BrowserRpcStreamFrame>,
          },
          cancel(reason: string) {
            cancels.push(reason);
          },
          waitClosed: waitClosed as () => Promise<void>,
        };
      },
    });
    return Object.freeze({
      transport,
      cancels,
      opened: () => counter.opened,
    });
  }

  const negativeReceipts = [
    {
      label: "rejects",
      waitClosed: () => Promise.reject(new Error("never closed")),
    },
    {
      label: "throws synchronously",
      waitClosed: () => {
        throw new TypeError("waitClosed exploded");
      },
    },
    { label: "returns a non-promise", waitClosed: () => "closed" },
    { label: "never settles", waitClosed: () => new Promise<void>(() => {}) },
  ];

  for (const { label, waitClosed } of negativeReceipts) {
    it(`keeps the operation DRAINING when waitClosed ${label}`, async () => {
      const probe = leaseProbeWith(waitClosed);
      const runtime = createBrowserRpcRuntime({
        operations: { WATCH_RPC_RESOURCES: shortDeadline() },
        profiles: { CONNECT_REFERENCE_STREAM: streamProfile() },
        schemaCodecs: SCHEMA_CODECS,
        mappers: MAPPERS,
        requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER },
        transports: { CONNECT_REFERENCE_STREAM: probe.transport },
      });
      const stream = runtime.bindServerStream(
        "WATCH_RPC_RESOURCES",
        isResourceView,
      );

      await collect(stream.open({ resourceId: "scope-1" }));
      expect(probe.opened()).toBe(1);
      expect(probe.cancels).toHaveLength(1);

      const second = await collect(stream.open({ resourceId: "scope-1" }));
      expect(second).toHaveLength(1);
      expect(second[0]).toMatchObject({
        ok: false,
        error: { code: "RPC_STREAM_DRAINING" },
      });
      // No second physical stream was ever opened.
      expect(probe.opened()).toBe(1);
    });
  }

  it("admits a second stream once waitClosed fulfils", async () => {
    let release: (() => void) | undefined;
    const closed = new Promise<void>((resolve) => {
      release = resolve;
    });
    const probe = leaseProbeWith(() => closed);
    const runtime = createBrowserRpcRuntime({
      operations: { WATCH_RPC_RESOURCES: shortDeadline() },
      profiles: { CONNECT_REFERENCE_STREAM: streamProfile() },
      schemaCodecs: SCHEMA_CODECS,
      mappers: MAPPERS,
      requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER },
      transports: { CONNECT_REFERENCE_STREAM: probe.transport },
    });
    const stream = runtime.bindServerStream(
      "WATCH_RPC_RESOURCES",
      isResourceView,
    );

    await collect(stream.open({ resourceId: "scope-1" }));
    release?.();
    await new Promise((resolve) => setTimeout(resolve, 5));
    await collect(stream.open({ resourceId: "scope-1" }));
    expect(probe.opened()).toBe(2);
  });

  /**
   * RPC-02. Cleanup reads foreign state, so it must stay inside the result
   * boundary: a throwing `return` accessor replaced the already selected
   * outcome with a native rejection and skipped the rest of the teardown.
   */
  it("keeps the selected outcome when the iterator return accessor throws", async () => {
    const cancels: string[] = [];
    const transport = defineBrowserRpcTransport({
      runtimeProfileId: "CONNECT_REFERENCE_STREAM",
      providerId: "REFERENCE_RPC",
      protocol: "CONNECT_HTTP",
      rpcKind: "SERVER_STREAM",
      openServerStream: () => ({
        streamId: "physical-hostile",
        frames: {
          [Symbol.asyncIterator]: () =>
            Object.defineProperty(
              { next: () => new Promise<never>(() => {}) },
              "return",
              {
                enumerable: true,
                get() {
                  throw new TypeError("return getter escaped");
                },
              },
            ) as AsyncIterator<BrowserRpcStreamFrame>,
        },
        cancel(reason: string) {
          cancels.push(reason);
        },
        waitClosed: async () => {},
      }),
    });
    const runtime = createBrowserRpcRuntime({
      operations: { WATCH_RPC_RESOURCES: shortDeadline() },
      profiles: { CONNECT_REFERENCE_STREAM: streamProfile() },
      schemaCodecs: SCHEMA_CODECS,
      mappers: MAPPERS,
      requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER },
      transports: { CONNECT_REFERENCE_STREAM: transport },
    });

    const results = await collect(
      runtime
        .bindServerStream("WATCH_RPC_RESOURCES", isResourceView)
        .open({ resourceId: "scope-1" }),
    );

    expect(results).toHaveLength(1);
    // The subject is that a throwing accessor cannot replace the outcome the
    // runtime already selected, not which deadline won. `totalDeadlineMs` and
    // `idleDeadlineMs` are deliberately equal here, so pinning one of the two
    // timeout codes would make this assertion a coin flip under load.
    expect(results[0]).toMatchObject({ ok: false });
    const failure = results[0] as { error: { kind: string; code: string } };
    expect(failure.error.kind).toBe("REQUEST_TIMEOUT");
    expect([
      "RPC_TOTAL_DEADLINE_EXCEEDED",
      "RPC_STREAM_IDLE_TIMEOUT",
    ]).toContain(failure.error.code);
    expect(cancels).toHaveLength(1);
  });

  it("decodes a hostile frames iterator to null instead of throwing", async () => {
    const hostile = {
      streamId: "physical-hostile",
      frames: Object.defineProperty({}, Symbol.asyncIterator, {
        enumerable: true,
        get() {
          throw new TypeError("async iterator getter escaped");
        },
      }),
      cancel() {},
      async waitClosed() {},
    };
    expect(decodeServerStreamLease(hostile)).toBeNull();
  });
});

/**
 * RPC-04. An exact union is exactly these own data keys on a plain object. A
 * custom prototype could otherwise carry metadata past the boundary while the
 * own shape still looked valid, and a missing wire field could reach a
 * permissive schema as `undefined`.
 */
describe("RPC-04 transport results are an exact union", () => {
  const hostileResults = [
    {
      label: "valid own fields with an inherited extra",
      result: () =>
        Object.assign(Object.create({ injected: "prototype" }), {
          ok: true,
          message: { id: "a", name: "A" },
          encodedBytes: 8,
        }),
    },
    {
      label: "a missing message",
      result: () => ({ ok: true, encodedBytes: 8 }),
    },
    {
      label: "a non-enumerable own extra",
      result: () =>
        Object.defineProperty(
          { ok: true, message: { id: "a", name: "A" }, encodedBytes: 8 },
          "injected",
          { enumerable: false, value: true },
        ),
    },
    {
      label: "a failure with an inherited code",
      result: () => ({
        ok: false,
        failure: Object.create({ code: "SERVER_FAILURE" }) as object,
      }),
    },
  ];

  for (const { label, result } of hostileResults) {
    it(`rejects ${label} as a protocol mismatch`, async () => {
      const runtime = unaryRuntime(
        unaryTransport(async () => result() as never),
      );
      await expect(
        runtime
          .bindUnary("GET_RPC_RESOURCE", isResourceView)
          .execute({ resourceId: "scope-1" }),
      ).resolves.toMatchObject({
        ok: false,
        error: { code: "RPC_PROTOCOL_MISMATCH" },
      });
    });
  }
});

/**
 * RPC-03. Every registry is snapshotted before anything validates it, so a
 * hostile accessor never runs, and a row that hides fields behind a prototype
 * or a non-enumerable key is refused rather than installed.
 */
describe("RPC-03 registries are snapshotted before they are validated", () => {
  const build = (operations: Record<string, unknown>) =>
    createBrowserRpcRuntime({
      operations: operations as never,
      profiles: { CONNECT_REFERENCE_UNARY: unaryProfile() },
      schemaCodecs: SCHEMA_CODECS,
      mappers: MAPPERS,
      requestEncoders: { RpcResourceRequestEncoder: UNARY_ENCODER },
      transports: {
        CONNECT_REFERENCE_UNARY: unaryTransport(async () => ({
          ok: true,
          message: { id: "a", name: "A" },
          encodedBytes: 8,
        })),
      },
    });

  it("never invokes a registry accessor", () => {
    let reads = 0;
    const operations = Object.defineProperty({}, "GET_RPC_RESOURCE", {
      enumerable: true,
      get() {
        reads += 1;
        return unaryOperation();
      },
    });
    expect(() => build(operations)).toThrow(TypeError);
    expect(reads).toBe(0);
  });

  const hostileRows = [
    {
      label: "an inherited extra field",
      row: () =>
        Object.assign(Object.create({ injected: true }), unaryOperation()),
    },
    {
      label: "a non-enumerable own extra field",
      row: () =>
        Object.defineProperty({ ...unaryOperation() }, "injected", {
          enumerable: false,
          value: true,
        }),
    },
    {
      label: "a symbol field",
      row: () => ({
        ...unaryOperation(),
        [Symbol.for("injected")]: true,
      }),
    },
  ];

  for (const { label, row } of hostileRows) {
    it(`refuses to install a row with ${label}`, () => {
      expect(() => build({ GET_RPC_RESOURCE: row() })).toThrow(TypeError);
    });
  }
});