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

import type {
  ExternalRealtimeEventContext,
  RealtimeEventAuthority,
  RealtimeObservation,
  RealtimeRecoveryCommit,
  RealtimeRecoveryRequest,
} from "../../../src/application/ports/realtime/event-authority.ts";
import type { RealtimeResult } from "../../../src/application/ports/realtime/shared.ts";
import {
  createRealtimeStreamCoordinator,
} from "../../../src/adapters/realtime/stream-coordinator.ts";
import {
  realtimeFailure,
  realtimeSuccess,
} from "../../../src/adapters/realtime/result.ts";
import type { ValidatedRealtimeEventDto } from "../../../src/adapters/realtime/event-codec.ts";
import type { RealtimePolicyRegistry } from "../../../src/contracts/realtime-streams.ts";
import {
  STREAM_ID,
  TEST_LIMITS,
  TEST_MAPPER,
  TEST_MAPPERS,
  createTestRealtimeCodec,
  createTestRealtimeRegistry,
  realtimeEventJson,
  type EventOverrides,
} from "./fixture.ts";

describe("transport-independent realtime stream coordinator", () => {
  it("keeps the stream DRAINING until a non-cooperative effect settles", async () => {
    const wedged = deferred<RealtimeResult<void>>();
    const harness = createHarness({
      apply: async () => wedged.promise,
      taskLimits: { effectTimeoutMs: 5, drainTimeoutMs: 5 },
    });
    await harness.initialize();

    const applied = await harness.coordinator.accept(
      harness.event({
        eventId: "event-00000001",
        sequence: "1",
        resumeCursor: "cursor-00000001",
      }),
    );
    // Bounded for the caller, and explicitly non-retryable.
    expect(applied).toMatchObject({
      ok: false,
      error: { kind: "IDLE_TIMEOUT", operation: "APPLY", retryable: false },
    });
    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING");

    // DRAINING refuses new admission rather than queueing behind the wedge.
    await expect(
      harness.coordinator.accept(
        harness.event({
          eventId: "event-00000002",
          sequence: "2",
          resumeCursor: "cursor-00000002",
        }),
      ),
    ).resolves.toMatchObject({ ok: false, error: { kind: "CLOSED" } });

    // close() cannot claim quiescence while the task is still running.
    await expect(harness.coordinator.close()).resolves.toMatchObject({
      ok: false,
      error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
    });

    // Only actual settlement ends DRAINING.
    wedged.resolve(realtimeSuccess(undefined));
    await expect(harness.coordinator.close()).resolves.toMatchObject({
      ok: true,
    });
    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("CLOSED");
  });

  it("bounds non-cooperative recovery and rejects its late checkpoint", async () => {
    const wedged = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
    let recoveries = 0;
    const harness = createHarness({
      recover: async () => {
        recoveries += 1;
        return recoveries === 1
          ? realtimeSuccess(snapshotCommit("0"))
          : wedged.promise;
      },
      taskLimits: { recoveryTimeoutMs: 5, drainTimeoutMs: 5 },
    });
    await harness.initialize();

    const recovered = await harness.coordinator.recover(
      STREAM_ID,
      "SEQUENCE_GAP",
    );
    expect(recovered).toMatchObject({
      ok: false,
      error: { kind: "IDLE_TIMEOUT", operation: "RECOVER", retryable: false },
    });
    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING");

    const beforeLateCommit = harness.coordinator.getResumeState(STREAM_ID);

    // A late checkpoint from the abandoned attempt cannot commit.
    wedged.resolve(realtimeSuccess(snapshotCommit("9")));
    for (let flush = 0; flush < 10; flush += 1) await Promise.resolve();
    expect(harness.coordinator.getResumeState(STREAM_ID)).toEqual(
      beforeLateCommit,
    );
    // Settlement returns the stream to OPEN, marked STALE for an authoritative
    // recovery rather than silently trusting the abandoned attempt.
    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("OPEN");
    expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe("STALE");
  });

  it("applies one stream sequentially and commits each cursor after its effect", async () => {
    const first = deferred<RealtimeResult<void>>();
    const applied: string[] = [];
    const harness = createHarness({
      async apply(_profile, value) {
        const selected = (value as Readonly<{ value: string }>).value;
        applied.push(selected);
        return selected === "first"
          ? first.promise
          : realtimeSuccess(undefined);
      },
    });
    await harness.initialize();

    const firstResult = harness.coordinator.accept(
      harness.event({
        eventId: "event-00000001",
        sequence: "1",
        resumeCursor: "cursor-00000001",
        payload: { value: "first" },
      }),
    );
    const secondResult = harness.coordinator.accept(
      harness.event({
        eventId: "event-00000002",
        sequence: "2",
        resumeCursor: "cursor-00000002",
        payload: { value: "second" },
      }),
    );

    await vi.waitFor(() => expect(applied).toEqual(["first"]));
    expect(harness.coordinator.getResumeState(STREAM_ID)).toMatchObject({
      lastAppliedSequence: "0",
      resumeCursor: "cursor-snapshot-0",
    });

    first.resolve(realtimeSuccess(undefined));
    await expect(firstResult).resolves.toMatchObject({
      ok: true,
      value: { outcome: "APPLIED" },
    });
    await expect(secondResult).resolves.toMatchObject({
      ok: true,
      value: { outcome: "APPLIED" },
    });
    expect(applied).toEqual(["first", "second"]);
    expect(harness.coordinator.getResumeState(STREAM_ID)).toEqual({
      recoveryMode: "CURSOR",
      streamEpoch: "stream-epoch-0001",
      lastAppliedSequence: "2",
      resumeCursor: "cursor-00000002",
    });
  });

  it("drops exact duplicates and stale events without replaying effects", async () => {
    const harness = createHarness();
    await harness.initialize();
    const event = harness.event();

    await harness.coordinator.accept(event);
    await expect(harness.coordinator.accept(event)).resolves.toEqual({
      ok: true,
      value: {
        outcome: "DROPPED",
        reason: "DUPLICATE_EVENT",
      },
    });
    await expect(
      harness.coordinator.accept(
        harness.event({
          eventId: "event-stale-0001",
          sequence: "0",
          resumeCursor: "cursor-stale-0001",
        }),
      ),
    ).resolves.toEqual({
      ok: true,
      value: {
        outcome: "DROPPED",
        reason: "STALE_EVENT",
      },
    });
    expect(harness.effects).toHaveBeenCalledTimes(1);
    expect(harness.coordinator.inspect(STREAM_ID).dedupeEntries).toBe(1);
  });

  it("recovers without applying a conflicting ID, sequence gap or epoch", async () => {
    for (const [overrides, expectedReason] of [
      [
        {
          eventId: "event-00000001",
          sequence: "1",
          payload: { value: "conflict" },
        },
        "EVENT_CONFLICT",
      ],
      [
        {
          eventId: "event-gap-000001",
          sequence: "3",
          resumeCursor: "cursor-gap-000001",
        },
        "SEQUENCE_GAP",
      ],
      [
        {
          eventId: "event-epoch-0001",
          sequence: "2",
          streamEpoch: "stream-epoch-0002",
          resumeCursor: "cursor-epoch-0001",
        },
        "STREAM_EPOCH_CHANGED",
      ],
    ] as const) {
      const harness = createHarness();
      await harness.initialize();
      await harness.coordinator.accept(harness.event());

      await expect(
        harness.coordinator.accept(harness.event(overrides)),
      ).resolves.toMatchObject({
        ok: true,
        value: {
          outcome: "RECOVERED",
          reason: expectedReason,
        },
      });
      expect(harness.recoveryReasons).toEqual([
        "INITIALIZE",
        expectedReason,
      ]);
      expect(harness.effects).toHaveBeenCalledTimes(1);
    }
  });

  it("does not apply or advance when mapping fails", async () => {
    const harness = createHarness({
      mappers: {
        ReferenceRealtimeMapper: {
          ...TEST_MAPPER,
          map: () => ({
            ok: false as const,
            code: "MAPPING_INVARIANT_REJECTED" as const,
          }),
        },
      },
    });
    await harness.initialize();

    await expect(
      harness.coordinator.accept(harness.event()),
    ).resolves.toMatchObject({
      ok: true,
      value: {
        outcome: "RECOVERED",
        reason: "MAPPING_CONTRACT_VIOLATION",
      },
    });
    expect(harness.effects).not.toHaveBeenCalled();
    expect(harness.recoveryReasons).toEqual([
      "INITIALIZE",
      "MAPPING_CONTRACT_VIOLATION",
    ]);
  });

  it("leaves the prior checkpoint when an effect and its recovery fail", async () => {
    let recoveryCount = 0;
    const harness = createHarness({
      apply: async () => realtimeFailure("APPLY_FAILED", "APPLY"),
      recover: async () => {
        recoveryCount += 1;
        return recoveryCount === 1
          ? realtimeSuccess(snapshotCommit("0"))
          : realtimeFailure("PROVIDER_UNAVAILABLE", "RECOVER");
      },
    });
    await harness.initialize();

    await expect(
      harness.coordinator.accept(harness.event()),
    ).resolves.toMatchObject({
      ok: false,
      error: {
        kind: "PROVIDER_UNAVAILABLE",
        operation: "RECOVER",
      },
    });
    expect(harness.coordinator.getResumeState(STREAM_ID)).toMatchObject({
      lastAppliedSequence: "0",
      resumeCursor: "cursor-snapshot-0",
    });
    expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe(
      "UNKNOWN",
    );
    await harness.coordinator.accept(
      harness.event({
        eventId: "event-00000002",
        sequence: "2",
        resumeCursor: "cursor-00000002",
      }),
    );
    expect(harness.effects).toHaveBeenCalledTimes(1);
  });

  it("rejects malformed authority results without advancing a checkpoint", async () => {
    let recoveryCount = 0;
    const harness = createHarness({
      apply: async () =>
        ({
          ok: true,
          value: "not-void",
        }) as unknown as RealtimeResult<void>,
      recover: async () => {
        recoveryCount += 1;
        return recoveryCount === 1
          ? realtimeSuccess(snapshotCommit("0"))
          : ({
              ok: false,
              error: {
                kind: "FORBIDDEN",
                operation: "APPLY",
                retryable: false,
              },
            } as RealtimeResult<RealtimeRecoveryCommit>);
      },
    });
    await harness.initialize();

    await expect(
      harness.coordinator.accept(harness.event()),
    ).resolves.toEqual(
      realtimeFailure("PROTOCOL_MISMATCH", "RECOVER"),
    );
    expect(harness.coordinator.getResumeState(STREAM_ID)).toMatchObject({
      lastAppliedSequence: "0",
    });
    expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe(
      "UNKNOWN",
    );
  });

  it("rejects recovery accessors without invoking them", async () => {
    let checkpointReads = 0;
    const accessorCommit = Object.defineProperties(
      {},
      {
        checkpoint: {
          enumerable: true,
          get() {
            checkpointReads += 1;
            return snapshotCommit("0");
          },
        },
        kind: {
          enumerable: true,
          value: "SNAPSHOT_RESET",
        },
      },
    ) as RealtimeRecoveryCommit;
    const harness = createHarness({
      recover: async () => realtimeSuccess(accessorCommit),
    });

    await expect(
      harness.coordinator.recover(STREAM_ID, "INITIALIZE"),
    ).resolves.toEqual(
      realtimeFailure("PROTOCOL_MISMATCH", "RECOVER"),
    );
    expect(checkpointReads).toBe(0);
    expect(harness.coordinator.getResumeState(STREAM_ID)).toBeNull();
  });

  it("commits only the one-shot descriptor snapshot of a recovery proxy", async () => {
    const descriptorCommit = snapshotCommit("0");
    const propertyReads: PropertyKey[] = [];
    const proxyCommit = new Proxy(descriptorCommit, {
      get(_target, key) {
        propertyReads.push(key);
        if (key === "kind") return "SESSION_REBUILD";
        if (key === "checkpoint") return snapshotCommit("99");
        return undefined;
      },
    });
    const harness = createHarness({
      recover: async () => realtimeSuccess(proxyCommit),
    });

    const recovered = await harness.coordinator.recover(
      STREAM_ID,
      "INITIALIZE",
    );

    expect(recovered).toMatchObject({
      ok: true,
      value: {
        lastAppliedSequence: "0",
        resumeCursor: "cursor-snapshot-0",
      },
    });
    expect(propertyReads).toEqual([]);
  });

  it.each([
    Object.freeze({
      ...snapshotCommit("0"),
      unexpected: true,
    }),
    Object.assign(
      Object.create({ inherited: true }) as Record<string, unknown>,
      snapshotCommit("0"),
    ),
  ])(
    "rejects extra and inherited recovery commit shapes",
    async (commit) => {
      const harness = createHarness({
        recover: async () =>
          realtimeSuccess(commit as RealtimeRecoveryCommit),
      });

      await expect(
        harness.coordinator.recover(STREAM_ID, "INITIALIZE"),
      ).resolves.toEqual(
        realtimeFailure("PROTOCOL_MISMATCH", "RECOVER"),
      );
      expect(harness.coordinator.getResumeState(STREAM_ID)).toBeNull();
    },
  );

  it("expires effect and recovery commit leases when callbacks settle", async () => {
    let effectContext: ExternalRealtimeEventContext | undefined;
    let recoveryRequest: RealtimeRecoveryRequest | undefined;
    const harness = createHarness({
      async apply(_profile, _value, context) {
        effectContext = context;
        expect(context.isCurrent()).toBe(true);
        return realtimeSuccess(undefined);
      },
      async recover(request) {
        recoveryRequest = request;
        expect(request.isCurrent()).toBe(true);
        return realtimeSuccess(snapshotCommit("0"));
      },
    });

    await harness.initialize();
    expect(recoveryRequest?.isCurrent()).toBe(false);
    await harness.coordinator.accept(harness.event());
    expect(effectContext?.isCurrent()).toBe(false);
  });

  it("fences an in-flight successful effect before cursor commit", async () => {
    const pending = deferred<RealtimeResult<void>>();
    let effectContext: ExternalRealtimeEventContext | undefined;
    const harness = createHarness({
      apply: async (_profile, _value, context) => {
        effectContext = context;
        return pending.promise;
      },
    });
    await harness.initialize();

    const accepted = harness.coordinator.accept(harness.event());
    await vi.waitFor(() =>
      expect(harness.effects).toHaveBeenCalledOnce(),
    );
    harness.setCurrent(false);
    expect(effectContext?.isCurrent()).toBe(false);
    pending.resolve(realtimeSuccess(undefined));

    await expect(accepted).resolves.toEqual({
      ok: true,
      value: {
        outcome: "DROPPED",
        reason: "SCOPE_FENCED",
      },
    });
    expect(harness.coordinator.getResumeState(STREAM_ID)).toBeNull();
    expect(harness.coordinator.inspect(STREAM_ID)).toMatchObject({
      freshness: "UNKNOWN",
      dedupeEntries: 0,
      queuedEvents: 0,
      awaitingTransportBarrier: false,
    });
  });

  it("aborts and quiesces the active effect before queue-overflow recovery", async () => {
    const pending = deferred<RealtimeResult<void>>();
    let capturedSignal: AbortSignal | undefined;
    const registry = createTestRealtimeRegistry({
      limits: {
        ...TEST_LIMITS,
        maxQueueEvents: 1,
      },
    });
    const harness = createHarness({
      registry,
      async apply(_profile, _value, _context, signal) {
        capturedSignal = signal;
        return pending.promise;
      },
    });
    await harness.initialize();

    const first = harness.coordinator.accept(harness.event());
    await vi.waitFor(() =>
      expect(harness.effects).toHaveBeenCalledOnce(),
    );
    const overflow = harness.coordinator.accept(
      harness.event({
        eventId: "event-00000002",
        sequence: "2",
        resumeCursor: "cursor-00000002",
      }),
    );

    expect(capturedSignal?.aborted).toBe(true);
    expect(harness.recoveryReasons).toEqual(["INITIALIZE"]);
    pending.resolve(realtimeSuccess(undefined));
    await expect(first).resolves.toMatchObject({
      ok: true,
      value: { outcome: "DROPPED" },
    });
    await expect(overflow).resolves.toMatchObject({
      ok: true,
      value: {
        outcome: "RECOVERED",
        reason: "QUEUE_OVERFLOW",
      },
    });
    expect(harness.recoveryReasons).toEqual([
      "INITIALIZE",
      "QUEUE_OVERFLOW",
    ]);
  });

  it("treats a current scope mismatch as a protocol violation, not a stale callback", async () => {
    const harness = createHarness();
    await harness.initialize();

    await expect(
      harness.coordinator.accept(
        harness.event({ scopeBinding: "other-scope-binding" }),
      ),
    ).resolves.toMatchObject({
      ok: true,
      value: {
        outcome: "RECOVERED",
        reason: "SCOPE_PROTOCOL_VIOLATION",
      },
    });
    expect(harness.effects).not.toHaveBeenCalled();
    expect(harness.recoveryReasons).toEqual([
      "INITIALIZE",
      "SCOPE_PROTOCOL_VIOLATION",
    ]);
  });

  it("never claims CURRENT for a SNAPSHOT_ONLY stream without a barrier", async () => {
    const registry = createTestRealtimeRegistry({
      recovery: {
        mode: "SNAPSHOT_ONLY",
        snapshotOperationId: "GET_REFERENCE_REALTIME_SNAPSHOT",
        checkpointCodecId: "ReferenceRealtimeCheckpoint",
        barrier: "NONE",
      },
      delivery: "INVALIDATION_HINT",
      stateBearing: false,
    });
    const harness = createHarness({
      registry,
      recover: async () =>
        realtimeSuccess({
          kind: "SNAPSHOT_RESET",
          checkpoint: {
            recoveryMode: "SNAPSHOT_ONLY",
            streamEpoch: "stream-epoch-0001",
            lastAppliedSequence: "0",
            resumeCursor: null,
            snapshotRevision: "snapshot-revision-0",
          },
        }),
    });
    await harness.initialize();
    expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe("STALE");

    await expect(
      harness.coordinator.accept(
        harness.event({
          recoveryMode: "SNAPSHOT_ONLY",
          resumeCursor: null,
        }),
      ),
    ).resolves.toMatchObject({
      ok: true,
      value: { outcome: "APPLIED" },
    });
    expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe("STALE");
  });

  it("requires an exact transport barrier before promoting a recovered stream to CURRENT", async () => {
    const harness = createHarness();
    const recovered = await harness.coordinator.recover(
      STREAM_ID,
      "INITIALIZE",
    );
    expect(recovered.ok).toBe(true);
    if (!recovered.ok) return;
    await Promise.resolve();

    expect(harness.coordinator.inspect(STREAM_ID)).toMatchObject({
      freshness: "STALE",
      awaitingTransportBarrier: true,
    });
    expect(
      harness.coordinator.confirmTransportBarrier(
        STREAM_ID,
        Object.freeze({
          ...recovered.value,
        }) as typeof recovered.value,
      ),
    ).toMatchObject({
      ok: false,
      error: { kind: "PROTOCOL_MISMATCH" },
    });
    const publicResumeState =
      harness.coordinator.getResumeState(STREAM_ID);
    expect(publicResumeState).not.toBe(recovered.value);
    expect(
      harness.coordinator.confirmTransportBarrier(
        STREAM_ID,
        publicResumeState as typeof recovered.value,
      ),
    ).toMatchObject({
      ok: false,
      error: { kind: "PROTOCOL_MISMATCH" },
    });
    expect(
      harness.coordinator.confirmTransportBarrier(
        STREAM_ID,
        recovered.value,
      ),
    ).toEqual(realtimeSuccess(undefined));
    expect(harness.coordinator.inspect(STREAM_ID)).toMatchObject({
      freshness: "CURRENT",
      awaitingTransportBarrier: false,
    });
  });

  it("marks a successful invalidation hint stale until authoritative refresh", async () => {
    const registry = createTestRealtimeRegistry({
      recovery: {
        mode: "SNAPSHOT_ONLY",
        snapshotOperationId: "GET_REFERENCE_REALTIME_SNAPSHOT",
        checkpointCodecId: "ReferenceRealtimeCheckpoint",
        barrier: "CONNECT_BUFFER",
      },
      delivery: "INVALIDATION_HINT",
      stateBearing: false,
    });
    const harness = createHarness({
      registry,
      recover: async () =>
        realtimeSuccess({
          kind: "SNAPSHOT_RESET",
          checkpoint: {
            recoveryMode: "SNAPSHOT_ONLY",
            streamEpoch: "stream-epoch-0001",
            lastAppliedSequence: "0",
            resumeCursor: null,
            snapshotRevision: "snapshot-revision-0",
          },
        }),
    });
    await harness.initialize();
    expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe(
      "CURRENT",
    );

    await expect(
      harness.coordinator.accept(
        harness.event({
          recoveryMode: "SNAPSHOT_ONLY",
          resumeCursor: null,
        }),
      ),
    ).resolves.toMatchObject({
      ok: true,
      value: { outcome: "APPLIED" },
    });
    expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe("STALE");
  });

  it("recovers instead of evicting and continuing after dedupe capacity", async () => {
    const harness = createHarness({
      registry: createTestRealtimeRegistry({
        limits: {
          ...TEST_LIMITS,
          maxDedupeEntries: 1,
        },
      }),
    });
    await harness.initialize();
    await harness.coordinator.accept(harness.event());

    await expect(
      harness.coordinator.accept(
        harness.event({
          eventId: "event-00000002",
          sequence: "2",
          resumeCursor: "cursor-00000002",
        }),
      ),
    ).resolves.toMatchObject({
      ok: true,
      value: {
        outcome: "RECOVERED",
        reason: "DEDUPE_OVERFLOW",
      },
    });
    expect(harness.effects).toHaveBeenCalledTimes(1);
    expect(harness.coordinator.inspect(STREAM_ID).dedupeEntries).toBe(0);
  });

  it("rejects a regressing same-epoch recovery checkpoint atomically", async () => {
    let checkpoint = "5";
    const harness = createHarness({
      recover: async () => realtimeSuccess(snapshotCommit(checkpoint)),
    });
    await harness.initialize();
    checkpoint = "4";

    await expect(
      harness.coordinator.recover(STREAM_ID, "CURSOR_EXPIRED"),
    ).resolves.toMatchObject({
      ok: false,
      error: { kind: "PROTOCOL_MISMATCH" },
    });
    expect(harness.coordinator.getResumeState(STREAM_ID)).toMatchObject({
      lastAppliedSequence: "5",
    });
  });

  it("coalesces recovery, closes idempotently and rejects forged DTOs", async () => {
    const pending = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
    const recover = vi.fn(async () => pending.promise);
    const harness = createHarness({ recover });

    const first = harness.coordinator.recover(STREAM_ID, "INITIALIZE");
    const second = harness.coordinator.recover(STREAM_ID, "SEQUENCE_GAP");
    await vi.waitFor(() => expect(recover).toHaveBeenCalledOnce());
    pending.resolve(realtimeSuccess(snapshotCommit("0")));
    await expect(first).resolves.toMatchObject({ ok: true });
    await expect(second).resolves.toMatchObject({ ok: true });

    harness.coordinator.close();
    harness.coordinator.close();
    expect(harness.coordinator.getResumeState(STREAM_ID)).toBeNull();
    await expect(
      harness.coordinator.accept(
        Object.freeze({}) as ValidatedRealtimeEventDto,
      ),
    ).resolves.toMatchObject({
      ok: false,
      error: { kind: "MALFORMED_EVENT" },
    });
    await expect(
      harness.coordinator.accept(harness.event()),
    ).resolves.toEqual({
      ok: true,
      value: { outcome: "DROPPED", reason: "CLOSED" },
    });
  });

  it("returns the exact shared checkpoint to an event waiting on active recovery", async () => {
    const pending = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
    const recover = vi.fn(async () => pending.promise);
    const harness = createHarness({ recover });

    const recovering = harness.coordinator.recover(
      STREAM_ID,
      "INITIALIZE",
    );
    await vi.waitFor(() => expect(recover).toHaveBeenCalledOnce());
    const accepting = harness.coordinator.accept(harness.event());
    pending.resolve(realtimeSuccess(snapshotCommit("0")));

    const recovered = await recovering;
    const accepted = await accepting;
    expect(recovered.ok).toBe(true);
    expect(accepted.ok).toBe(true);
    if (!recovered.ok || !accepted.ok) return;
    expect(accepted.value).toMatchObject({
      outcome: "RECOVERED",
      reason: "INITIALIZE",
    });
    if (accepted.value.outcome !== "RECOVERED") return;
    expect(accepted.value.resumeState).toBe(recovered.value);
  });

  it("aborts only an event waiter without cancelling shared recovery", async () => {
    const pending = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
    let recoverySignal: AbortSignal | undefined;
    const harness = createHarness({
      recover: async (request) => {
        recoverySignal = request.signal;
        return pending.promise;
      },
    });
    const recovering = harness.coordinator.recover(
      STREAM_ID,
      "INITIALIZE",
    );
    await vi.waitFor(() => expect(recoverySignal).toBeDefined());
    const controller = new AbortController();
    const accepting = harness.coordinator.accept(
      harness.event(),
      controller.signal,
    );

    controller.abort();
    await expect(accepting).resolves.toEqual(
      realtimeFailure("ABORTED", "RECEIVE"),
    );
    expect(recoverySignal?.aborted).toBe(false);

    pending.resolve(realtimeSuccess(snapshotCommit("0")));
    await expect(recovering).resolves.toMatchObject({ ok: true });
  });

  it("aborts the active recovery authority when closed", async () => {
    let recoverySignal: AbortSignal | undefined;
    const harness = createHarness({
      recover: async (request) => {
        recoverySignal = request.signal;
        return new Promise((resolve) => {
          request.signal.addEventListener(
            "abort",
            () => resolve(realtimeFailure("ABORTED", "RECOVER")),
            { once: true },
          );
        });
      },
    });

    const recovering = harness.coordinator.recover(
      STREAM_ID,
      "INITIALIZE",
    );
    await vi.waitFor(() => expect(recoverySignal).toBeDefined());
    harness.coordinator.close();

    expect(recoverySignal?.aborted).toBe(true);
    await expect(recovering).resolves.toMatchObject({
      ok: false,
      error: { kind: "CLOSED", operation: "RECOVER" },
    });
  });

  it("does not commit a non-cooperative recovery after caller cancellation", async () => {
    const pending = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
    let recoveryRequest: RealtimeRecoveryRequest | undefined;
    const harness = createHarness({
      recover: async (request) => {
        recoveryRequest = request;
        return pending.promise;
      },
    });
    const controller = new AbortController();
    const recovering = harness.coordinator.recover(
      STREAM_ID,
      "INITIALIZE",
      controller.signal,
    );
    await vi.waitFor(() => expect(recoveryRequest).toBeDefined());

    controller.abort();
    expect(recoveryRequest?.isCurrent()).toBe(false);
    pending.resolve(realtimeSuccess(snapshotCommit("0")));

    await expect(recovering).resolves.toEqual(
      realtimeFailure("ABORTED", "RECOVER"),
    );
    expect(recoveryRequest?.signal.aborted).toBe(true);
    expect(recoveryRequest?.isCurrent()).toBe(false);
    expect(harness.coordinator.getResumeState(STREAM_ID)).toBeNull();
    expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe(
      "UNKNOWN",
    );
  });

  it("emits only closed redacted observations and ignores sink failure", async () => {
    const observations: RealtimeObservation[] = [];
    const harness = createHarness({
      observe(observation) {
        observations.push(observation);
        throw new Error("diagnostic sink unavailable");
      },
    });
    await harness.initialize();
    await harness.coordinator.accept(
      harness.event({
        eventId: "sensitive-event-id",
        resumeCursor: "sensitive-cursor",
        payload: { value: "sensitive-payload" },
      }),
    );

    const serialized = JSON.stringify(observations);
    expect(serialized).not.toContain("sensitive-event-id");
    expect(serialized).not.toContain("sensitive-cursor");
    expect(serialized).not.toContain("sensitive-payload");
    expect(serialized).not.toContain("scope-binding-0001");
    expect(observations).toEqual(
      expect.arrayContaining([
        expect.objectContaining({
          operation: "RECOVER",
          outcome: "RECOVERED",
          reason: "INITIALIZE",
        }),
        expect.objectContaining({
          operation: "APPLY",
          outcome: "APPLIED",
        }),
      ]),
    );
  });
});

