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

import type { ClockPort } from "../../../src/application/ports/clock-port.ts";
import type {
  RealtimeFailureKind,
  RealtimeResult,
} from "../../../src/application/ports/realtime/shared.ts";
import type {
  RealtimeRecoveryCheckpoint,
} from "../../../src/application/ports/realtime/event-authority.ts";
import {
  createRealtimeReconnectCoordinator,
  type RealtimeCommittedRecovery,
  type RealtimeReconnectAttemptContext,
  type RealtimeReconnectAttemptSuccess,
  type RealtimeReconnectEnvironment,
  type RealtimeReconnectOutcome,
  type RealtimeReconnectSession,
  type RealtimeRecoveryReconnectDirective,
} from "../../../src/adapters/realtime/index.ts";
import {
  defineReconnectPolicy,
  REALTIME_RECONNECT_CEILINGS,
  type ReconnectPolicy,
} from "../../../src/adapters/realtime/reconnect-policy.ts";
import {
  realtimeFailure,
  realtimeSuccess,
} from "../../../src/adapters/realtime/result.ts";

type Sleeper = Readonly<{
  dueAt: number;
  signal?: AbortSignal;
  resolve(): void;
  reject(): void;
}>;

class ManualClock implements ClockPort {
  current = 0;
  readonly sleepCalls: number[] = [];
  readonly sleepers: Sleeper[] = [];

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

  sleep(
    milliseconds: number,
    signal?: AbortSignal,
  ): Promise<void> {
    this.sleepCalls.push(milliseconds);
    return new Promise((resolve, reject) => {
      if (signal?.aborted) {
        reject(new Error("Aborted."));
        return;
      }
      let onAbort: (() => void) | undefined;
      const sleeper: Sleeper = {
        dueAt: this.current + milliseconds,
        signal,
        resolve: () => {
          signal?.removeEventListener("abort", onAbort!);
          resolve();
        },
        reject: () => {
          signal?.removeEventListener("abort", onAbort!);
          reject(new Error("Aborted."));
        },
      };
      onAbort = () => {
        this.remove(sleeper);
        sleeper.reject();
      };
      signal?.addEventListener("abort", onAbort, { once: true });
      this.sleepers.push(sleeper);
    });
  }

  advance(milliseconds: number): void {
    this.current += milliseconds;
    const ready = this.sleepers
      .filter((sleeper) => sleeper.dueAt <= this.current)
      .sort((left, right) => left.dueAt - right.dueAt);
    for (const sleeper of ready) {
      this.remove(sleeper);
      sleeper.resolve();
    }
  }

  private remove(target: Sleeper): void {
    const index = this.sleepers.indexOf(target);
    if (index >= 0) this.sleepers.splice(index, 1);
  }
}

class MutableOnlineEnvironment
  implements RealtimeReconnectEnvironment
{
  readonly listeners = new Set<(online: boolean) => void>();

  constructor(public connected: boolean) {}

  online(): boolean {
    return this.connected;
  }

  subscribeOnline(
    listener: (online: boolean) => void,
  ): () => void {
    this.listeners.add(listener);
    return () => this.listeners.delete(listener);
  }

  emit(online: boolean): void {
    this.connected = online;
    for (const listener of this.listeners) listener(online);
  }
}

function policy(
  override: Partial<ReconnectPolicy> = {},
): ReconnectPolicy {
  return defineReconnectPolicy({
    baseDelayMs: 100,
    maxDelayMs: 1_000,
    maxAttempts: 2,
    maxElapsedMs: 5_000,
    stableOpenMs: 500,
    ...override,
  });
}

function failed(
  result: Extract<RealtimeResult<never>, { ok: false }>,
  serverNotBeforeMs?: number,
): RealtimeReconnectOutcome<never> {
  return Object.freeze({
    result,
    ...(serverNotBeforeMs === undefined
      ? {}
      : { serverNotBeforeMs }),
  });
}

function opened<ClosedReceipt>(
  session: RealtimeReconnectSession<ClosedReceipt>,
  establishedRecoveryBarrier?: RealtimeCommittedRecovery,
): RealtimeReconnectOutcome<
  RealtimeReconnectAttemptSuccess<ClosedReceipt>
> {
  return Object.freeze({
    result: realtimeSuccess(
      Object.freeze({
        session,
        ...(establishedRecoveryBarrier === undefined
          ? {}
          : { establishedRecoveryBarrier }),
      }),
    ),
  });
}

function closedSuccessfully(): RealtimeReconnectOutcome<void> {
  return Object.freeze({
    result: realtimeSuccess(undefined),
  });
}

function confirmationSucceeded(): RealtimeResult<void> {
  return realtimeSuccess(undefined);
}

function recoveryToken(
  sequence = "7",
): RealtimeCommittedRecovery {
  const checkpoint = Object.freeze({
    lastAppliedSequence: sequence,
    recoveryMode: "CURSOR",
    resumeCursor: `cursor-${sequence}`,
    streamEpoch: "epoch-1",
  }) as unknown as RealtimeRecoveryCheckpoint;
  return Object.freeze({
    kind: "RECOVERY_COMMITTED",
    streamId: "orders",
    checkpoint,
  }) as unknown as RealtimeCommittedRecovery;
}

