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

import type { ClockPort } from "../../../src/application/ports/clock-port.ts";
import {
  realtimeTransportRecoveryCommitted,
  type RealtimeRecoveryCheckpoint,
} from "../../../src/application/ports/realtime/event-authority.ts";
import type {
  RealtimeResult,
} from "../../../src/application/ports/realtime/shared.ts";
import {
  realtimeFailure,
  realtimeSuccess,
} from "../../../src/adapters/realtime/result.ts";
import {
  createWebSocketConnection,
  type WebSocketConnectionDependencies,
  type WebSocketFacade,
  type WebSocketInboundEventOutcome,
  type WebSocketRecoveryRequest,
} from "../../../src/adapters/realtime/websocket/websocket-connection.ts";
import {
  REALTIME_WEBSOCKET_PROTOCOL,
  type WebSocketAdvertisedLimits,
} from "../../../src/adapters/realtime/websocket/websocket-protocol.ts";
import type {
  StreamRegistrationId,
} from "../../../src/contracts/realtime-streams.ts";

const LIMITS: WebSocketAdvertisedLimits = Object.freeze({
  maxFrameBytes: 64 * 1_024,
  maxSubscriptions: 32,
  maxInboundQueueCount: 256,
  maxInboundQueueBytes: 4 * 1_024 * 1_024,
  maxOutboundQueueCount: 128,
  maxOutboundQueueBytes: 256 * 1_024,
  maxBufferedAmountBytes: 256 * 1_024,
  maxEventsPerSecond: 256,
});
const RECOVERY_CHECKPOINT = Object.freeze({
  recoveryMode: "SESSION_REBUILD" as const,
  streamEpoch: "stream-epoch.0001",
  lastAppliedSequence: "0",
  resumeCursor: null,
}) as RealtimeRecoveryCheckpoint;
const RECOVERY_OUTCOME = realtimeTransportRecoveryCommitted(
  "presence.v1" as StreamRegistrationId,
  RECOVERY_CHECKPOINT,
);

type PendingSleep = {
  readonly milliseconds: number;
  readonly resolve: () => void;
  readonly reject: (reason?: unknown) => void;
  readonly signal?: AbortSignal;
  readonly abort: () => void;
  settled: boolean;
};

class ManualClock implements ClockPort {
  current = 1_000;
  readonly sleeps: PendingSleep[] = [];

  now(): number {
    return this.current;
  }

  sleep(
    milliseconds: number,
    signal?: AbortSignal,
  ): Promise<void> {
    return new Promise((resolve, reject) => {
      const pending: PendingSleep = {
        milliseconds,
        resolve: () => {
          if (pending.settled) return;
          pending.settled = true;
          signal?.removeEventListener("abort", pending.abort);
          this.current += milliseconds;
          resolve();
        },
        reject,
        signal,
        abort: () => {
          if (pending.settled) return;
          pending.settled = true;
          reject(new DOMException("aborted", "AbortError"));
        },
        settled: false,
      };
      if (signal?.aborted) {
        pending.abort();
        return;
      }
      signal?.addEventListener("abort", pending.abort, { once: true });
      this.sleeps.push(pending);
    });
  }

  resolveDuration(milliseconds: number): void {
    const pending = this.sleeps.find(
      (sleep) =>
        !sleep.settled && sleep.milliseconds === milliseconds,
    );
    if (!pending) {
      throw new Error(`No pending ${milliseconds}ms sleep.`);
    }
    pending.resolve();
  }
}

class FakeSocket implements WebSocketFacade {
  protocol: string = REALTIME_WEBSOCKET_PROTOCOL;
  readyState = 0;
  bufferedAmount = 0;
  readonly sent: string[] = [];
  readonly closes: Array<Readonly<{ code?: number; reason?: string }>> = [];
  readonly listeners = new Map<
    string,
    Set<(event: unknown) => void>
  >();

  send(data: string): void {
    if (this.readyState !== 1) throw new Error("not open");
    this.sent.push(data);
  }

  close(code?: number, reason?: string): void {
    this.closes.push({ code, reason });
    this.readyState = 3;
  }

  addEventListener(
    type: "open" | "message" | "close" | "error",
    listener: (event: unknown) => void,
  ): void {
    const listeners = this.listeners.get(type) ?? new Set();
    listeners.add(listener);
    this.listeners.set(type, listeners);
  }