/**
 * RT-RR-01 / RT-RR-02. A physical effect exists from the moment the coordinator
 * calls the authority, not from the moment its public wait expires. Registering
 * only on timeout let a `close()` that arrived first see an empty retained set
 * and report quiescence while the raw task was still running against the
 * authority.
 */
describe("realtime physical task ownership", () => {
  it("does not report quiescence while an apply is still running", async () => {
    const applyGate = deferred<RealtimeResult<void>>();
    const harness = createHarness({
      apply: async () => await applyGate.promise,
      taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 },
    });
    await harness.initialize();

    const applying = harness.coordinator.accept(harness.event());
    await Promise.resolve();
    await Promise.resolve();

    // close() arrives long before the effect deadline.
    const closed = await harness.coordinator.close();
    expect(closed.ok).toBe(false);
    expect(closed.ok ? null : closed.error.kind).toBe("IDLE_TIMEOUT");
    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING");

    applyGate.resolve(realtimeSuccess(undefined));
    await applying;
  });

  it("reports quiescence once the raw task settles", async () => {
    const applyGate = deferred<RealtimeResult<void>>();
    const harness = createHarness({
      apply: async () => await applyGate.promise,
      taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 200 },
    });
    await harness.initialize();

    const applying = harness.coordinator.accept(harness.event());
    await Promise.resolve();
    const closing = harness.coordinator.close();
    applyGate.resolve(realtimeSuccess(undefined));
    await applying;

    expect(await closing).toMatchObject({ ok: true });
  });

  it("does not start a queued event once the stream is DRAINING", async () => {
    const firstApply = deferred<RealtimeResult<void>>();
    let applyCalls = 0;
    const harness = createHarness({
      apply: async () => {
        applyCalls += 1;
        if (applyCalls === 1) return await firstApply.promise;
        return realtimeSuccess(undefined);
      },
      taskLimits: { effectTimeoutMs: 15, drainTimeoutMs: 50 },
    });
    await harness.initialize();

    const first = harness.coordinator.accept(
      harness.event({ eventId: "event-0001", sequence: "1" }),
    );
    // Queued behind the first while it is still inside its deadline.
    const second = harness.coordinator.accept(
      harness.event({ eventId: "event-0002", sequence: "2" }),
    );

    await first;
    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING");

    // The queued event is dropped at execution time, not applied and not
    // recovered: admission happened before DRAINING, execution happens inside
    // it, and the second decision is the one that counts.
    const secondResult = await second;
    expect(secondResult).toMatchObject({
      ok: true,
      value: { outcome: "DROPPED", reason: "CLOSED" },
    });
    expect(applyCalls).toBe(1);

    firstApply.resolve(realtimeSuccess(undefined));
  });

  it("discards the resume token when an effect is abandoned", async () => {
    const firstApply = deferred<RealtimeResult<void>>();
    let applyCalls = 0;
    const harness = createHarness({
      apply: async () => {
        applyCalls += 1;
        if (applyCalls === 1) return await firstApply.promise;
        return realtimeSuccess(undefined);
      },
      taskLimits: { effectTimeoutMs: 15, drainTimeoutMs: 50 },
    });
    await harness.initialize();
    expect(harness.coordinator.inspect(STREAM_ID).hasResumeState).toBe(true);

    await harness.coordinator.accept(
      harness.event({ eventId: "event-0001", sequence: "1" }),
    );

    // The abandoned effect may have applied part of its change, so the token it
    // was based on is no longer authoritative evidence.
    const inspection = harness.coordinator.inspect(STREAM_ID);
    expect(inspection.hasResumeState).toBe(false);
    expect(inspection.freshness).toBe("UNKNOWN");

    firstApply.resolve(realtimeSuccess(undefined));
  });
});