function recoveryDirective(
  recovery: RealtimeCommittedRecovery,
  terminalResult: Extract<
    RealtimeResult<never>,
    { ok: false }
  >,
  serverNotBeforeMs?: number,
): RealtimeRecoveryReconnectDirective {
  return Object.freeze({
    kind: "RECOVERY_RECONNECT",
    recovery,
    terminalResult,
    ...(serverNotBeforeMs === undefined
      ? {}
      : { serverNotBeforeMs }),
  });
}

function deferred<Value>(): Readonly<{
  promise: Promise<Value>;
  resolve(value: Value): void;
}> {
  let resolvePromise: ((value: Value) => void) | undefined;
  const promise = new Promise<Value>((resolve) => {
    resolvePromise = resolve;
  });
  return Object.freeze({
    promise,
    resolve(value) {
      resolvePromise?.(value);
    },
  });
}

async function flush(): Promise<void> {
  for (let turn = 0; turn < 8; turn += 1) {
    await Promise.resolve();
  }
}

async function flushUntil(
  condition: () => boolean,
): Promise<void> {
  for (let turn = 0; turn < 50; turn += 1) {
    if (condition()) return;
    await Promise.resolve();
  }
}

describe("realtime reconnect coordinator", () => {
  it("owns one bounded retry loop and returns the exact last failure", async () => {
    const clock = new ManualClock();
    const environment = new MutableOnlineEnvironment(true);
    const failures = [
      realtimeFailure("CONNECT_TIMEOUT", "CONNECT", true),
      realtimeFailure("PROVIDER_UNAVAILABLE", "CONNECT", true),
      realtimeFailure("IDLE_TIMEOUT", "RECEIVE", true),
    ] as const;
    let attempts = 0;
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment,
      clock,
      random: () => 0.5,
      confirmTransportBarrier: confirmationSucceeded,
      attempt: () => {
        const next = failures[attempts++]!;
        return failed(
          next,
          next === failures[1] ? 0 : undefined,
        );
      },
    });

    const running = coordinator.run();
    await flush();
    expect(attempts).toBe(1);
    expect(clock.sleepCalls).toEqual([50]);

    clock.advance(50);
    await flush();
    expect(attempts).toBe(2);
    expect(clock.sleepCalls).toEqual([50, 100]);

    clock.advance(100);
    const result = await running;

    expect(attempts).toBe(3);
    expect(result).toBe(failures[2]);
    expect(coordinator.getState()).toBe("IDLE");
  });

  it("does not clamp a server not-before value past elapsed budget", async () => {
    const clock = new ManualClock();
    const terminal = realtimeFailure(
      "RATE_LIMITED",
      "CONNECT",
      true,
    );
    const random = vi.fn(() => 0);
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy({
        maxAttempts: 1,
        maxElapsedMs: 500,
      }),
      environment: new MutableOnlineEnvironment(true),
      clock,
      random,
      confirmTransportBarrier: confirmationSucceeded,
      attempt: () => failed(terminal, 500),
    });

    const result = await coordinator.run();

    expect(result).toBe(terminal);
    expect(clock.sleepCalls).toEqual([]);
    expect(random).not.toHaveBeenCalled();
  });

  it.each([
    "RATE_LIMITED",
    "PROVIDER_UNAVAILABLE",
  ] as const)(
    "does not reconnect an external %s attempt failure without a server hint",
    async (kind) => {
      const clock = new ManualClock();
      const terminal = realtimeFailure(
        kind,
        "CONNECT",
        true,
      );
      const attempt = vi.fn(() => failed(terminal));
      const coordinator = createRealtimeReconnectCoordinator({
        policy: policy(),
        environment: new MutableOnlineEnvironment(true),
        clock,
        confirmTransportBarrier: confirmationSucceeded,
        attempt,
      });

      const result = await coordinator.run();

      expect(result).toBe(terminal);
      expect(attempt).toHaveBeenCalledOnce();
      expect(clock.sleepCalls).toEqual([]);
    },
  );

  it("does not reconnect an external close failure that requires but omits a server hint", async () => {
    const clock = new ManualClock();
    const terminal = realtimeFailure(
      "PROVIDER_UNAVAILABLE",
      "RECEIVE",
      true,
    );
    const attempt = vi.fn(() =>
      opened(
        Object.freeze({
          classifyClosed: (
            receipt: RealtimeReconnectOutcome<void>,
          ) => receipt,
          close: vi.fn(),
          waitClosed: () => failed(terminal),
        }),
      ),
    );
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment: new MutableOnlineEnvironment(true),
      clock,
      confirmTransportBarrier: confirmationSucceeded,
      attempt,
    });

    const result = await coordinator.run();

    expect(result).toBe(terminal);
    expect(attempt).toHaveBeenCalledOnce();
    expect(clock.sleepCalls).toEqual([]);
  });

  it("waits for an explicit online signal without a retry timer", async () => {
    const clock = new ManualClock();
    const environment = new MutableOnlineEnvironment(false);
    const terminal = realtimeFailure(
      "FORBIDDEN",
      "CONNECT",
      false,
    );
    const attempt = vi.fn(() => failed(terminal));
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment,
      clock,
      random: () => 0.5,
      confirmTransportBarrier: confirmationSucceeded,
      attempt,
    });

    const running = coordinator.run();
    await flush();
    clock.advance(100);
    await flush();

    expect(attempt).not.toHaveBeenCalled();
    expect(clock.sleepCalls).toEqual([]);

    environment.emit(true);
    await flush();
    expect(clock.sleepCalls).toEqual([50]);
    expect(attempt).not.toHaveBeenCalled();

    clock.advance(50);
    const result = await running;
    expect(attempt).toHaveBeenCalledTimes(1);
    expect(result).toBe(terminal);
  });

  it("resets a spent retry only after policy eligibility", async () => {
    const runScenario = async (markValid: boolean) => {
      const clock = new ManualClock();
      const firstFailure = realtimeFailure(
        "CONNECT_TIMEOUT",
        "CONNECT",
        true,
      );
      const sessionFailure = realtimeFailure(
        "PROVIDER_UNAVAILABLE",
        "RECEIVE",
        true,
      );
      const terminal = realtimeFailure(
        "FORBIDDEN",
        "CONNECT",
        false,
      );
      const closeSession = vi.fn();
      let attempts = 0;
      const coordinator = createRealtimeReconnectCoordinator({
        policy: policy({ maxAttempts: 1 }),
        environment: new MutableOnlineEnvironment(true),
        clock,
        random: () => 0,
        confirmTransportBarrier: confirmationSucceeded,
        attempt: (
          context: RealtimeReconnectAttemptContext,
        ) => {
          attempts += 1;
          if (attempts === 1) return failed(firstFailure);
          if (attempts === 2) {
            if (markValid) {
              context.markValidHeartbeatOrEvent();
            }
            return opened<RealtimeReconnectOutcome<void>>(
              Object.freeze({
                classifyClosed: (
                  receipt: RealtimeReconnectOutcome<void>,
                ) => receipt,
                close: closeSession,
                waitClosed: () =>
                  failed(sessionFailure, 0),
              }),
            );
          }
          return failed(terminal);
        },
      });

      const running = coordinator.run();
      await flush();
      clock.advance(0);
      await flush();
      if (markValid) {
        await flushUntil(() => clock.sleepCalls.length === 2);
        expect(clock.sleepCalls).toEqual([0, 0]);
        clock.advance(0);
      }
      const result = await running;
      return {
        attempts,
        closeSession,
        result,
        sessionFailure,
        terminal,
      };
    };

    const unstable = await runScenario(false);
    expect(unstable.attempts).toBe(2);
    expect(unstable.result).toBe(unstable.sessionFailure);
    expect(unstable.closeSession).toHaveBeenCalledOnce();

    const eligible = await runScenario(true);
    expect(eligible.attempts).toBe(3);
    expect(eligible.result).toBe(eligible.terminal);
    expect(eligible.closeSession).toHaveBeenCalledOnce();
  });

  it("does not reconnect before the session's authoritative close result", async () => {
    const clock = new ManualClock();
    const closure =
      deferred<RealtimeReconnectOutcome<void>>();
    const terminal = realtimeFailure(
      "FORBIDDEN",
      "RECEIVE",
      false,
    );
    const closeSession = vi.fn();
    const attempt = vi.fn(() =>
      opened<RealtimeReconnectOutcome<void>>(
        Object.freeze({
          classifyClosed: (
            receipt: RealtimeReconnectOutcome<void>,
          ) => receipt,
          close: closeSession,
          waitClosed: () => closure.promise,
        }),
      ),
    );
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment: new MutableOnlineEnvironment(true),
      clock,
      confirmTransportBarrier: confirmationSucceeded,
      attempt,
    });
    let settled = false;

    const running = coordinator.run().then((result) => {
      settled = true;
      return result;
    });
    await flush();

    expect(attempt).toHaveBeenCalledOnce();
    expect(clock.sleepCalls).toEqual([]);
    expect(settled).toBe(false);
    expect(coordinator.getState()).toBe("RUNNING");

    closure.resolve(failed(terminal));
    const result = await running;
    expect(result).toBe(terminal);
    expect(closeSession).toHaveBeenCalledOnce();
  });

  it("carries an exact recovery token into the next attempt and opens only after confirmation", async () => {
    const clock = new ManualClock();
    const recovery = recoveryToken();
    const terminal = realtimeFailure(
      "CURSOR_EXPIRED",
      "RECOVER",
      false,
    );
    const directive = recoveryDirective(recovery, terminal);
    const completed = closedSuccessfully();
    const confirmation = realtimeSuccess(undefined);
    const confirmTransportBarrier = vi.fn(
      (received: RealtimeCommittedRecovery) => {
        expect(received).toBe(recovery);
        return confirmation;
      },
    );
    let attempts = 0;
    let gateSettled = false;
    let gateResult: RealtimeResult<void> | undefined;
    type RecoveryCloseReceipt =
      | RealtimeRecoveryReconnectDirective
      | RealtimeReconnectOutcome<void>;
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment: new MutableOnlineEnvironment(true),
      clock,
      random: () => 0,
      confirmTransportBarrier,
      attempt: (context) => {
        attempts += 1;
        if (attempts === 1) {
          expect(context.pendingRecovery).toBeNull();
          return opened<RecoveryCloseReceipt>(
            Object.freeze({
              classifyClosed: (
                receipt: RecoveryCloseReceipt,
              ) => receipt,
              close: vi.fn(),
              waitClosed: () => directive,
            }),
          );
        }

        expect(context.pendingRecovery).toBe(recovery);
        const gate =
          context.waitForRecoveryBarrierConfirmation();
        void gate.then((result) => {
          gateSettled = true;
          gateResult = result;
        });
        expect(gateSettled).toBe(false);
        return opened<RecoveryCloseReceipt>(
          Object.freeze({
            classifyClosed: (
              receipt: RecoveryCloseReceipt,
            ) => receipt,
            close: vi.fn(),
            waitClosed: () => completed,
          }),
          recovery,
        );
      },
    });

    const running = coordinator.run();
    await flushUntil(() => clock.sleepCalls.length === 1);
    expect(clock.sleepCalls).toEqual([0]);
    clock.advance(0);
    const result = await running;
    await flush();

    expect(attempts).toBe(2);
    expect(confirmTransportBarrier).toHaveBeenCalledOnce();
    expect(gateSettled).toBe(true);
    expect(gateResult).toBe(confirmation);
    expect(result).toBe(completed.result);
  });

  it.each([
    ["missing", undefined],
    ["cloned", recoveryToken()],
  ] as const)(
    "rejects a %s recovery proof and closes the new session",
    async (_case, proof) => {
      const clock = new ManualClock();
      const recovery = recoveryToken();
      const terminal = realtimeFailure(
        "CURSOR_EXPIRED",
        "RECOVER",
        false,
      );
      const closeSession = vi.fn();
      const waitClosed = vi.fn(closedSuccessfully);
      const confirmTransportBarrier = vi.fn(
        confirmationSucceeded,
      );
      const coordinator = createRealtimeReconnectCoordinator({
        policy: policy({ maxAttempts: 1 }),
        environment: new MutableOnlineEnvironment(true),
        clock,
        random: () => 0,
        confirmTransportBarrier,
        attempt: (context) => {
          expect(context.pendingRecovery).toBe(recovery);
          const session = Object.freeze({
            classifyClosed: (
              receipt: RealtimeReconnectOutcome<void>,
            ) => receipt,
            close: closeSession,
            waitClosed,
          });
          return proof === undefined
            ? opened(session)
            : opened(session, proof);
        },
      });

      const running = coordinator.run({
        initialRecovery: recoveryDirective(
          recovery,
          terminal,
        ),
      });
      await flushUntil(() => clock.sleepCalls.length === 1);
      clock.advance(0);
      const result = await running;

      expect(result).toEqual(
        realtimeFailure(
          "PROTOCOL_MISMATCH",
          "RECOVER",
          false,
        ),
      );
      expect(closeSession).toHaveBeenCalledOnce();
      expect(waitClosed).not.toHaveBeenCalled();
      expect(confirmTransportBarrier).not.toHaveBeenCalled();
    },
  );

  it("settles the readiness gate and run with the exact confirmation failure", async () => {
    const clock = new ManualClock();
    const recovery = recoveryToken();
    const terminal = realtimeFailure(
      "CURSOR_EXPIRED",
      "RECOVER",
      false,
    );
    const confirmationFailure = realtimeFailure(
      "SCOPE_PROTOCOL_VIOLATION",
      "RECOVER",
      false,
    );
    const closeSession = vi.fn();
    const waitClosed = vi.fn(closedSuccessfully);
    let gateResult: RealtimeResult<void> | undefined;
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy({ maxAttempts: 1 }),
      environment: new MutableOnlineEnvironment(true),
      clock,
      random: () => 0,
      confirmTransportBarrier: () => confirmationFailure,
      attempt: (context) => {
        void context
          .waitForRecoveryBarrierConfirmation()
          .then((result) => {
            gateResult = result;
          });
        return opened(
          Object.freeze({
            classifyClosed: (
              receipt: RealtimeReconnectOutcome<void>,
            ) => receipt,
            close: closeSession,
            waitClosed,
          }),
          recovery,
        );
      },
    });

    const running = coordinator.run({
      initialRecovery: recoveryDirective(
        recovery,
        terminal,
      ),
    });
    await flushUntil(() => clock.sleepCalls.length === 1);
    clock.advance(0);
    const result = await running;
    await flush();

    expect(result).toBe(confirmationFailure);
    expect(gateResult).toBe(confirmationFailure);
    expect(closeSession).toHaveBeenCalledOnce();
    expect(waitClosed).not.toHaveBeenCalled();
  });

  it("retains the exact recovery token across retryable attempt failures and returns its terminal result on exhaustion", async () => {
    const clock = new ManualClock();
    const recovery = recoveryToken();
    const terminal = realtimeFailure(
      "CURSOR_EXPIRED",
      "RECOVER",
      false,
    );
    const attemptFailure = realtimeFailure(
      "CONNECT_TIMEOUT",
      "CONNECT",
      true,
    );
    const gateResults: RealtimeResult<void>[] = [];
    const seenRecoveries: Array<
      RealtimeCommittedRecovery | null
    > = [];
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy({ maxAttempts: 2 }),
      environment: new MutableOnlineEnvironment(true),
      clock,
      random: () => 0,
      confirmTransportBarrier: confirmationSucceeded,
      attempt: (context) => {
        seenRecoveries.push(context.pendingRecovery);
        void context
          .waitForRecoveryBarrierConfirmation()
          .then((result) => gateResults.push(result));
        return failed(attemptFailure);
      },
    });

    const running = coordinator.run({
      initialRecovery: recoveryDirective(
        recovery,
        terminal,
      ),
    });
    await flushUntil(() => clock.sleepCalls.length === 1);
    clock.advance(0);
    await flushUntil(() => clock.sleepCalls.length === 2);
    clock.advance(0);
    const result = await running;
    await flush();

    expect(seenRecoveries).toEqual([recovery, recovery]);
    expect(gateResults).toEqual([
      attemptFailure,
      attemptFailure,
    ]);
    expect(gateResults[0]).toBe(attemptFailure);
    expect(gateResults[1]).toBe(attemptFailure);
    expect(result).toBe(terminal);
  });

  it("fails a no-pending gate immediately and rejects any unsolicited proof", async () => {
    const proof = recoveryToken();
    const closeSession = vi.fn();
    const confirmTransportBarrier = vi.fn(
      confirmationSucceeded,
    );
    let gateResult: RealtimeResult<void> | undefined;
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment: new MutableOnlineEnvironment(true),
      clock: new ManualClock(),
      confirmTransportBarrier,
      attempt: async (context) => {
        gateResult =
          await context.waitForRecoveryBarrierConfirmation();
        return opened(
          Object.freeze({
            classifyClosed: (
              receipt: RealtimeReconnectOutcome<void>,
            ) => receipt,
            close: closeSession,
            waitClosed: closedSuccessfully,
          }),
          proof,
        );
      },
    });

    const result = await coordinator.run();

    expect(gateResult).toEqual(
      realtimeFailure(
        "PROTOCOL_MISMATCH",
        "RECOVER",
        false,
      ),
    );
    expect(result).toBe(gateResult);
    expect(closeSession).toHaveBeenCalledOnce();
    expect(confirmTransportBarrier).not.toHaveBeenCalled();
  });

  it("applies retry ceilings to an initial recovery and preserves its exact terminal result", async () => {
    const recovery = recoveryToken();
    const terminal = realtimeFailure(
      "CURSOR_EXPIRED",
      "RECOVER",
      false,
    );
    const attempt = vi.fn();
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy({
        maxAttempts: 1,
        maxElapsedMs: 500,
      }),
      environment: new MutableOnlineEnvironment(true),
      clock: new ManualClock(),
      confirmTransportBarrier: confirmationSucceeded,
      attempt,
    });

    const result = await coordinator.run({
      initialRecovery: recoveryDirective(
        recovery,
        terminal,
        500,
      ),
    });

    expect(result).toBe(terminal);
    expect(attempt).not.toHaveBeenCalled();
  });

  it("settles a pending readiness gate when the run is aborted", async () => {
    const clock = new ManualClock();
    const recovery = recoveryToken();
    const terminal = realtimeFailure(
      "CURSOR_EXPIRED",
      "RECOVER",
      false,
    );
    const attemptResult =
      deferred<
        RealtimeReconnectOutcome<
          RealtimeReconnectAttemptSuccess<
            RealtimeReconnectOutcome<void>
          >
        >
      >();
    let gateResult: RealtimeResult<void> | undefined;
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy({ maxAttempts: 1 }),
      environment: new MutableOnlineEnvironment(true),
      clock,
      random: () => 0,
      confirmTransportBarrier: confirmationSucceeded,
      attempt: (context) => {
        void context
          .waitForRecoveryBarrierConfirmation()
          .then((result) => {
            gateResult = result;
          });
        return attemptResult.promise;
      },
    });
    const controller = new AbortController();

    const running = coordinator.run({
      signal: controller.signal,
      initialRecovery: recoveryDirective(
        recovery,
        terminal,
      ),
    });
    await flushUntil(() => clock.sleepCalls.length === 1);
    clock.advance(0);
    await flush();
    controller.abort();
    const result = await running;
    await flush();

    expect(result).toEqual(
      realtimeFailure("ABORTED", "CONNECT", false),
    );
    expect(gateResult).toEqual(
      realtimeFailure("ABORTED", "RECOVER", false),
    );

    attemptResult.resolve(
      opened(
        Object.freeze({
          classifyClosed: (
            receipt: RealtimeReconnectOutcome<void>,
          ) => receipt,
          close: vi.fn(),
          waitClosed: closedSuccessfully,
        }),
        recovery,
      ),
    );
    await flush();
  });

  it("rejects accessors without invoking them and consumes proxy values only from one descriptor snapshot", async () => {
    let getterReads = 0;
    const accessorOutcome = Object.defineProperty({}, "result", {
      enumerable: true,
      get() {
        getterReads += 1;
        return realtimeSuccess(undefined);
      },
    });
    const accessorCoordinator =
      createRealtimeReconnectCoordinator({
        policy: policy(),
        environment: new MutableOnlineEnvironment(true),
        clock: new ManualClock(),
        confirmTransportBarrier: confirmationSucceeded,
        attempt: () =>
          accessorOutcome as unknown as RealtimeReconnectOutcome<
            RealtimeReconnectAttemptSuccess<unknown>
          >,
      });

    await expect(accessorCoordinator.run()).resolves.toEqual(
      realtimeFailure(
        "PROTOCOL_MISMATCH",
        "CONNECT",
        false,
      ),
    );
    expect(getterReads).toBe(0);

    const completed = closedSuccessfully();
    const session = Object.freeze({
      classifyClosed: (
        receipt: RealtimeReconnectOutcome<void>,
      ) => receipt,
      close: vi.fn(),
      waitClosed: () => completed,
    });
    const rawOutcome = opened(session);
    const proxyReads: PropertyKey[] = [];
    const proxyOutcome = new Proxy(rawOutcome, {
      get(_target, key) {
        proxyReads.push(key);
        if (key === "then") return undefined;
        throw new Error("raw property reread");
      },
    });
    const proxyCoordinator =
      createRealtimeReconnectCoordinator({
        policy: policy(),
        environment: new MutableOnlineEnvironment(true),
        clock: new ManualClock(),
        confirmTransportBarrier: confirmationSucceeded,
        attempt: () => proxyOutcome,
      });

    await expect(proxyCoordinator.run()).resolves.toBe(
      completed.result,
    );
    expect(proxyReads).toEqual(["then"]);
  });

  it("rechecks scope after attempt and waitClosed boundaries before accepting callbacks", async () => {
    let attemptScopeCurrent = true;
    const lateAttemptClose = vi.fn();
    const lateAttemptWaitClosed = vi.fn(closedSuccessfully);
    const lateAttemptCoordinator =
      createRealtimeReconnectCoordinator({
        policy: policy(),
        environment: new MutableOnlineEnvironment(true),
        clock: new ManualClock(),
        isCurrent: () => attemptScopeCurrent,
        confirmTransportBarrier: confirmationSucceeded,
        attempt: () => {
          attemptScopeCurrent = false;
          return opened(
            Object.freeze({
              classifyClosed: (
                receipt: RealtimeReconnectOutcome<void>,
              ) => receipt,
              close: lateAttemptClose,
              waitClosed: lateAttemptWaitClosed,
            }),
          );
        },
      });

    await expect(lateAttemptCoordinator.run()).resolves.toEqual(
      realtimeFailure("SCOPE_FENCED", "CONNECT", false),
    );
    expect(lateAttemptClose).toHaveBeenCalledOnce();
    expect(lateAttemptWaitClosed).not.toHaveBeenCalled();

    let closeScopeCurrent = true;
    const classifyClosed = vi.fn(
      (receipt: RealtimeReconnectOutcome<void>) => receipt,
    );
    const closeSession = vi.fn();
    const closeCoordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment: new MutableOnlineEnvironment(true),
      clock: new ManualClock(),
      isCurrent: () => closeScopeCurrent,
      confirmTransportBarrier: confirmationSucceeded,
      attempt: () =>
        opened(
          Object.freeze({
            classifyClosed,
            close: closeSession,
            waitClosed: () => {
              closeScopeCurrent = false;
              return closedSuccessfully();
            },
          }),
        ),
    });

    await expect(closeCoordinator.run()).resolves.toEqual(
      realtimeFailure("SCOPE_FENCED", "CONNECT", false),
    );
    expect(classifyClosed).not.toHaveBeenCalled();
    expect(closeSession).toHaveBeenCalledOnce();
  });

  it("rejects extra and missing outcome keys through the common result codec", async () => {
    const validSession = Object.freeze({
      classifyClosed: closedSuccessfully,
      close: vi.fn(),
      waitClosed: closedSuccessfully,
    });
    const invalidOutcomes = [
      {
        extra: true,
        result: realtimeSuccess(validSession),
      },
      {
        result: { ok: true },
      },
      {
        result: {
          ok: false,
          error: {
            extra: true,
            kind: "FORBIDDEN",
            operation: "CONNECT",
            retryable: false,
          },
        },
      },
    ];

    for (const invalid of invalidOutcomes) {
      const coordinator = createRealtimeReconnectCoordinator({
        policy: policy(),
        environment: new MutableOnlineEnvironment(true),
        clock: new ManualClock(),
        confirmTransportBarrier: confirmationSucceeded,
        attempt: () =>
          invalid as unknown as RealtimeReconnectOutcome<
            RealtimeReconnectAttemptSuccess<unknown>
          >,
      });

      await expect(coordinator.run()).resolves.toEqual(
        realtimeFailure(
          "PROTOCOL_MISMATCH",
          "CONNECT",
          false,
        ),
      );
    }
  });

  it("rechecks abort immediately after listener registration", async () => {
    let abortReads = 0;
    let listener: (() => void) | undefined;
    const signal = {
      get aborted() {
        abortReads += 1;
        return abortReads >= 2;
      },
      addEventListener(
        type: string,
        nextListener: () => void,
      ) {
        if (type === "abort") listener = nextListener;
      },
      removeEventListener() {
        listener = undefined;
      },
    } as unknown as AbortSignal;
    const attempt = vi.fn();
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment: new MutableOnlineEnvironment(true),
      clock: new ManualClock(),
      confirmTransportBarrier: confirmationSucceeded,
      attempt,
    });

    const result = await coordinator.run({ signal });

    expect(listener).toBeUndefined();
    expect(attempt).not.toHaveBeenCalled();
    expect(result).toEqual(
      realtimeFailure("ABORTED", "CONNECT", false),
    );
  });

  it("bounds a non-cooperative retry sleep after its phase is aborted", async () => {
    vi.useFakeTimers();
    const environment = new MutableOnlineEnvironment(true);
    const sleep = vi.fn(
      () => new Promise<void>(() => undefined),
    );
    const clock: ClockPort = Object.freeze({
      now: () => 0,
      sleep,
    });
    const retryable = realtimeFailure(
      "CONNECT_TIMEOUT",
      "CONNECT",
      true,
    );
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment,
      clock,
      random: () => 0,
      confirmTransportBarrier: confirmationSucceeded,
      attempt: () => failed(retryable),
    });

    try {
      const running = coordinator.run();
      await flushUntil(() => sleep.mock.calls.length === 1);
      environment.emit(false);
      await flush();

      await vi.advanceTimersByTimeAsync(
        REALTIME_RECONNECT_CEILINGS.drainTimeoutMs,
      );
      const result = await running;

      expect(result).toEqual(
        realtimeFailure(
          "PROVIDER_UNAVAILABLE",
          "CONNECT",
          false,
        ),
      );
      expect(coordinator.getState()).toBe("DRAINING");
      await expect(coordinator.run()).resolves.toEqual(
        realtimeFailure(
          "PROTOCOL_MISMATCH",
          "CONNECT",
          false,
        ),
      );
      expect(environment.listeners.size).toBe(0);
      expect(vi.getTimerCount()).toBe(0);
    } finally {
      coordinator.close();
      vi.useRealTimers();
    }
  });

  it("bounds a non-cooperative attempt and closes its late successful session before leaving DRAINING", async () => {
    vi.useFakeTimers();
    const environment = new MutableOnlineEnvironment(true);
    const attemptResult =
      deferred<
        RealtimeReconnectOutcome<
          RealtimeReconnectAttemptSuccess<
            RealtimeReconnectOutcome<void>
          >
        >
      >();
    const lateClose = vi.fn();
    const lateWaitClosed = vi.fn(closedSuccessfully);
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment,
      clock: new ManualClock(),
      confirmTransportBarrier: confirmationSucceeded,
      attempt: () => attemptResult.promise,
    });

    try {
      const running = coordinator.run();
      await flush();
      environment.emit(false);
      await flush();

      await vi.advanceTimersByTimeAsync(
        REALTIME_RECONNECT_CEILINGS.drainTimeoutMs,
      );
      const result = await running;

      expect(result).toEqual(
        realtimeFailure(
          "PROVIDER_UNAVAILABLE",
          "CONNECT",
          false,
        ),
      );
      expect(coordinator.getState()).toBe("DRAINING");
      await expect(coordinator.run()).resolves.toEqual(
        realtimeFailure(
          "PROTOCOL_MISMATCH",
          "CONNECT",
          false,
        ),
      );
      expect(environment.listeners.size).toBe(0);
      expect(vi.getTimerCount()).toBe(0);

      attemptResult.resolve(
        opened(
          Object.freeze({
            classifyClosed: (
              receipt: RealtimeReconnectOutcome<void>,
            ) => receipt,
            close: lateClose,
            waitClosed: lateWaitClosed,
          }),
        ),
      );
      await flush();

      expect(lateClose).toHaveBeenCalledOnce();
      expect(lateWaitClosed).not.toHaveBeenCalled();
      expect(coordinator.getState()).toBe("IDLE");
    } finally {
      coordinator.close();
      vi.useRealTimers();
    }
  });

  it("cancels the drain timer on abort and keeps the late generation fenced", async () => {
    vi.useFakeTimers();
    const environment = new MutableOnlineEnvironment(true);
    const attemptResult =
      deferred<
        RealtimeReconnectOutcome<
          RealtimeReconnectAttemptSuccess<
            RealtimeReconnectOutcome<void>
          >
        >
      >();
    const lateClose = vi.fn();
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment,
      clock: new ManualClock(),
      confirmTransportBarrier: confirmationSucceeded,
      attempt: () => attemptResult.promise,
    });
    const controller = new AbortController();

    try {
      const running = coordinator.run({
        signal: controller.signal,
      });
      await flush();
      environment.emit(false);
      await flush();
      expect(vi.getTimerCount()).toBe(1);

      controller.abort();
      const result = await running;

      expect(result).toEqual(
        realtimeFailure("ABORTED", "CONNECT", false),
      );
      expect(coordinator.getState()).toBe("DRAINING");
      expect(environment.listeners.size).toBe(0);
      expect(vi.getTimerCount()).toBe(0);

      attemptResult.resolve(
        opened(
          Object.freeze({
            classifyClosed: (
              receipt: RealtimeReconnectOutcome<void>,
            ) => receipt,
            close: lateClose,
            waitClosed: closedSuccessfully,
          }),
        ),
      );
      await flush();

      expect(lateClose).toHaveBeenCalledOnce();
      expect(coordinator.getState()).toBe("IDLE");
    } finally {
      coordinator.close();
      vi.useRealTimers();
    }
  });

  it("keeps a healthy waitClosed unbounded but bounds its post-abort drain", async () => {
    vi.useFakeTimers();
    const environment = new MutableOnlineEnvironment(true);
    const closure =
      deferred<RealtimeReconnectOutcome<void>>();
    const classifyClosed = vi.fn(
      (receipt: RealtimeReconnectOutcome<void>) => receipt,
    );
    const closeSession = vi.fn();
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment,
      clock: new ManualClock(),
      confirmTransportBarrier: confirmationSucceeded,
      attempt: () =>
        opened(
          Object.freeze({
            classifyClosed,
            close: closeSession,
            waitClosed: () => closure.promise,
          }),
        ),
    });

    try {
      let settled = false;
      const running = coordinator.run().then((result) => {
        settled = true;
        return result;
      });
      await flush();

      await vi.advanceTimersByTimeAsync(
        REALTIME_RECONNECT_CEILINGS.drainTimeoutMs * 2,
      );
      expect(settled).toBe(false);
      expect(coordinator.getState()).toBe("RUNNING");
      expect(vi.getTimerCount()).toBe(0);

      environment.emit(false);
      await flush();
      await vi.advanceTimersByTimeAsync(
        REALTIME_RECONNECT_CEILINGS.drainTimeoutMs,
      );
      const result = await running;

      expect(result).toEqual(
        realtimeFailure(
          "PROVIDER_UNAVAILABLE",
          "CONNECT",
          false,
        ),
      );
      expect(closeSession).toHaveBeenCalled();
      expect(classifyClosed).not.toHaveBeenCalled();
      expect(coordinator.getState()).toBe("DRAINING");
      expect(environment.listeners.size).toBe(0);
      expect(vi.getTimerCount()).toBe(0);

      closure.resolve(closedSuccessfully());
      await flush();

      expect(classifyClosed).not.toHaveBeenCalled();
      expect(coordinator.getState()).toBe("IDLE");
    } finally {
      coordinator.close();
      vi.useRealTimers();
    }
  });

  it("uses waitClosed as authority and fences an aborted late generation", async () => {
    const clock = new ManualClock();
    const attemptResult =
      deferred<
        RealtimeReconnectOutcome<
          RealtimeReconnectAttemptSuccess<
            RealtimeReconnectOutcome<void>
          >
        >
      >();
    let captured:
      | RealtimeReconnectAttemptContext
      | undefined;
    const lateClose = vi.fn();
    const lateWaitClosed = vi.fn(closedSuccessfully);
    const coordinator = createRealtimeReconnectCoordinator({
      policy: policy(),
      environment: new MutableOnlineEnvironment(true),
      clock,
      confirmTransportBarrier: confirmationSucceeded,
      attempt: (context) => {
        captured = context;
        return attemptResult.promise;
      },
    });
    const controller = new AbortController();

    const running = coordinator.run({ signal: controller.signal });
    await flush();
    expect(captured?.isCurrent()).toBe(true);

    controller.abort();
    const result = await running;
    expect(result).toEqual(
      realtimeFailure("ABORTED", "CONNECT", false),
    );
    expect(coordinator.getState()).toBe("DRAINING");
    expect(captured?.isCurrent()).toBe(false);

    attemptResult.resolve(
      opened(
        Object.freeze({
          classifyClosed: (
            receipt: RealtimeReconnectOutcome<void>,
          ) => receipt,
          close: lateClose,
          waitClosed: lateWaitClosed,
        }),
      ),
    );
    await flush();

    expect(lateClose).toHaveBeenCalledOnce();
    expect(lateWaitClosed).not.toHaveBeenCalled();
    expect(coordinator.getState()).toBe("IDLE");
  });
});