import { describe, expect, it } from 'vitest';
import {
  DurableSessionCoordinator,
  InMemoryDurableSessionStore,
  type DurableSessionIdentity,
} from './durable-session.js';

const IDENTITY: DurableSessionIdentity = {
  agentName: 'Nova',
  sessionId: 'tess-session-1',
  tenantId: 'tenant-1',
  ownerId: 'owner-1',
  providerId: 'fleet',
  runtimeSessionId: 'nova',
};

describe('DurableSessionCoordinator', () => {
  it('reconstructs an exact session identity, pending inbox/outbox, checkpoint, and handoff after a simulated process restart', async () => {
    const store = new InMemoryDurableSessionStore();
    const beforeRestart = new DurableSessionCoordinator(store);

    await beforeRestart.create(IDENTITY);
    await beforeRestart.receive({
      sessionId: IDENTITY.sessionId,
      idempotencyKey: 'ingress-1',
      correlationId: 'correlation-1',
      content: 'continue the session',
    });
    await beforeRestart.enqueueOutbox({
      sessionId: IDENTITY.sessionId,
      idempotencyKey: 'outbox-1',
      correlationId: 'correlation-1',
      channelId: 'cli',
      kind: 'provider.send',
      content: 'resumable response',
    });
    await beforeRestart.checkpoint({
      sessionId: IDENTITY.sessionId,
      checkpointId: 'checkpoint-1',
      cursor: 'cursor-42',
      summary: 'operator asked for recovery proof',
      compactionEpoch: 0,
    });
    await beforeRestart.handoff({
      sessionId: IDENTITY.sessionId,
      handoffId: 'handoff-1',
      destination: 'mos',
      correlationId: 'correlation-1',
      checkpointId: 'checkpoint-1',
      status: 'pending',
    });

    // Simulate an ungraceful process death: no in-memory coordinator state survives.
    const afterRestart = new DurableSessionCoordinator(store);
    const recovered = await afterRestart.recover(IDENTITY.sessionId);

    expect(recovered.identity).toEqual(IDENTITY);
    expect(recovered.inbox).toMatchObject([{ idempotencyKey: 'ingress-1', status: 'pending' }]);
    expect(recovered.outbox).toMatchObject([{ idempotencyKey: 'outbox-1', status: 'pending' }]);
    expect(recovered.checkpoint).toMatchObject({
      checkpointId: 'checkpoint-1',
      cursor: 'cursor-42',
    });
    expect(recovered.handoffs).toMatchObject([{ handoffId: 'handoff-1', status: 'pending' }]);
  });

  it('rebinds a recovered runtime while preserving the immutable conversation owner scope', async () => {
    const coordinator = new DurableSessionCoordinator(new InMemoryDurableSessionStore());
    await coordinator.create(IDENTITY);
    await coordinator.create({
      ...IDENTITY,
      providerId: 'fleet-next',
      runtimeSessionId: 'nova-next',
    });

    await expect(coordinator.snapshot(IDENTITY.sessionId)).resolves.toMatchObject({
      identity: { ...IDENTITY, providerId: 'fleet-next', runtimeSessionId: 'nova-next' },
    });
    await expect(coordinator.create({ ...IDENTITY, ownerId: 'other-owner' })).rejects.toThrow(
      /identity conflict/,
    );
  });

  it('deduplicates duplicate ingress and never reprocesses an inbox record after restart or compaction', async () => {
    const store = new InMemoryDurableSessionStore();
    const firstProcess = new DurableSessionCoordinator(store);
    const handled: string[] = [];

    await firstProcess.create(IDENTITY);
    await expect(
      firstProcess.receive({
        sessionId: IDENTITY.sessionId,
        idempotencyKey: 'ingress-duplicate',
        correlationId: 'correlation-2',
        content: 'only process me once',
      }),
    ).resolves.toMatchObject({ accepted: true });
    await expect(
      firstProcess.receive({
        sessionId: IDENTITY.sessionId,
        idempotencyKey: 'ingress-duplicate',
        correlationId: 'correlation-2',
        content: 'only process me once',
      }),
    ).resolves.toMatchObject({ accepted: false, status: 'pending' });
    await expect(
      firstProcess.receive({
        sessionId: IDENTITY.sessionId,
        idempotencyKey: 'ingress-duplicate',
        correlationId: 'forged-correlation',
        content: 'only process me once',
      }),
    ).rejects.toThrow(/idempotency conflict/);

    await firstProcess.drainInbox(IDENTITY.sessionId, async (entry) => {
      handled.push(entry.idempotencyKey);
    });
    await firstProcess.checkpoint({
      sessionId: IDENTITY.sessionId,
      checkpointId: 'checkpoint-after-inbox',
      cursor: 'cursor-43',
      summary: 'safe to compact',
      compactionEpoch: 1,
    });

    const afterRestartAndCompaction = new DurableSessionCoordinator(store);
    await afterRestartAndCompaction.recover(IDENTITY.sessionId);
    await afterRestartAndCompaction.drainInbox(IDENTITY.sessionId, async (entry) => {
      handled.push(entry.idempotencyKey);
    });

    expect(handled).toEqual(['ingress-duplicate']);
  });

  it('does not redispatch an already applied outbox side effect after replay, restart, or compaction', async () => {
    const store = new InMemoryDurableSessionStore();
    const beforeRestart = new DurableSessionCoordinator(store);
    const appliedEffects: string[] = [];

    await beforeRestart.create(IDENTITY);
    await beforeRestart.enqueueOutbox({
      sessionId: IDENTITY.sessionId,
      idempotencyKey: 'effect-1',
      correlationId: 'correlation-3',
      channelId: 'cli',
      kind: 'provider.send',
      content: 'send exactly once',
    });
    await expect(
      beforeRestart.enqueueOutbox({
        sessionId: IDENTITY.sessionId,
        idempotencyKey: 'effect-1',
        correlationId: 'correlation-3',
        channelId: 'cli',
        kind: 'provider.send',
        content: 'send exactly once',
      }),
    ).resolves.toMatchObject({ accepted: false, status: 'pending' });

    await beforeRestart.dispatchOutbox(IDENTITY.sessionId, async (entry) => {
      appliedEffects.push(entry.idempotencyKey);
    });
    await beforeRestart.checkpoint({
      sessionId: IDENTITY.sessionId,
      checkpointId: 'checkpoint-after-effect',
      cursor: 'cursor-44',
      summary: 'effect persisted before compaction',
      compactionEpoch: 1,
    });

    const afterRestartAndCompaction = new DurableSessionCoordinator(store);
    await afterRestartAndCompaction.recover(IDENTITY.sessionId);
    await afterRestartAndCompaction.dispatchOutbox(IDENTITY.sessionId, async (entry) => {
      appliedEffects.push(entry.idempotencyKey);
    });

    expect(appliedEffects).toEqual(['effect-1']);
  });

  it('rejects outbox idempotency-key reuse when immutable effect data differs', async () => {
    const store = new InMemoryDurableSessionStore();
    const coordinator = new DurableSessionCoordinator(store);

    await coordinator.create(IDENTITY);
    await coordinator.enqueueOutbox({
      sessionId: IDENTITY.sessionId,
      idempotencyKey: 'outbox-conflict',
      correlationId: 'correlation-outbox',
      channelId: 'cli',
      kind: 'provider.send',
      content: 'original effect',
    });
    await expect(
      coordinator.enqueueOutbox({
        sessionId: IDENTITY.sessionId,
        idempotencyKey: 'outbox-conflict',
        correlationId: 'correlation-outbox',
        channelId: 'forged-channel',
        kind: 'provider.send',
        content: 'original effect',
      }),
    ).rejects.toThrow(/idempotency conflict/);
  });

  it('retains an ambiguous failed provider effect as processing until explicit recovery', async () => {
    const store = new InMemoryDurableSessionStore();
    const beforeRestart = new DurableSessionCoordinator(store);

    await beforeRestart.create(IDENTITY);
    await beforeRestart.enqueueOutbox({
      sessionId: IDENTITY.sessionId,
      idempotencyKey: 'ambiguous-effect',
      correlationId: 'correlation-ambiguous',
      channelId: 'cli',
      kind: 'provider.send',
      content: 'preserve this effect claim',
    });
    await expect(
      beforeRestart.dispatchOutbox(IDENTITY.sessionId, async (): Promise<void> => {
        throw new Error('provider connection dropped after submit');
      }),
    ).rejects.toThrow(/connection dropped/);

    const afterRestart = new DurableSessionCoordinator(store);
    await afterRestart.recover(IDENTITY.sessionId);
    const calls: string[] = [];
    await afterRestart.dispatchOutbox(IDENTITY.sessionId, async (entry): Promise<void> => {
      calls.push(entry.idempotencyKey);
    });

    expect(calls).toEqual([]);
    expect(await afterRestart.snapshot(IDENTITY.sessionId)).toMatchObject({
      outbox: [{ idempotencyKey: 'ambiguous-effect', status: 'processing' }],
    });
  });

  it('rejects a handoff-id replay with different immutable state', async () => {
    const store = new InMemoryDurableSessionStore();
    const coordinator = new DurableSessionCoordinator(store);

    await coordinator.create(IDENTITY);
    await coordinator.checkpoint({
      sessionId: IDENTITY.sessionId,
      checkpointId: 'checkpoint-conflict',
      cursor: 'cursor-conflict',
      summary: 'handoff conflict proof',
      compactionEpoch: 0,
    });
    await coordinator.handoff({
      sessionId: IDENTITY.sessionId,
      handoffId: 'handoff-conflict',
      destination: 'mos',
      correlationId: 'correlation-conflict',
      checkpointId: 'checkpoint-conflict',
      status: 'pending',
    });

    await expect(
      coordinator.handoff({
        sessionId: IDENTITY.sessionId,
        handoffId: 'handoff-conflict',
        destination: 'forged-destination',
        correlationId: 'correlation-conflict',
        checkpointId: 'checkpoint-conflict',
        status: 'pending',
      }),
    ).rejects.toThrow(/identity conflict/);
  });

  it('keeps a handoff portable and resumes it from its durable checkpoint without process-local references', async () => {
    const store = new InMemoryDurableSessionStore();
    const source = new DurableSessionCoordinator(store);

    await source.create(IDENTITY);
    await source.checkpoint({
      sessionId: IDENTITY.sessionId,
      checkpointId: 'checkpoint-handoff',
      cursor: 'cursor-45',
      summary: 'portable state',
      compactionEpoch: 2,
    });
    await source.handoff({
      sessionId: IDENTITY.sessionId,
      handoffId: 'handoff-portable',
      destination: 'mos',
      correlationId: 'correlation-4',
      checkpointId: 'checkpoint-handoff',
      status: 'pending',
    });
    await source.checkpoint({
      sessionId: IDENTITY.sessionId,
      checkpointId: 'checkpoint-later',
      cursor: 'cursor-46',
      summary: 'newer compacted state must not strand the handoff',
      compactionEpoch: 3,
    });

    const receivingProcess = new DurableSessionCoordinator(store);
    const handoff = await receivingProcess.resumeHandoff('handoff-portable');

    expect(handoff.identity).toEqual(IDENTITY);
    expect(handoff.checkpoint).toMatchObject({ checkpointId: 'checkpoint-handoff' });
    expect(handoff.handoff).toMatchObject({ handoffId: 'handoff-portable', destination: 'mos' });
  });
});