type HarnessOptions = Readonly<{
  registry?: RealtimePolicyRegistry;
  mappers?: typeof TEST_MAPPERS | Readonly<Record<string, typeof TEST_MAPPER>>;
  apply?: RealtimeEventAuthority["effects"]["apply"];
  recover?: (
    request: RealtimeRecoveryRequest,
  ) => Promise<RealtimeResult<RealtimeRecoveryCommit>>;
  observe?: (observation: RealtimeObservation) => void;
  taskLimits?: Readonly<{
    effectTimeoutMs?: number;
    recoveryTimeoutMs?: number;
    drainTimeoutMs?: number;
  }>;
  scheduleTimeout?: (callback: () => void, delayMs: number) => unknown;
  clearScheduledTimeout?: (handle: unknown) => void;
}>;

function createHarness(options: HarnessOptions = {}) {
  const registry = options.registry ?? createTestRealtimeRegistry();
  const codec = createTestRealtimeCodec(registry);
  let current = true;
  const recoveryReasons: string[] = [];
  let defaultRecoveryCount = 0;
  const effects = vi.fn(
    options.apply ??
      (async () => realtimeSuccess(undefined)),
  );
  const recover = vi.fn(
    options.recover ??
      (async (request: RealtimeRecoveryRequest) => {
        recoveryReasons.push(request.reason);
        defaultRecoveryCount += 1;
        return realtimeSuccess(
          snapshotCommit(
            "0",
            `stream-epoch-${String(defaultRecoveryCount).padStart(4, "0")}`,
          ),
        );
      }),
  );
  const authority: RealtimeEventAuthority = {
    effects: { apply: effects },
    recovery: {
      async recover(request) {
        if (options.recover) recoveryReasons.push(request.reason);
        return recover(request);
      },
    },
  };
  const coordinator = createRealtimeStreamCoordinator({
    registry,
    mappers: options.mappers ?? TEST_MAPPERS,
    authority,
    scope: {
      generation: 7,
      scopeBinding: "scope-binding-0001",
      isCurrent: () => current,
    },
    now: () => 10_000,
    observe: options.observe,
    ...(options.taskLimits ? { taskLimits: options.taskLimits } : {}),
    ...(options.scheduleTimeout
      ? { scheduleTimeout: options.scheduleTimeout }
      : {}),
    ...(options.clearScheduledTimeout
      ? { clearScheduledTimeout: options.clearScheduledTimeout }
      : {}),
  });

  return {
    coordinator,
    effects,
    recoveryReasons,
    event(overrides: EventOverrides = {}) {
      const decoded = codec.decode(realtimeEventJson(overrides));
      if (!decoded.ok) {
        throw new Error(`test event failed: ${decoded.error.kind}`);
      }
      return decoded.value;
    },
    async initialize() {
      const result = await coordinator.recover(
        STREAM_ID,
        "INITIALIZE",
      );
      await Promise.resolve();
      if (
        result.ok &&
        coordinator.inspect(STREAM_ID).awaitingTransportBarrier
      ) {
        const confirmed = coordinator.confirmTransportBarrier(
          STREAM_ID,
          result.value,
        );
        if (!confirmed.ok) {
          throw new Error("test transport barrier failed");
        }
      }
      return result;
    },
    setCurrent(value: boolean) {
      current = value;
    },
  };
}