  removeEventListener(
    type: "open" | "message" | "close" | "error",
    listener: (event: unknown) => void,
  ): void {
    this.listeners.get(type)?.delete(listener);
  }

  open(): void {
    this.readyState = 1;
    this.emit("open", {});
  }

  message(value: unknown): void {
    this.emit("message", { data: value });
  }

  nativeClose(code: number, reason = "private provider text"): void {
    this.readyState = 3;
    this.emit("close", { code, reason });
  }

  private emit(type: string, event: unknown): void {
    for (const listener of [...(this.listeners.get(type) ?? [])]) {
      listener(event);
    }
  }
}

function welcome(
  overrides: Partial<{
    heartbeatMs: number;
    heartbeatAckTimeoutMs: number;
    limits: WebSocketAdvertisedLimits;
  }> = {},
): string {
  return JSON.stringify({
    type: "WELCOME",
    protocol: REALTIME_WEBSOCKET_PROTOCOL,
    connectionId: "connection.0001",
    heartbeatMs: 5_000,
    heartbeatAckTimeoutMs: 1_000,
    limits: LIMITS,
    ...overrides,
  });
}

function serverFrame(value: Record<string, unknown>): string {
  return JSON.stringify({
    protocol: REALTIME_WEBSOCKET_PROTOCOL,
    ...value,
  });
}

function setup(
  overrides: Partial<WebSocketConnectionDependencies> = {},
) {
  const socket = new FakeSocket();
  const clock = new ManualClock();
  const recoveries: WebSocketRecoveryRequest[] = [];
  const observations: unknown[] = [];
  const createSocket = vi.fn(() => socket);
  const onEvent = vi.fn(async () =>
    realtimeSuccess(
      Object.freeze({ kind: "CONTINUE" as const }),
    ),
  );
  const connection = createWebSocketConnection({
    origin: "https://app.example.test",
    endpointPath: "/_realtime/v1",
    clock,
    createNonce: () => "nonce.0001",
    createSocket,
    onEvent,
    onRecoveryRequired: (request) => recoveries.push(request),
    observe: (observation) => observations.push(observation),
    ...overrides,
  });
  return {
    connection,
    socket,
    clock,
    recoveries,
    observations,
    createSocket,
    onEvent,
  };
}

async function openConnection(
  runtime: ReturnType<typeof setup>,
): Promise<void> {
  const opened = runtime.connection.connect();
  runtime.socket.open();
  runtime.socket.message(welcome());
  await expect(opened).resolves.toMatchObject({
    ok: true,
    value: { protocol: REALTIME_WEBSOCKET_PROTOCOL },
  });
}

async function flush(): Promise<void> {
  await Promise.resolve();
  await Promise.resolve();
}

