import { containerTest, assertNonNullable } from "@internal/testcontainers";
import { trace } from "@internal/tracing";
import { expect } from "vitest";
import { Decimal } from "@trigger.dev/database";
import { RunEngine } from "../index.js";
import { setTimeout } from "timers/promises";
import { EventBusEventArgs } from "../eventBus.js";
import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js";

vi.setConfig({ testTimeout: 60_000 });

describe("RunEngine ttl", () => {
  containerTest("Run expiring (ttl)", async ({ prisma, redisOptions }) => {
    //create environment
    const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

    const engine = new RunEngine({
      prisma,
      worker: {
        redis: redisOptions,
        workers: 1,
        tasksPerWorker: 10,
        pollIntervalMs: 100,
      },
      queue: {
        redis: redisOptions,
        processWorkerQueueDebounceMs: 50,
        masterQueueConsumersDisabled: true,
        ttlSystem: {
          pollIntervalMs: 100,
          batchSize: 10,
          batchMaxWaitMs: 100,
        },
      },
      runLock: {
        redis: redisOptions,
      },
      machines: {
        defaultMachine: "small-1x",
        machines: {
          "small-1x": {
            name: "small-1x" as const,
            cpu: 0.5,
            memory: 0.5,
            centsPerMs: 0.0001,
          },
        },
        baseCostInCents: 0.0001,
      },
      tracer: trace.getTracer("test", "0.0.0"),
    });

    try {
      const taskIdentifier = "test-task";

      //create background worker
      const backgroundWorker = await setupBackgroundWorker(
        engine,
        authenticatedEnvironment,
        taskIdentifier
      );

      // TTL only expires runs still queued waiting on a concurrency slot.
      // Force env concurrency to 0 so the run never gets dequeued and stays
      // in the TTL set long enough for the consumer to expire it.
      await engine.runQueue.updateEnvConcurrencyLimits({
        ...authenticatedEnvironment,
        maximumConcurrencyLimit: 0,
      });

      //trigger the run
      const run = await engine.trigger(
        {
          number: 1,
          friendlyId: "run_1234",
          environment: authenticatedEnvironment,
          taskIdentifier,
          payload: "{}",
          payloadType: "application/json",
          context: {},
          traceContext: {},
          traceId: "t12345",
          spanId: "s12345",
          workerQueue: "main",
          queue: "task/test-task",
          isTest: false,
          tags: [],
          ttl: "1s",
        },
        prisma
      );

      const executionData = await engine.getRunExecutionData({ runId: run.id });
      assertNonNullable(executionData);
      expect(executionData.snapshot.executionStatus).toBe("QUEUED");

      let expiredEventData: EventBusEventArgs<"runExpired">[0] | undefined = undefined;
      engine.eventBus.on("runExpired", (result) => {
        expiredEventData = result;
      });

      //wait for 1 seconds
      await setTimeout(1_500);

      assertNonNullable(expiredEventData);
      const assertedExpiredEventData = expiredEventData as EventBusEventArgs<"runExpired">[0];
      expect(assertedExpiredEventData.run.spanId).toBe(run.spanId);

      // Check the run status directly from the database (the batch TTL path
      // does not create execution snapshots, so getRunExecutionData may not reflect it)
      const expiredRun = await prisma.taskRun.findUnique({
        where: { id: run.id },
        select: { status: true },
      });
      expect(expiredRun?.status).toBe("EXPIRED");

      //concurrency should have been released
      const envConcurrencyCompleted =
        await engine.runQueue.currentConcurrencyOfEnvironment(authenticatedEnvironment);
      expect(envConcurrencyCompleted).toBe(0);

      // Queue sorted set should be empty (run removed from queue)
      const queueLength = await engine.runQueue.lengthOfQueue(
        authenticatedEnvironment,
        "task/test-task"
      );
      expect(queueLength).toBe(0);

      // Env queue sorted set should be empty
      const envQueueLength = await engine.runQueue.lengthOfEnvQueue(authenticatedEnvironment);
      expect(envQueueLength).toBe(0);

      // Message key should be deleted
      const messageExists = await engine.runQueue.messageExists(
        authenticatedEnvironment.organization.id,
        run.id
      );
      expect(messageExists).toBe(0);
    } finally {
      await engine.quit();
    }
  });

  containerTest(
    "First enqueue from trigger includes ttlExpiresAt in message",
    async ({ prisma, redisOptions }) => {
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          processWorkerQueueDebounceMs: 50,
          masterQueueConsumersDisabled: true,
          ttlSystem: {
            pollIntervalMs: 100,
            batchSize: 10,
            batchMaxWaitMs: 100,
          },
        },
        runLock: {
          redis: redisOptions,
        },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": {
              name: "small-1x" as const,
              cpu: 0.5,
              memory: 0.5,
              centsPerMs: 0.0001,
            },
          },
          baseCostInCents: 0.0001,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const taskIdentifier = "test-task";
        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

        const run = await engine.trigger(
          {
            number: 1,
            friendlyId: "run_ttlmsg1",
            environment: authenticatedEnvironment,
            taskIdentifier,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t_ttl",
            spanId: "s_ttl",
            workerQueue: "main",
            queue: "task/test-task",
            isTest: false,
            tags: [],
            ttl: "1s",
          },
          prisma
        );

        const message = await engine.runQueue.readMessage(
          authenticatedEnvironment.organization.id,
          run.id
        );
        assertNonNullable(message);
        expect(message.ttlExpiresAt).toBeDefined();
        expect(typeof message.ttlExpiresAt).toBe("number");
      } finally {
        await engine.quit();
      }
    }
  );

  containerTest(
    "Re-enqueue with includeTtl false does not set ttlExpiresAt",
    async ({ prisma, redisOptions }) => {
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          processWorkerQueueDebounceMs: 50,
          masterQueueConsumersDisabled: true,
          ttlSystem: {
            pollIntervalMs: 100,
            batchSize: 10,
            batchMaxWaitMs: 100,
          },
        },
        runLock: {
          redis: redisOptions,
        },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": {
              name: "small-1x" as const,
              cpu: 0.5,
              memory: 0.5,
              centsPerMs: 0.0001,
            },
          },
          baseCostInCents: 0.0001,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const taskIdentifier = "test-task";
        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

        const run = await engine.trigger(
          {
            number: 1,
            friendlyId: "run_reenq01",
            environment: authenticatedEnvironment,
            taskIdentifier,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t_re",
            spanId: "s_re",
            workerQueue: "main",
            queue: "task/test-task",
            isTest: false,
            tags: [],
            ttl: "1s",
          },
          prisma
        );

        const messageAfterTrigger = await engine.runQueue.readMessage(
          authenticatedEnvironment.organization.id,
          run.id
        );
        assertNonNullable(messageAfterTrigger);
        expect(messageAfterTrigger.ttlExpiresAt).toBeDefined();

        await engine.enqueueSystem.enqueueRun({
          run,
          env: authenticatedEnvironment,
          tx: prisma,
          skipRunLock: true,
          includeTtl: false,
        });

        const messageAfterReenqueue = await engine.runQueue.readMessage(
          authenticatedEnvironment.organization.id,
          run.id
        );
        assertNonNullable(messageAfterReenqueue);
        expect(messageAfterReenqueue.ttlExpiresAt).toBeUndefined();
      } finally {
        await engine.quit();
      }
    }
  );

  containerTest(
    "Re-enqueued runs are not expired by TTL once they have started",
    async ({ prisma, redisOptions }) => {
      // Contract: TTL only applies to runs that are queued and have never started.
      // Once a run has been dequeued (started executing), a subsequent re-enqueue
      // (e.g. after a waitpoint, checkpoint resume, or pending-version flow)
      // must not re-arm TTL, even if the original TTL deadline has long passed.
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

      const expiredEvents: EventBusEventArgs<"runExpired">[0][] = [];

      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          processWorkerQueueDebounceMs: 50,
          masterQueueConsumersDisabled: true,
          ttlSystem: {
            pollIntervalMs: 100,
            batchSize: 10,
            batchMaxWaitMs: 100,
          },
        },
        runLock: {
          redis: redisOptions,
        },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": {
              name: "small-1x" as const,
              cpu: 0.5,
              memory: 0.5,
              centsPerMs: 0.0001,
            },
          },
          baseCostInCents: 0.0001,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const taskIdentifier = "test-task";
        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

        engine.eventBus.on("runExpired", (result) => {
          expiredEvents.push(result);
        });

        const run = await engine.trigger(
          {
            number: 1,
            friendlyId: "run_restart01",
            environment: authenticatedEnvironment,
            taskIdentifier,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t_re2",
            spanId: "s_re2",
            workerQueue: "main",
            queue: "task/test-task",
            isTest: false,
            tags: [],
            ttl: "1s",
          },
          prisma
        );

        // Dequeue the run — this simulates the run starting to execute, which
        // ZREMs its TTL set entry.
        await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 10);
        const dequeued = await engine.dequeueFromWorkerQueue({
          consumerId: "test-consumer",
          workerQueue: "main",
          blockingPopTimeoutSeconds: 1,
        });
        expect(dequeued.length).toBe(1);

        // Re-enqueue without includeTtl — this is what waitpoint/checkpoint
        // resume paths do.
        await engine.enqueueSystem.enqueueRun({
          run,
          env: authenticatedEnvironment,
          tx: prisma,
          skipRunLock: true,
          includeTtl: false,
        });

        // Wait well past the original 1s TTL deadline. The run was first
        // enqueued ~0s ago, so this is far beyond the original deadline.
        await setTimeout(2_500);

        // Run must still exist and must NOT have been expired.
        expect(expiredEvents.length).toBe(0);
        const reenqueuedRun = await prisma.taskRun.findUnique({
          where: { id: run.id },
          select: { status: true },
        });
        // Whatever status the dequeue/re-enqueue flow leaves the run in, it
        // must NOT be EXPIRED — that's the contract this test locks in.
        expect(reenqueuedRun?.status).not.toBe("EXPIRED");
      } finally {
        await engine.quit();
      }
    }
  );

  containerTest(
    "DEV runs sitting on worker queue still expire via legacy per-run job",
    async ({ prisma, redisOptions }) => {
      // The batch TTL path only expires runs still in the queue sorted set.
      // In DEV, runs are fast-pathed straight to the worker queue, and if the
      // dev CLI isn't running they can sit there forever. The legacy per-run
      // expireRun job is kept for DEV specifically to cover this case.
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "DEVELOPMENT");

      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          processWorkerQueueDebounceMs: 50,
          masterQueueConsumersDisabled: true,
          // TTL batch path is enabled but should never see this run: it goes
          // straight to the worker queue via fast-path. The legacy per-run
          // job is what should expire it.
          ttlSystem: {
            pollIntervalMs: 100,
            batchSize: 10,
            batchMaxWaitMs: 100,
          },
        },
        runLock: {
          redis: redisOptions,
        },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": {
              name: "small-1x" as const,
              cpu: 0.5,
              memory: 0.5,
              centsPerMs: 0.0001,
            },
          },
          baseCostInCents: 0.0001,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const taskIdentifier = "test-task";

        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

        let expiredEventData: EventBusEventArgs<"runExpired">[0] | undefined;
        engine.eventBus.on("runExpired", (result) => {
          expiredEventData = result;
        });

        // Trigger a DEV run with fast-path enabled and a short TTL. The run
        // should land in the worker queue without entering the TTL set.
        const run = await engine.trigger(
          {
            number: 1,
            friendlyId: "run_devttl1",
            environment: authenticatedEnvironment,
            taskIdentifier,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "tdevttl1",
            spanId: "sdevttl1",
            workerQueue: "main",
            queue: "task/test-task",
            isTest: false,
            tags: [],
            ttl: "1s",
            enableFastPath: true,
          },
          prisma
        );

        // Wait past the TTL. The legacy per-run job should fire and expire it.
        await setTimeout(1_500);

        assertNonNullable(expiredEventData);
        const expiredRun = await prisma.taskRun.findUnique({
          where: { id: run.id },
          select: { status: true },
        });
        expect(expiredRun?.status).toBe("EXPIRED");
      } finally {
        await engine.quit();
      }
    }
  );

  containerTest("Multiple runs expiring via TTL batch", async ({ prisma, redisOptions }) => {
    const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

    const expiredEvents: EventBusEventArgs<"runExpired">[0][] = [];

    const engine = new RunEngine({
      prisma,
      worker: {
        redis: redisOptions,
        workers: 1,
        tasksPerWorker: 10,
        pollIntervalMs: 100,
      },
      queue: {
        redis: redisOptions,
        processWorkerQueueDebounceMs: 50,
        masterQueueConsumersDisabled: true,
        ttlSystem: {
          pollIntervalMs: 100,
          batchSize: 10,
          batchMaxWaitMs: 100,
        },
      },
      runLock: {
        redis: redisOptions,
      },
      machines: {
        defaultMachine: "small-1x",
        machines: {
          "small-1x": {
            name: "small-1x" as const,
            cpu: 0.5,
            memory: 0.5,
            centsPerMs: 0.0001,
          },
        },
        baseCostInCents: 0.0001,
      },
      tracer: trace.getTracer("test", "0.0.0"),
    });

    try {
      const taskIdentifier = "test-task";

      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      // TTL only expires runs still queued waiting on a concurrency slot.
      await engine.runQueue.updateEnvConcurrencyLimits({
        ...authenticatedEnvironment,
        maximumConcurrencyLimit: 0,
      });

      engine.eventBus.on("runExpired", (result) => {
        expiredEvents.push(result);
      });

      // Trigger multiple runs with short TTL
      const runs = await Promise.all(
        [1, 2, 3].map((n) =>
          engine.trigger(
            {
              number: n,
              friendlyId: `run_b${n}234`,
              environment: authenticatedEnvironment,
              taskIdentifier,
              payload: "{}",
              payloadType: "application/json",
              context: {},
              traceContext: {},
              traceId: `t${n}`,
              spanId: `s${n}`,
              workerQueue: "main",
              queue: "task/test-task",
              isTest: false,
              tags: [],
              ttl: "1s",
            },
            prisma
          )
        )
      );

      // Verify all runs are queued
      for (const run of runs) {
        const executionData = await engine.getRunExecutionData({ runId: run.id });
        assertNonNullable(executionData);
        expect(executionData.snapshot.executionStatus).toBe("QUEUED");
      }

      // Wait for TTL to expire. Concurrent triggers can land in different
      // 100ms TTL-poll windows, so allow enough headroom for any stragglers
      // to be claimed in a subsequent poll and flushed.
      await setTimeout(2_500);

      // All runs should be expired
      expect(expiredEvents.length).toBe(3);

      // Check the run status directly from the database (the batch TTL path
      // does not create execution snapshots, so getRunExecutionData may not reflect it)
      for (const run of runs) {
        const expiredRun = await prisma.taskRun.findUnique({
          where: { id: run.id },
          select: { status: true },
        });
        expect(expiredRun?.status).toBe("EXPIRED");
      }

      // Concurrency should be released for all
      const envConcurrency =
        await engine.runQueue.currentConcurrencyOfEnvironment(authenticatedEnvironment);
      expect(envConcurrency).toBe(0);

      // Queue sorted set should be empty (all runs removed from queue)
      const queueLength = await engine.runQueue.lengthOfQueue(
        authenticatedEnvironment,
        "task/test-task"
      );
      expect(queueLength).toBe(0);

      // Env queue sorted set should be empty
      const envQueueLength = await engine.runQueue.lengthOfEnvQueue(authenticatedEnvironment);
      expect(envQueueLength).toBe(0);

      // All message keys should be deleted
      for (const run of runs) {
        const messageExists = await engine.runQueue.messageExists(
          authenticatedEnvironment.organization.id,
          run.id
        );
        expect(messageExists).toBe(0);
      }
    } finally {
      await engine.quit();
    }
  });

  containerTest("Run without TTL does not expire", async ({ prisma, redisOptions }) => {
    const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

    const expiredEvents: EventBusEventArgs<"runExpired">[0][] = [];

    const engine = new RunEngine({
      prisma,
      worker: {
        redis: redisOptions,
        workers: 1,
        tasksPerWorker: 10,
        pollIntervalMs: 100,
      },
      queue: {
        redis: redisOptions,
        processWorkerQueueDebounceMs: 50,
        masterQueueConsumersDisabled: true,
        ttlSystem: {
          pollIntervalMs: 100,
          batchSize: 10,
          batchMaxWaitMs: 100,
        },
      },
      runLock: {
        redis: redisOptions,
      },
      machines: {
        defaultMachine: "small-1x",
        machines: {
          "small-1x": {
            name: "small-1x" as const,
            cpu: 0.5,
            memory: 0.5,
            centsPerMs: 0.0001,
          },
        },
        baseCostInCents: 0.0001,
      },
      tracer: trace.getTracer("test", "0.0.0"),
    });

    try {
      const taskIdentifier = "test-task";

      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      engine.eventBus.on("runExpired", (result) => {
        expiredEvents.push(result);
      });

      // Trigger a run WITHOUT TTL
      const run = await engine.trigger(
        {
          number: 1,
          friendlyId: "run_n1234",
          environment: authenticatedEnvironment,
          taskIdentifier,
          payload: "{}",
          payloadType: "application/json",
          context: {},
          traceContext: {},
          traceId: "t1",
          spanId: "s1",
          workerQueue: "main",
          queue: "task/test-task",
          isTest: false,
          tags: [],
          // No TTL specified
        },
        prisma
      );

      // Wait a bit
      await setTimeout(500);

      // Run should still be queued, not expired
      expect(expiredEvents.length).toBe(0);

      const executionData = await engine.getRunExecutionData({ runId: run.id });
      assertNonNullable(executionData);
      expect(executionData.snapshot.executionStatus).toBe("QUEUED");
      expect(executionData.run.status).toBe("PENDING");
    } finally {
      await engine.quit();
    }
  });

  containerTest(
    "TTL consumer expires runs before they can be dequeued",
    async ({ prisma, redisOptions }) => {
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

      const expiredEvents: EventBusEventArgs<"runExpired">[0][] = [];

      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          processWorkerQueueDebounceMs: 50,
          masterQueueConsumersDisabled: true,
          ttlSystem: {
            pollIntervalMs: 100,
            batchSize: 10,
            batchMaxWaitMs: 100,
          },
        },
        runLock: {
          redis: redisOptions,
        },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": {
              name: "small-1x" as const,
              cpu: 0.5,
              memory: 0.5,
              centsPerMs: 0.0001,
            },
          },
          baseCostInCents: 0.0001,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const taskIdentifier = "test-task";

        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

        // TTL only expires runs still queued waiting on a concurrency slot.
        await engine.runQueue.updateEnvConcurrencyLimits({
          ...authenticatedEnvironment,
          maximumConcurrencyLimit: 0,
        });

        engine.eventBus.on("runExpired", (result) => {
          expiredEvents.push(result);
        });

        // Trigger a run with short TTL
        const expiredRun = await engine.trigger(
          {
            number: 1,
            friendlyId: "run_e1234",
            environment: authenticatedEnvironment,
            taskIdentifier,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t1",
            spanId: "s1",
            workerQueue: "main",
            queue: "task/test-task",
            isTest: false,
            tags: [],
            ttl: "1s", // Short TTL
          },
          prisma
        );

        // Wait for TTL to expire and TTL consumer to process it
        await setTimeout(1500);

        // The run should have been expired by the TTL consumer
        expect(expiredEvents.length).toBe(1);
        expect(expiredEvents[0]?.run.id).toBe(expiredRun.id);

        // Check the run status directly from the database (the batch TTL path
        // does not create execution snapshots, so getRunExecutionData may not reflect it)
        const expiredRunData = await prisma.taskRun.findUnique({
          where: { id: expiredRun.id },
          select: { status: true },
        });
        expect(expiredRunData?.status).toBe("EXPIRED");

        // The run should have been removed from the queue by the TTL Lua script
        // So dequeue should return nothing
        const dequeued = await engine.dequeueFromWorkerQueue({
          consumerId: "test-consumer",
          workerQueue: "main",
          maxRunCount: 1,
          backgroundWorkerId: (await prisma.backgroundWorker.findFirst({
            where: { runtimeEnvironmentId: authenticatedEnvironment.id },
          }))!.id,
        });

        expect(dequeued.length).toBe(0);
      } finally {
        await engine.quit();
      }
    }
  );

  containerTest(
    "Dequeue skips TTL-expired runs and TTL consumer expires them",
    async ({ prisma, redisOptions }) => {
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

      const expiredEvents: EventBusEventArgs<"runExpired">[0][] = [];

      // Use masterQueueConsumersDisabled so we can manually trigger dequeue via
      // processMasterQueueForEnvironment. TTL consumers start independently and
      // will expire the run after their poll interval.
      const engine = new RunEngine({
        prisma,
        worker: {
          disabled: true,
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          processWorkerQueueDebounceMs: 50,
          masterQueueConsumersDisabled: true,
          ttlSystem: {
            pollIntervalMs: 5000,
            batchSize: 10,
            batchMaxWaitMs: 100,
          },
        },
        runLock: {
          redis: redisOptions,
        },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": {
              name: "small-1x" as const,
              cpu: 0.5,
              memory: 0.5,
              centsPerMs: 0.0001,
            },
          },
          baseCostInCents: 0.0001,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const taskIdentifier = "test-task";

        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

        engine.eventBus.on("runExpired", (result) => {
          expiredEvents.push(result);
        });

        // Trigger a run with short TTL
        const run = await engine.trigger(
          {
            number: 1,
            friendlyId: "run_dq1234",
            environment: authenticatedEnvironment,
            taskIdentifier,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t1",
            spanId: "s1",
            workerQueue: "main",
            queue: "task/test-task",
            isTest: false,
            tags: [],
            ttl: "1s",
          },
          prisma
        );

        // Verify run is queued
        const executionData = await engine.getRunExecutionData({ runId: run.id });
        assertNonNullable(executionData);
        expect(executionData.snapshot.executionStatus).toBe("QUEUED");

        // Wait for TTL to expire
        await setTimeout(1_500);

        // Manually process the master queue - the dequeue Lua script should
        // encounter the expired message and skip it (removing from queue sorted
        // sets but leaving messageKey and ttlQueueKey for TTL consumer)
        await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 10);

        // Try to dequeue from worker queue - nothing should be there since
        // the expired message was skipped by the Lua script
        const dequeued = await engine.dequeueFromWorkerQueue({
          consumerId: "test-consumer",
          workerQueue: "main",
          blockingPopTimeoutSeconds: 1,
        });
        expect(dequeued.length).toBe(0);

        // The run should still be PENDING in the database (not yet expired by TTL consumer)
        const executionData2 = await engine.getRunExecutionData({ runId: run.id });
        assertNonNullable(executionData2);
        expect(executionData2.run.status).toBe("PENDING");

        // Wait (event-driven) for the TTL consumer to poll and expire the run. pollIntervalMs is
        // 5000ms here so the consumer fires only after the dequeue-skip assertions above; waitFor
        // resolves as soon as the event lands instead of a fixed worst-case sleep.
        await vi.waitFor(() => expect(expiredEvents.length).toBe(1), {
          timeout: 15_000,
          interval: 100,
        });
        expect(expiredEvents[0]?.run.id).toBe(run.id);

        // Check the run status directly from the database (the batch TTL path
        // does not create execution snapshots, so getRunExecutionData may not reflect it)
        const expiredRunData = await prisma.taskRun.findUnique({
          where: { id: run.id },
          select: { status: true },
        });
        expect(expiredRunData?.status).toBe("EXPIRED");

        // Concurrency should be released
        const envConcurrency =
          await engine.runQueue.currentConcurrencyOfEnvironment(authenticatedEnvironment);
        expect(envConcurrency).toBe(0);
      } finally {
        await engine.quit();
      }
    }
  );

  containerTest(
    "TTL expiration clears env concurrency keys with proj segment",
    async ({ prisma, redisOptions }) => {
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

      const engine = new RunEngine({
        prisma,
        worker: {
          disabled: true,
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          processWorkerQueueDebounceMs: 50,
          masterQueueConsumersDisabled: true,
          ttlSystem: {
            pollIntervalMs: 5000,
            batchSize: 10,
            batchMaxWaitMs: 100,
          },
        },
        runLock: {
          redis: redisOptions,
        },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": {
              name: "small-1x" as const,
              cpu: 0.5,
              memory: 0.5,
              centsPerMs: 0.0001,
            },
          },
          baseCostInCents: 0.0001,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const taskIdentifier = "test-task";
        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

        const run = await engine.trigger(
          {
            number: 1,
            friendlyId: "run_envkeys",
            environment: authenticatedEnvironment,
            taskIdentifier,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t1",
            spanId: "s1",
            workerQueue: "main",
            queue: "task/test-task",
            isTest: false,
            tags: [],
            ttl: "1s",
          },
          prisma
        );

        const queue = engine.runQueue.keys.queueKey(authenticatedEnvironment, "task/test-task");
        const envConcurrencyKey = engine.runQueue.keys.envCurrentConcurrencyKeyFromQueue(queue);
        const envDequeuedKey = engine.runQueue.keys.envCurrentDequeuedKeyFromQueue(queue);

        await engine.runQueue.redis.sadd(envConcurrencyKey, run.id);
        await engine.runQueue.redis.sadd(envDequeuedKey, run.id);

        const concurrencyBefore =
          await engine.runQueue.getCurrentConcurrencyOfEnvironment(authenticatedEnvironment);
        expect(concurrencyBefore).toContain(run.id);

        await setTimeout(1_500);
        await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 10);
        // Wait (event-driven) for the TTL consumer to expire the run; resolves as soon as the DB
        // reflects EXPIRED instead of a fixed worst-case sleep (pollIntervalMs is 5000ms here).
        await vi.waitFor(
          async () => {
            const expiredRun = await prisma.taskRun.findUnique({
              where: { id: run.id },
              select: { status: true },
            });
            expect(expiredRun?.status).toBe("EXPIRED");
          },
          { timeout: 15_000, interval: 200 }
        );

        const concurrencyAfter =
          await engine.runQueue.getCurrentConcurrencyOfEnvironment(authenticatedEnvironment);
        expect(concurrencyAfter).not.toContain(run.id);

        const stillInDequeued = await engine.runQueue.redis.sismember(envDequeuedKey, run.id);
        expect(stillInDequeued).toBe(0);
      } finally {
        await engine.quit();
      }
    }
  );

  containerTest(
    "Dequeue returns non-expired runs while skipping expired ones",
    async ({ prisma, redisOptions }) => {
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

      // Use masterQueueConsumersDisabled so we can manually trigger dequeue.
      // Very long TTL consumer interval so it doesn't interfere.
      const engine = new RunEngine({
        prisma,
        worker: {
          disabled: true,
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          processWorkerQueueDebounceMs: 50,
          masterQueueConsumersDisabled: true,
          ttlSystem: {
            pollIntervalMs: 30000, // Very long so TTL consumer doesn't interfere
            batchSize: 10,
          },
        },
        runLock: {
          redis: redisOptions,
        },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": {
              name: "small-1x" as const,
              cpu: 0.5,
              memory: 0.5,
              centsPerMs: 0.0001,
            },
          },
          baseCostInCents: 0.0001,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const taskIdentifier = "test-task";

        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

        // Trigger a run with short TTL (will expire)
        const expiringRun = await engine.trigger(
          {
            number: 1,
            friendlyId: "run_exp1",
            environment: authenticatedEnvironment,
            taskIdentifier,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t1",
            spanId: "s1",
            workerQueue: "main",
            queue: "task/test-task",
            isTest: false,
            tags: [],
            ttl: "1s",
          },
          prisma
        );

        // Wait for first run's TTL to expire
        await setTimeout(1_500);

        // Trigger a second run WITHOUT TTL (should be dequeued normally)
        const normalRun = await engine.trigger(
          {
            number: 2,
            friendlyId: "run_norm1",
            environment: authenticatedEnvironment,
            taskIdentifier,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t2",
            spanId: "s2",
            workerQueue: "main",
            queue: "task/test-task",
            isTest: false,
            tags: [],
            // No TTL
          },
          prisma
        );

        // Manually process the master queue - the Lua script should skip the
        // expired message and dequeue only the non-expired one to the worker queue
        await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 10);

        // Dequeue from worker queue - only the non-expired run should be there
        const dequeued = await engine.dequeueFromWorkerQueue({
          consumerId: "test-consumer",
          workerQueue: "main",
        });
        expect(dequeued.length).toBe(1);
        expect(dequeued[0]?.run.id).toBe(normalRun.id);

        // The expired run should still be PENDING (waiting for TTL consumer)
        const expiringRunData = await engine.getRunExecutionData({ runId: expiringRun.id });
        assertNonNullable(expiringRunData);
        expect(expiringRunData.run.status).toBe("PENDING");
      } finally {
        await engine.quit();
      }
    }
  );

  containerTest("expireRunsBatch skips runs that are locked", async ({ prisma, redisOptions }) => {
    const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

    const engine = new RunEngine({
      prisma,
      worker: {
        redis: redisOptions,
        workers: 1,
        tasksPerWorker: 10,
        pollIntervalMs: 100,
      },
      queue: {
        redis: redisOptions,
        processWorkerQueueDebounceMs: 50,
        masterQueueConsumersDisabled: true,
        ttlSystem: {
          disabled: true, // We'll manually test the batch function
        },
      },
      runLock: {
        redis: redisOptions,
      },
      machines: {
        defaultMachine: "small-1x",
        machines: {
          "small-1x": {
            name: "small-1x" as const,
            cpu: 0.5,
            memory: 0.5,
            centsPerMs: 0.0001,
          },
        },
        baseCostInCents: 0.0001,
      },
      tracer: trace.getTracer("test", "0.0.0"),
    });

    try {
      const taskIdentifier = "test-task";

      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      // Trigger a run with TTL
      const run = await engine.trigger(
        {
          number: 1,
          friendlyId: "run_l1234",
          environment: authenticatedEnvironment,
          taskIdentifier,
          payload: "{}",
          payloadType: "application/json",
          context: {},
          traceContext: {},
          traceId: "t1",
          spanId: "s1",
          workerQueue: "main",
          queue: "task/test-task",
          isTest: false,
          tags: [],
          ttl: "1s",
        },
        prisma
      );

      // Manually lock the run (simulating it being about to execute)
      await prisma.taskRun.update({
        where: { id: run.id },
        data: { lockedAt: new Date() },
      });

      // Try to expire the run via batch
      const result = await engine.ttlSystem.expireRunsBatch([run.id]);

      // Should be skipped because it's locked
      expect(result.expired.length).toBe(0);
      expect(result.skipped.length).toBe(1);
      expect(result.skipped[0]?.reason).toBe("locked");

      // Run should still be PENDING
      const executionData = await engine.getRunExecutionData({ runId: run.id });
      assertNonNullable(executionData);
      expect(executionData.run.status).toBe("PENDING");
    } finally {
      await engine.quit();
    }
  });

  containerTest(
    "expireRunsBatch skips runs with non-PENDING status",
    async ({ prisma, redisOptions }) => {
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          processWorkerQueueDebounceMs: 50,
          masterQueueConsumersDisabled: true,
          ttlSystem: {
            disabled: true,
          },
        },
        runLock: {
          redis: redisOptions,
        },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": {
              name: "small-1x" as const,
              cpu: 0.5,
              memory: 0.5,
              centsPerMs: 0.0001,
            },
          },
          baseCostInCents: 0.0001,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const taskIdentifier = "test-task";

        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

        // Trigger a run with TTL
        const run = await engine.trigger(
          {
            number: 1,
            friendlyId: "run_x1234",
            environment: authenticatedEnvironment,
            taskIdentifier,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t1",
            spanId: "s1",
            workerQueue: "main",
            queue: "task/test-task",
            isTest: false,
            tags: [],
            ttl: "1s",
          },
          prisma
        );

        // Manually change status to EXECUTING (simulating the run started)
        await prisma.taskRun.update({
          where: { id: run.id },
          data: { status: "EXECUTING" },
        });

        // Try to expire the run via batch
        const result = await engine.ttlSystem.expireRunsBatch([run.id]);

        // Should be skipped because it's not PENDING
        expect(result.expired.length).toBe(0);
        expect(result.skipped.length).toBe(1);
        expect(result.skipped[0]?.reason).toBe("status_EXECUTING");

        // Run should still be EXECUTING
        const dbRun = await prisma.taskRun.findUnique({ where: { id: run.id } });
        expect(dbRun?.status).toBe("EXECUTING");
      } finally {
        await engine.quit();
      }
    }
  );

  containerTest("expireRunsBatch handles non-existent runs", async ({ prisma, redisOptions }) => {
    const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

    const engine = new RunEngine({
      prisma,
      worker: {
        redis: redisOptions,
        workers: 1,
        tasksPerWorker: 10,
        pollIntervalMs: 100,
      },
      queue: {
        redis: redisOptions,
        processWorkerQueueDebounceMs: 50,
        masterQueueConsumersDisabled: true,
        ttlSystem: {
          disabled: true,
        },
      },
      runLock: {
        redis: redisOptions,
      },
      machines: {
        defaultMachine: "small-1x",
        machines: {
          "small-1x": {
            name: "small-1x" as const,
            cpu: 0.5,
            memory: 0.5,
            centsPerMs: 0.0001,
          },
        },
        baseCostInCents: 0.0001,
      },
      tracer: trace.getTracer("test", "0.0.0"),
    });

    try {
      // Try to expire a non-existent run
      const result = await engine.ttlSystem.expireRunsBatch(["non_existent_run_id"]);

      // Should be skipped as not found
      expect(result.expired.length).toBe(0);
      expect(result.skipped.length).toBe(1);
      expect(result.skipped[0]?.reason).toBe("not_found");
    } finally {
      await engine.quit();
    }
  });

  containerTest(
    "TTL-expired child run completes waitpoint and resumes parent",
    async ({ prisma, redisOptions }) => {
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          processWorkerQueueDebounceMs: 50,
          masterQueueConsumersDisabled: true,
          ttlSystem: {
            pollIntervalMs: 100,
            batchSize: 10,
            batchMaxWaitMs: 100,
          },
        },
        runLock: {
          redis: redisOptions,
        },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": {
              name: "small-1x" as const,
              cpu: 0.5,
              memory: 0.5,
              centsPerMs: 0.0001,
            },
          },
          baseCostInCents: 0.0001,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const parentTask = "parent-task";
        const childTask = "child-task";

        await setupBackgroundWorker(engine, authenticatedEnvironment, [parentTask, childTask]);

        // TTL only expires runs still queued waiting on a concurrency slot.
        // Cap env concurrency at exactly 1 (limit=1, burstFactor=1) so the
        // parent takes the only slot and the child stays queued long enough
        // for the new TTL path to expire it.
        await engine.runQueue.updateEnvConcurrencyLimits({
          ...authenticatedEnvironment,
          maximumConcurrencyLimit: 1,
          concurrencyLimitBurstFactor: new Decimal(1.0),
        });

        // Trigger the parent run
        const parentRun = await engine.trigger(
          {
            number: 1,
            friendlyId: "run_p1234",
            environment: authenticatedEnvironment,
            taskIdentifier: parentTask,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t12345",
            spanId: "s12345",
            queue: `task/${parentTask}`,
            isTest: false,
            tags: [],
            workerQueue: "main",
          },
          prisma
        );

        // Dequeue and start parent
        await setTimeout(500);
        const dequeued = await engine.dequeueFromWorkerQueue({
          consumerId: "test_12345",
          workerQueue: "main",
        });

        const initialExecutionData = await engine.getRunExecutionData({ runId: parentRun.id });
        assertNonNullable(initialExecutionData);
        await engine.startRunAttempt({
          runId: parentRun.id,
          snapshotId: initialExecutionData.snapshot.id,
        });

        // Trigger child run with TTL and resumeParentOnCompletion
        const childRun = await engine.trigger(
          {
            number: 2,
            friendlyId: "run_c1234",
            environment: authenticatedEnvironment,
            taskIdentifier: childTask,
            payload: "{}",
            payloadType: "application/json",
            context: {},
            traceContext: {},
            traceId: "t12346",
            spanId: "s12346",
            queue: `task/${childTask}`,
            isTest: false,
            tags: [],
            resumeParentOnCompletion: true,
            parentTaskRunId: parentRun.id,
            workerQueue: "main",
            ttl: "1s",
          },
          prisma
        );

        // Verify parent is waiting on child
        const parentExecutionData = await engine.getRunExecutionData({ runId: parentRun.id });
        assertNonNullable(parentExecutionData);
        expect(parentExecutionData.snapshot.executionStatus).toBe("EXECUTING_WITH_WAITPOINTS");

        const runWaitpoint = await prisma.taskRunWaitpoint.findFirst({
          where: { taskRunId: parentRun.id },
          include: { waitpoint: true },
        });
        assertNonNullable(runWaitpoint);
        expect(runWaitpoint.waitpoint.type).toBe("RUN");
        expect(runWaitpoint.waitpoint.completedByTaskRunId).toBe(childRun.id);

        // Wait for TTL to expire + finishWaitpoint worker job to process
        await setTimeout(3_000);

        // Child run should be EXPIRED
        const expiredChild = await prisma.taskRun.findUnique({
          where: { id: childRun.id },
          select: { status: true },
        });
        expect(expiredChild?.status).toBe("EXPIRED");

        // Waitpoint should be completed with error output
        const waitpointAfter = await prisma.waitpoint.findFirst({
          where: { id: runWaitpoint.waitpointId },
        });
        assertNonNullable(waitpointAfter);
        expect(waitpointAfter.status).toBe("COMPLETED");
        expect(waitpointAfter.completedAt).not.toBeNull();
        expect(waitpointAfter.outputIsError).toBe(true);

        // TaskRunWaitpoint linking parent to child should be removed
        const runWaitpointAfter = await prisma.taskRunWaitpoint.findFirst({
          where: { taskRunId: parentRun.id },
        });
        expect(runWaitpointAfter).toBeNull();

        // Parent should be back to EXECUTING
        const parentExecutionDataAfter = await engine.getRunExecutionData({ runId: parentRun.id });
        assertNonNullable(parentExecutionDataAfter);
        expect(parentExecutionDataAfter.snapshot.executionStatus).toBe("EXECUTING");

        // Parent's completedWaitpoints should contain the waitpoint with error output
        expect(parentExecutionDataAfter.completedWaitpoints?.length).toBe(1);
        expect(parentExecutionDataAfter.completedWaitpoints![0].id).toBe(runWaitpoint.waitpointId);
        expect(parentExecutionDataAfter.completedWaitpoints![0].outputIsError).toBe(true);
      } finally {
        await engine.quit();
      }
    }
  );

  containerTest("expireRunsBatch handles empty array", async ({ prisma, redisOptions }) => {
    const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

    const engine = new RunEngine({
      prisma,
      worker: {
        redis: redisOptions,
        workers: 1,
        tasksPerWorker: 10,
        pollIntervalMs: 100,
      },
      queue: {
        redis: redisOptions,
        processWorkerQueueDebounceMs: 50,
        masterQueueConsumersDisabled: true,
        ttlSystem: {
          disabled: true,
        },
      },
      runLock: {
        redis: redisOptions,
      },
      machines: {
        defaultMachine: "small-1x",
        machines: {
          "small-1x": {
            name: "small-1x" as const,
            cpu: 0.5,
            memory: 0.5,
            centsPerMs: 0.0001,
          },
        },
        baseCostInCents: 0.0001,
      },
      tracer: trace.getTracer("test", "0.0.0"),
    });

    try {
      // Try to expire an empty array
      const result = await engine.ttlSystem.expireRunsBatch([]);

      expect(result.expired.length).toBe(0);
      expect(result.skipped.length).toBe(0);
    } finally {
      await engine.quit();
    }
  });
});