function snapshotCommit(
  sequence: string,
  streamEpoch = "stream-epoch-0001",
): RealtimeRecoveryCommit {
  return Object.freeze({
    kind: "SNAPSHOT_RESET",
    checkpoint: Object.freeze({
      recoveryMode: "CURSOR",
      streamEpoch,
      lastAppliedSequence: sequence,
      resumeCursor: `cursor-snapshot-${sequence}`,
      snapshotRevision: `snapshot-revision-${sequence}`,
    }),
  });
}

function deferred<Value>() {
  let resolve!: (value: Value) => void;
  const promise = new Promise<Value>((next) => {
    resolve = next;
  });
  return { promise, resolve };
}

/**
 * RT-01. The registry entry has to exist before the collaborator is called.
 * Registering after the invocation returned left a window in which an authority
 * that re-entered `close()` from inside its own callback saw an empty set, so
 * `close()` reported quiescence while its effect was still running.
 */
describe("RT-01 reentrant close cannot pass the registration window", () => {
  it("refuses quiescence when apply re-enters close during its invocation", async () => {
    const applyGate = deferred<RealtimeResult<void>>();
    let closeResult: RealtimeResult<void> | undefined;
    let harness: ReturnType<typeof createHarness> | undefined;
    harness = createHarness({
      apply: () => {
        // Re-entered from inside the invocation itself.
        void harness!.coordinator.close().then((result) => {
          closeResult = result;
        });
        return applyGate.promise;
      },
      taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 },
    });
    await harness.initialize();

    const applying = harness.coordinator.accept(harness.event());
    await new Promise((resolve) => setTimeout(resolve, 50));

    expect(closeResult?.ok).toBe(false);
    expect(closeResult?.ok ? null : closeResult?.error.kind).toBe(
      "IDLE_TIMEOUT",
    );

    applyGate.resolve(realtimeSuccess(undefined));
    await applying;
    // Once the raw task settled, a second close does converge.
    await expect(harness.coordinator.close()).resolves.toMatchObject({
      ok: true,
    });
  });

  it("refuses quiescence when recovery re-enters close during its invocation", async () => {
    const recoveryGate = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
    let closeResult: RealtimeResult<void> | undefined;
    let harness: ReturnType<typeof createHarness> | undefined;
    let recoveries = 0;
    harness = createHarness({
      recover: () => {
        recoveries += 1;
        if (recoveries === 1) {
          return Promise.resolve(realtimeSuccess(snapshotCommit("0")));
        }
        void harness!.coordinator.close().then((result) => {
          closeResult = result;
        });
        return recoveryGate.promise;
      },
      taskLimits: { recoveryTimeoutMs: 10_000, drainTimeoutMs: 20 },
    });
    await harness.initialize();

    const recovering = harness.coordinator.recover(STREAM_ID, "SEQUENCE_GAP");
    await new Promise((resolve) => setTimeout(resolve, 50));

    expect(closeResult?.ok).toBe(false);
    expect(closeResult?.ok ? null : closeResult?.error.kind).toBe(
      "IDLE_TIMEOUT",
    );

    recoveryGate.resolve(realtimeSuccess(snapshotCommit("0")));
    await recovering;
  });
});