describe("WebSocket connection generation", () => {
  it("uses only a fixed same-origin wss endpoint and exact subprotocol", async () => {
    const runtime = setup();
    await openConnection(runtime);

    expect(runtime.createSocket).toHaveBeenCalledWith(
      "wss://app.example.test/_realtime/v1",
      REALTIME_WEBSOCKET_PROTOCOL,
    );
    expect(runtime.connection.inspect()).toMatchObject({
      status: "OPEN",
      subscriptionCount: 0,
    });

    const closed = runtime.connection.waitClosed();
    runtime.connection.close();
    runtime.connection.close();
    await expect(closed).resolves.toEqual({
      category: "NORMAL",
      error: {
        kind: "CLOSED",
        operation: "CLOSE",
        retryable: false,
      },
      recovery: null,
    });
    expect(runtime.socket.closes).toEqual([
      { code: 1_000, reason: undefined },
    ]);
    await expect(runtime.connection.connect()).resolves.toMatchObject({
      ok: false,
      error: { kind: "CLOSED", operation: "CONNECT" },
    });
  });

  it("rejects configuration that weakens hard timing ceilings", () => {
    expect(() =>
      setup({ ceilings: { connectTimeoutMs: 30_001 } }),
    ).toThrow("WebSocket ceiling is invalid");
    expect(() =>
      setup({ ceilings: { minHeartbeatMs: 4_999 } }),
    ).toThrow("WebSocket ceiling is invalid");
    expect(() =>
      setup({ ceilings: { maxApplyMs: 30_001 } }),
    ).toThrow("WebSocket ceiling is invalid");
  });

  it("serializes subscribe and unsubscribe through one outbound queue", async () => {
    const runtime = setup({
      ceilings: { maxOutboundQueueCount: 2 },
    });
    await openConnection(runtime);

    expect(
      runtime.connection.subscribe({
        subscriptionId: "subscription.queued",
        streamId: "presence.v1",
        scopeBinding: "scope.0001",
        stateBearing: false,
      }),
    ).toMatchObject({ ok: true });
    expect(
      runtime.connection.unsubscribe("subscription.queued"),
    ).toMatchObject({ ok: true });
    expect(runtime.connection.inspect()).toMatchObject({
      status: "OPEN",
      outboundQueueCount: 2,
    });
    expect(runtime.socket.sent).toEqual([]);

    await flush();

    expect(
      runtime.socket.sent.map(
        (frame) =>
          (JSON.parse(frame) as Readonly<{ type: string }>).type,
      ),
    ).toEqual(["SUBSCRIBE", "UNSUBSCRIBE"]);
    expect(runtime.connection.inspect()).toMatchObject({
      status: "OPEN",
      outboundQueueCount: 0,
      outboundQueueBytes: 0,
    });
    runtime.connection.close();
  });

  it("fails closed when the negotiated outbound message queue overflows", async () => {
    const runtime = setup({
      ceilings: { maxOutboundQueueCount: 2 },
    });
    await openConnection(runtime);
    const closed = runtime.connection.waitClosed();

    for (const suffix of ["one", "two"]) {
      expect(
        runtime.connection.subscribe({
          subscriptionId: `subscription.${suffix}`,
          streamId: "presence.v1",
          scopeBinding: "scope.0001",
          stateBearing: false,
        }),
      ).toMatchObject({ ok: true });
    }
    expect(runtime.connection.inspect().outboundQueueCount).toBe(2);
    expect(
      runtime.connection.subscribe({
        subscriptionId: "subscription.overflow",
        streamId: "presence.v1",
        scopeBinding: "scope.0001",
        stateBearing: false,
      }),
    ).toEqual({
      ok: false,
      error: {
        kind: "QUEUE_OVERFLOW",
        operation: "SUBSCRIBE",
        retryable: false,
      },
    });

    await expect(closed).resolves.toEqual({
      category: "OVERLOADED",
      error: {
        kind: "QUEUE_OVERFLOW",
        operation: "SUBSCRIBE",
        retryable: false,
      },
      recovery: null,
    });
    expect(runtime.recoveries).toEqual([
      {
        subscriptionId: null,
        reason: "QUEUE_OVERFLOW",
      },
    ]);
    expect(runtime.connection.inspect()).toMatchObject({
      status: "CLOSED",
      subscriptionCount: 0,
      outboundQueueCount: 0,
      outboundQueueBytes: 0,
    });
    await flush();
    expect(runtime.socket.sent).toEqual([]);
  });

  it("rejects a wrong negotiated subprotocol before accepting WELCOME", async () => {
    const runtime = setup();
    runtime.socket.protocol = "";
    const opened = runtime.connection.connect();

    runtime.socket.open();

    await expect(opened).resolves.toEqual({
      ok: false,
      error: {
        kind: "PROTOCOL_MISMATCH",
        operation: "CONNECT",
        retryable: false,
      },
    });
    expect(runtime.connection.inspect().status).toBe("CLOSED");
    expect(runtime.recoveries).toEqual([
      {
        subscriptionId: null,
        reason: "PROTOCOL_MISMATCH",
      },
    ]);
  });

  it("requires a checkpoint and rejects silent resume advancement", async () => {
    const runtime = setup();
    await openConnection(runtime);

    expect(
      runtime.connection.subscribe({
        subscriptionId: "subscription.no-checkpoint",
        streamId: "orders.v1",
        scopeBinding: "scope.0001",
        stateBearing: true,
      }),
    ).toMatchObject({
      ok: false,
      error: { kind: "PROTOCOL_MISMATCH" },
    });

    expect(
      runtime.connection.subscribe({
        subscriptionId: "subscription.0001",
        streamId: "orders.v1",
        scopeBinding: "scope.0001",
        stateBearing: true,
        checkpoint: {
          streamEpoch: "stream-epoch.0001",
          lastAppliedSequence: "10",
          cursor: "cursor.0010",
        },
      }),
    ).toEqual({
      ok: true,
      value: { acceptedLocally: true },
    });
    await flush();
    expect(JSON.parse(runtime.socket.sent.at(-1) ?? "")).toMatchObject({
      type: "SUBSCRIBE",
      cursor: "cursor.0010",
    });

    runtime.socket.message(
      serverFrame({
        type: "SUBSCRIBED",
        subscriptionId: "subscription.0001",
        streamEpoch: "stream-epoch.0001",
        acceptedCursor: "cursor.0011",
        nextExpectedSequence: "12",
      }),
    );
    await flush();

    expect(runtime.connection.inspect().status).toBe("CLOSED");
    expect(runtime.recoveries).toContainEqual({
      subscriptionId: "subscription.0001",
      reason: "SEQUENCE_GAP",
    });
    expect(runtime.onEvent).not.toHaveBeenCalled();
  });

  it("holds queued events behind the validated SUBSCRIBED barrier gate", async () => {
    let release:
      | ((result: RealtimeResult<void>) => void)
      | undefined;
    const barrier = new Promise<RealtimeResult<void>>((resolve) => {
      release = resolve;
    });
    const onSubscribed = vi.fn(() => barrier);
    const runtime = setup({ onSubscribed });
    await openConnection(runtime);
    runtime.connection.subscribe({
      subscriptionId: "subscription.0001",
      streamId: "orders.v1",
      scopeBinding: "scope.0001",
      stateBearing: true,
      checkpoint: {
        streamEpoch: "stream-epoch.0001",
        lastAppliedSequence: "10",
        cursor: "cursor.0010",
      },
    });
    runtime.socket.message(
      serverFrame({
        type: "SUBSCRIBED",
        subscriptionId: "subscription.0001",
        streamEpoch: "stream-epoch.0001",
        acceptedCursor: "cursor.0010",
        nextExpectedSequence: "11",
      }),
    );
    runtime.socket.message(
      serverFrame({
        type: "EVENT",
        subscriptionId: "subscription.0001",
        envelope: {
          protocol: "REALTIME_EVENT_V1",
          streamId: "orders.v1",
        },
      }),
    );

    await vi.waitFor(() =>
      expect(onSubscribed).toHaveBeenCalledOnce(),
    );
    expect(onSubscribed).toHaveBeenCalledWith(
      {
        subscriptionId: "subscription.0001",
        streamId: "orders.v1",
        streamEpoch: "stream-epoch.0001",
        acceptedCursor: "cursor.0010",
        nextExpectedSequence: "11",
      },
      expect.any(AbortSignal),
    );
    expect(runtime.onEvent).not.toHaveBeenCalled();

    release?.(realtimeSuccess(undefined));
    await vi.waitFor(() =>
      expect(runtime.onEvent).toHaveBeenCalledOnce(),
    );
  });

  it("propagates an exact SUBSCRIBED barrier failure before event effects", async () => {
    const barrierFailure = realtimeFailure(
      "SCOPE_FENCED",
      "RECOVER",
      false,
    );
    const runtime = setup({
      onSubscribed: () => barrierFailure,
    });
    await openConnection(runtime);
    runtime.connection.subscribe({
      subscriptionId: "subscription.0001",
      streamId: "orders.v1",
      scopeBinding: "scope.0001",
      stateBearing: true,
      checkpoint: {
        streamEpoch: "stream-epoch.0001",
        lastAppliedSequence: "10",
        cursor: "cursor.0010",
      },
    });
    const closed = runtime.connection.waitClosed();

    runtime.socket.message(
      serverFrame({
        type: "SUBSCRIBED",
        subscriptionId: "subscription.0001",
        streamEpoch: "stream-epoch.0001",
        acceptedCursor: "cursor.0010",
        nextExpectedSequence: "11",
      }),
    );
    const receipt = await closed;

    expect(receipt.category).toBe("CURSOR_RESET");
    expect(receipt.error).toBe(barrierFailure.error);
    expect(receipt.recovery).toBeNull();
    expect(runtime.onEvent).not.toHaveBeenCalled();
    expect(runtime.recoveries).toContainEqual({
      subscriptionId: "subscription.0001",
      reason: "SCOPE_CHANGED",
    });
  });

  it("serializes active subscription events and closes on bounded queue overflow", async () => {
    let releaseFirst: (() => void) | undefined;
    const firstEffect = new Promise<
      RealtimeResult<WebSocketInboundEventOutcome>
    >((resolve) => {
      releaseFirst = () =>
        resolve(
          realtimeSuccess(
            Object.freeze({ kind: "CONTINUE" }),
          ),
        );
    });
    const onEvent = vi
      .fn<
        (
          frame: unknown,
          signal: AbortSignal,
        ) => Promise<
          RealtimeResult<WebSocketInboundEventOutcome>
        >
      >()
      .mockImplementationOnce(() => firstEffect)
      .mockResolvedValue(
        realtimeSuccess(Object.freeze({ kind: "CONTINUE" })),
      );
    const runtime = setup({
      onEvent,
      ceilings: { maxInboundQueueCount: 1 },
    });
    await openConnection(runtime);
    runtime.connection.subscribe({
      subscriptionId: "subscription.0001",
      streamId: "orders.v1",
      scopeBinding: "scope.0001",
      stateBearing: true,
      checkpoint: {
        streamEpoch: "stream-epoch.0001",
        lastAppliedSequence: "10",
        cursor: "cursor.0010",
      },
    });
    runtime.socket.message(
      serverFrame({
        type: "SUBSCRIBED",
        subscriptionId: "subscription.0001",
        streamEpoch: "stream-epoch.0001",
        acceptedCursor: "cursor.0010",
        nextExpectedSequence: "11",
      }),
    );
    await flush();

    const event = serverFrame({
      type: "EVENT",
      subscriptionId: "subscription.0001",
      envelope: {
        protocol: "REALTIME_EVENT_V1",
        streamId: "orders.v1",
      },
    });
    runtime.socket.message(event);
    runtime.socket.message(event);
    runtime.socket.message(event);
    await flush();

    expect(onEvent).toHaveBeenCalledTimes(1);
    expect(runtime.connection.inspect().status).toBe("CLOSED");
    expect(runtime.recoveries).toContainEqual({
      subscriptionId: null,
      reason: "QUEUE_OVERFLOW",
    });
    releaseFirst?.();
  });

  it("rejects an event routed through the wrong logical subscription", async () => {
    const runtime = setup();
    await openConnection(runtime);
    runtime.connection.subscribe({
      subscriptionId: "subscription.0001",
      streamId: "orders.v1",
      scopeBinding: "scope.0001",
      stateBearing: false,
    });
    runtime.socket.message(
      serverFrame({
        type: "SUBSCRIBED",
        subscriptionId: "subscription.0001",
        streamEpoch: "stream-epoch.0001",
        acceptedCursor: null,
        nextExpectedSequence: "0",
      }),
    );
    await flush();

    runtime.socket.message(
      serverFrame({
        type: "EVENT",
        subscriptionId: "subscription.0001",
        envelope: {
          protocol: "REALTIME_EVENT_V1",
          streamId: "another-stream.v1",
        },
      }),
    );
    await flush();

    expect(runtime.onEvent).not.toHaveBeenCalled();
    expect(runtime.connection.inspect().status).toBe("CLOSED");
    expect(runtime.recoveries).toContainEqual({
      subscriptionId: null,
      reason: "PROTOCOL_MISMATCH",
    });
  });

  it("keeps an unsubscribe tombstone until ACK, ignores late frames, and then permits ID reuse", async () => {
    const runtime = setup();
    await openConnection(runtime);
    runtime.connection.subscribe({
      subscriptionId: "subscription.0001",
      streamId: "presence.v1",
      scopeBinding: "scope.0001",
      stateBearing: false,
    });
    runtime.socket.message(
      serverFrame({
        type: "SUBSCRIBED",
        subscriptionId: "subscription.0001",
        streamEpoch: "stream-epoch.0001",
        acceptedCursor: null,
        nextExpectedSequence: "0",
      }),
    );
    await flush();

    expect(
      runtime.connection.unsubscribe("subscription.0001"),
    ).toMatchObject({ ok: true });
    expect(runtime.connection.inspect().subscriptionCount).toBe(0);
    expect(
      runtime.connection.subscribe({
        subscriptionId: "subscription.0001",
        streamId: "replacement.v1",
        scopeBinding: "scope.0001",
        stateBearing: false,
      }),
    ).toMatchObject({
      ok: false,
      error: { kind: "PROTOCOL_MISMATCH" },
    });
    runtime.socket.message(
      serverFrame({
        type: "EVENT",
        subscriptionId: "subscription.0001",
        envelope: {
          protocol: "REALTIME_EVENT_V1",
          streamId: "presence.v1",
        },
      }),
    );
    runtime.socket.message(
      serverFrame({
        type: "RESET_REQUIRED",
        subscriptionId: "subscription.0001",
        reason: "SERVER_RESET",
      }),
    );
    runtime.socket.message(
      serverFrame({
        type: "SUBSCRIBED",
        subscriptionId: "subscription.0001",
        streamEpoch: "late-stream-epoch.0001",
        acceptedCursor: null,
        nextExpectedSequence: "0",
      }),
    );
    await flush();

    expect(runtime.connection.inspect().status).toBe("OPEN");
    expect(runtime.onEvent).not.toHaveBeenCalled();
    expect(runtime.recoveries).toEqual([]);
    runtime.socket.message(
      serverFrame({
        type: "UNSUBSCRIBED",
        subscriptionId: "subscription.0001",
      }),
    );
    await flush();

    expect(
      runtime.connection.subscribe({
        subscriptionId: "subscription.0001",
        streamId: "replacement.v1",
        scopeBinding: "scope.0001",
        stateBearing: false,
      }),
    ).toEqual({
      ok: true,
      value: { acceptedLocally: true },
    });
    runtime.connection.close();
  });

  it("reclaims subscription quota after 32 subscribe-unsubscribe acknowledgements", async () => {
    const runtime = setup();
    await openConnection(runtime);

    for (let index = 0; index < 32; index += 1) {
      const subscriptionId = `subscription.${String(index).padStart(4, "0")}`;
      expect(
        runtime.connection.subscribe({
          subscriptionId,
          streamId: "presence.v1",
          scopeBinding: "scope.0001",
          stateBearing: false,
        }),
      ).toMatchObject({ ok: true });
      expect(
        runtime.connection.unsubscribe(subscriptionId),
      ).toMatchObject({ ok: true });
      runtime.socket.message(
        serverFrame({
          type: "UNSUBSCRIBED",
          subscriptionId,
        }),
      );
      await flush();
    }

    expect(runtime.connection.inspect()).toMatchObject({
      status: "OPEN",
      subscriptionCount: 0,
    });
    expect(
      runtime.connection.subscribe({
        subscriptionId: "subscription.new",
        streamId: "presence.v1",
        scopeBinding: "scope.0001",
        stateBearing: false,
      }),
    ).toEqual({
      ok: true,
      value: { acceptedLocally: true },
    });
    runtime.connection.close();
  });

  it("bounds the UNSUBSCRIBED acknowledgement and cancels its deadline on ACK", async () => {
    const missing = setup();
    await openConnection(missing);
    missing.connection.subscribe({
      subscriptionId: "subscription.missing-ack",
      streamId: "presence.v1",
      scopeBinding: "scope.0001",
      stateBearing: false,
    });
    missing.connection.unsubscribe("subscription.missing-ack");
    const closed = missing.connection.waitClosed();
    missing.clock.resolveDuration(30_000);
    await expect(closed).resolves.toMatchObject({
      category: "NETWORK_LOST",
      error: {
        kind: "IDLE_TIMEOUT",
        operation: "SUBSCRIBE",
        retryable: true,
      },
    });

    const acknowledged = setup();
    await openConnection(acknowledged);
    acknowledged.connection.subscribe({
      subscriptionId: "subscription.acknowledged",
      streamId: "presence.v1",
      scopeBinding: "scope.0001",
      stateBearing: false,
    });
    acknowledged.connection.unsubscribe(
      "subscription.acknowledged",
    );
    const acknowledgementDeadline = acknowledged.clock.sleeps.find(
      (sleep) => sleep.milliseconds === 30_000 && !sleep.settled,
    );
    expect(acknowledgementDeadline).toBeDefined();
    acknowledged.socket.message(
      serverFrame({
        type: "UNSUBSCRIBED",
        subscriptionId: "subscription.acknowledged",
      }),
    );
    await flush();
    expect(acknowledgementDeadline?.settled).toBe(true);
    expect(acknowledged.connection.inspect().status).toBe("OPEN");
    acknowledged.connection.close();
  });

  it.each([
    {
      name: "unknown",
      prepare: (_runtime: ReturnType<typeof setup>) => undefined,
      subscriptionId: "subscription.unknown",
    },
    {
      name: "non-unsubscribing",
      prepare: async (runtime: ReturnType<typeof setup>) => {
        runtime.connection.subscribe({
          subscriptionId: "subscription.active",
          streamId: "presence.v1",
          scopeBinding: "scope.0001",
          stateBearing: false,
        });
        runtime.socket.message(
          serverFrame({
            type: "SUBSCRIBED",
            subscriptionId: "subscription.active",
            streamEpoch: "stream-epoch.0001",
            acceptedCursor: null,
            nextExpectedSequence: "0",
          }),
        );
        await flush();
      },
      subscriptionId: "subscription.active",
    },
  ])("fails closed for an $name UNSUBSCRIBED acknowledgement", async ({
    prepare,
    subscriptionId,
  }) => {
    const runtime = setup();
    await openConnection(runtime);
    await prepare(runtime);
    const closed = runtime.connection.waitClosed();

    runtime.socket.message(
      serverFrame({
        type: "UNSUBSCRIBED",
        subscriptionId,
      }),
    );

    await expect(closed).resolves.toMatchObject({
      category: "PROTOCOL_MISMATCH",
      error: {
        kind: "PROTOCOL_MISMATCH",
        retryable: false,
      },
    });
    expect(runtime.recoveries).toContainEqual({
      subscriptionId: null,
      reason: "PROTOCOL_MISMATCH",
    });
  });

  it.each([
    { category: "OVERLOADED" as const, retryable: false },
    { category: "RESTART" as const, retryable: true },
    { category: "NETWORK_LOST" as const, retryable: true },
  ])(
    "classifies $category without inventing a server retry hint",
    async ({ category, retryable }) => {
      const runtime = setup();
      await openConnection(runtime);
      const closed = runtime.connection.waitClosed();

      runtime.socket.message(
        serverFrame({
          type: "CLOSE",
          category,
        }),
      );

      await expect(closed).resolves.toMatchObject({
        category,
        error: {
          kind: "PROVIDER_UNAVAILABLE",
          operation: "RECEIVE",
          retryable,
        },
      });
    },
  );

  it("propagates the canonical consumer failure through the terminal channel", async () => {
    const runtime = setup({
      onEvent: vi.fn(async () =>
        realtimeFailure("FORBIDDEN", "RECEIVE"),
      ),
    });
    await openConnection(runtime);
    runtime.connection.subscribe({
      subscriptionId: "subscription.0001",
      streamId: "presence.v1",
      scopeBinding: "scope.0001",
      stateBearing: false,
    });
    runtime.socket.message(
      serverFrame({
        type: "SUBSCRIBED",
        subscriptionId: "subscription.0001",
        streamEpoch: "stream-epoch.0001",
        acceptedCursor: null,
        nextExpectedSequence: "0",
      }),
    );
    await flush();
    const closed = runtime.connection.waitClosed();
    runtime.socket.message(
      serverFrame({
        type: "EVENT",
        subscriptionId: "subscription.0001",
        envelope: {
          protocol: "REALTIME_EVENT_V1",
          streamId: "presence.v1",
        },
      }),
    );
    await flush();

    await expect(closed).resolves.toEqual({
      category: "FORBIDDEN",
      error: {
        kind: "FORBIDDEN",
        operation: "RECEIVE",
        retryable: false,
      },
      recovery: null,
    });
    expect(runtime.recoveries).toEqual([]);
  });

  it("closes the old generation after authority recovery commits", async () => {
    const runtime = setup({
      onEvent: vi.fn(async () =>
        realtimeSuccess(RECOVERY_OUTCOME),
      ),
    });
    await openConnection(runtime);
    runtime.connection.subscribe({
      subscriptionId: "subscription.0001",
      streamId: "presence.v1",
      scopeBinding: "scope.0001",
      stateBearing: false,
    });
    runtime.socket.message(
      serverFrame({
        type: "SUBSCRIBED",
        subscriptionId: "subscription.0001",
        streamEpoch: "stream-epoch.0001",
        acceptedCursor: null,
        nextExpectedSequence: "0",
      }),
    );
    await flush();
    const closed = runtime.connection.waitClosed();
    runtime.socket.message(
      serverFrame({
        type: "EVENT",
        subscriptionId: "subscription.0001",
        envelope: {
          protocol: "REALTIME_EVENT_V1",
          streamId: "presence.v1",
        },
      }),
    );
    await flush();

    await expect(closed).resolves.toEqual({
      category: "CURSOR_RESET",
      error: {
        kind: "CURSOR_EXPIRED",
        operation: "RECEIVE",
        retryable: false,
      },
      recovery: RECOVERY_OUTCOME,
    });
    expect(runtime.connection.inspect().status).toBe("CLOSED");
    expect(runtime.recoveries).toEqual([]);
    expect(runtime.observations).toContainEqual({
      kind: "CLOSED",
      failureKind: "CURSOR_EXPIRED",
      closeCategory: "CURSOR_RESET",
    });
  });

  it("uses bounded application heartbeat and never exposes native close reason", async () => {
    const runtime = setup();
    await openConnection(runtime);

    runtime.clock.resolveDuration(5_000);
    await flush();
    const heartbeat = JSON.parse(runtime.socket.sent.at(-1) ?? "");
    expect(heartbeat).toMatchObject({
      type: "HEARTBEAT",
      nonce: "nonce.0001",
    });
    runtime.socket.message(
      serverFrame({
        type: "HEARTBEAT_ACK",
        nonce: "nonce.0001",
      }),
    );
    await flush();
    runtime.clock.resolveDuration(1_000);
    await flush();
    expect(runtime.connection.inspect().status).toBe("OPEN");

    runtime.socket.nativeClose(4_001, "token=do-not-observe");
    expect(runtime.connection.inspect().status).toBe("CLOSED");
    expect(JSON.stringify(runtime.observations)).not.toContain("token=");
    expect(runtime.observations).toContainEqual({
      kind: "CLOSED",
      failureKind: "AUTH_REQUIRED",
      closeCategory: "AUTH_REQUIRED",
    });
  });

  it("handles heartbeat acknowledgements outside the application-effect queue", async () => {
    let releaseEffect: (() => void) | undefined;
    const runtime = setup({
      onEvent: () =>
        new Promise((resolve) => {
          releaseEffect = () =>
            resolve(
              realtimeSuccess(
                Object.freeze({ kind: "CONTINUE" as const }),
              ),
            );
        }),
    });
    await openConnection(runtime);
    runtime.connection.subscribe({
      subscriptionId: "subscription.0001",
      streamId: "presence.v1",
      scopeBinding: "scope.0001",
      stateBearing: false,
    });
    runtime.socket.message(
      serverFrame({
        type: "SUBSCRIBED",
        subscriptionId: "subscription.0001",
        streamEpoch: "stream-epoch.0001",
        acceptedCursor: null,
        nextExpectedSequence: "0",
      }),
    );
    await flush();
    runtime.socket.message(
      serverFrame({
        type: "EVENT",
        subscriptionId: "subscription.0001",
        envelope: {
          protocol: "REALTIME_EVENT_V1",
          streamId: "presence.v1",
        },
      }),
    );
    await flush();

    runtime.clock.resolveDuration(5_000);
    await flush();
    runtime.socket.message(
      serverFrame({
        type: "HEARTBEAT_ACK",
        nonce: "nonce.0001",
      }),
    );
    runtime.clock.resolveDuration(1_000);
    await flush();

    expect(runtime.connection.inspect()).toMatchObject({
      status: "OPEN",
      awaitingHeartbeatAck: false,
      inboundQueueCount: 0,
    });
    releaseEffect?.();
    await flush();
    runtime.connection.close();
  });

  it("fails closed when heartbeat eligibility throws", async () => {
    const runtime = setup({
      heartbeatEligible() {
        throw new Error("broken visibility provider");
      },
    });
    await openConnection(runtime);
    const closed = runtime.connection.waitClosed();

    runtime.clock.resolveDuration(5_000);
    await flush();

    await expect(closed).resolves.toMatchObject({
      category: "NETWORK_LOST",
      error: {
        kind: "PROVIDER_UNAVAILABLE",
        operation: "RECEIVE",
      },
    });
  });
});