/**
 * RT-02. A scheduler that cannot install a deadline leaves the wait unbounded.
 * Letting the exception escape turned a typed realtime result into a native
 * rejection and, through the caller's own catch, started a recovery that
 * overlapped the effect still running.
 */
describe("RT-02 a throwing scheduler stays inside the result contract", () => {
  /** Installs deadlines normally until the stream is initialized. */
  function breakableScheduler() {
    const state = { broken: false };
    return {
      state,
      scheduleTimeout: (callback: () => void, delayMs: number) => {
        if (state.broken) throw new TypeError("scheduleTimeout exploded");
        return setTimeout(callback, delayMs);
      },
      clearScheduledTimeout: (handle: unknown) => {
        clearTimeout(handle as ReturnType<typeof setTimeout>);
      },
    };
  }

  it("does not start a recovery that overlaps a pending apply", async () => {
    const applyGate = deferred<RealtimeResult<void>>();
    const scheduler = breakableScheduler();
    let recoveriesAfterApply = 0;
    const harness = createHarness({
      apply: () => applyGate.promise,
      recover: async () => {
        recoveriesAfterApply += 1;
        return realtimeSuccess(snapshotCommit("0"));
      },
      taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 },
      scheduleTimeout: scheduler.scheduleTimeout,
      clearScheduledTimeout: scheduler.clearScheduledTimeout,
    });
    await harness.initialize();
    recoveriesAfterApply = 0;
    scheduler.state.broken = true;

    const accepted = await harness.coordinator.accept(harness.event());

    // The apply is still pending, so no recovery may have run beside it.
    expect(recoveriesAfterApply).toBe(0);
    expect(accepted.ok).toBe(false);
    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING");

    applyGate.resolve(realtimeSuccess(undefined));
  });

  it("returns a typed close failure instead of rejecting natively", async () => {
    const applyGate = deferred<RealtimeResult<void>>();
    const scheduler = breakableScheduler();
    const harness = createHarness({
      apply: () => applyGate.promise,
      taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 },
      scheduleTimeout: scheduler.scheduleTimeout,
      clearScheduledTimeout: scheduler.clearScheduledTimeout,
    });
    await harness.initialize();
    scheduler.state.broken = true;

    const applying = harness.coordinator.accept(harness.event());
    // Let the apply reach the authority before the fence arrives.
    await Promise.resolve();
    await Promise.resolve();
    const closed = await harness.coordinator.close();

    // A drain that could not be bounded is not proof of quiescence.
    expect(closed.ok).toBe(false);
    expect(closed.ok ? null : closed.error.kind).toBe("IDLE_TIMEOUT");

    applyGate.resolve(realtimeSuccess(undefined));
    await applying;
  });

  it("keeps the classified outcome when clearing a timer throws", async () => {
    const harness = createHarness({
      clearScheduledTimeout: () => {
        throw new TypeError("clearScheduledTimeout exploded");
      },
    });
    await harness.initialize();

    await expect(
      harness.coordinator.accept(harness.event()),
    ).resolves.toMatchObject({ ok: true });
    await expect(harness.coordinator.close()).resolves.toMatchObject({
      ok: true,
    });
  });
});