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

// Mock the db prisma client
vi.mock("~/db.server", () => ({
  prisma: {},
  $replica: {},
}));

vi.mock("~/services/platform.v3.server", async (importOriginal) => {
  const actual = (await importOriginal()) as Record<string, unknown>;
  return {
    ...actual,
    getEntitlement: vi.fn(),
  };
});

import { RunEngine } from "@internal/run-engine";
import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "@internal/run-engine/tests";
import { assertNonNullable, containerTest } from "@internal/testcontainers";
import { trace } from "@opentelemetry/api";
import { IOPacket } from "@trigger.dev/core/v3";
import { TaskRun } from "@trigger.dev/database";
import { Redis } from "ioredis";
import { IdempotencyKeyConcern } from "~/runEngine/concerns/idempotencyKeys.server";
import { DefaultQueueManager } from "~/runEngine/concerns/queues.server";
import { NoopTaskMetadataCache, RedisTaskMetadataCache } from "~/services/taskMetadataCache.server";
import {
  EntitlementValidationParams,
  MaxAttemptsValidationParams,
  ParentRunValidationParams,
  PayloadProcessor,
  TagValidationParams,
  TracedEventSpan,
  TraceEventConcern,
  TriggerRacepoints,
  TriggerRacepointSystem,
  TriggerTaskRequest,
  TriggerTaskValidator,
  ValidationResult,
} from "~/runEngine/types";
import { RunEngineTriggerTaskService } from "../../app/runEngine/services/triggerTask.server";
import { promiseWithResolvers } from "@trigger.dev/core";
import { setTimeout } from "node:timers/promises";

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

class MockPayloadProcessor implements PayloadProcessor {
  async process(request: TriggerTaskRequest): Promise<IOPacket> {
    return {
      data: JSON.stringify(request.body.payload),
      dataType: "application/json",
    };
  }
}

class MockTriggerTaskValidator implements TriggerTaskValidator {
  validateTags(params: TagValidationParams): ValidationResult {
    return { ok: true };
  }
  validateEntitlement(params: EntitlementValidationParams): Promise<ValidationResult> {
    return Promise.resolve({ ok: true });
  }
  validateMaxAttempts(params: MaxAttemptsValidationParams): ValidationResult {
    return { ok: true };
  }
  validateParentRun(params: ParentRunValidationParams): ValidationResult {
    return { ok: true };
  }
}

// Mirror the production ClickhouseEventRepository.traceEvent shape so
// callers that read `event.traceContext.traceparent` (e.g. the
// mollifier branch seeding the snapshot) get the same W3C-formatted
// value they'd get against a real event repository.
const MOCK_TRACE_ID = "0123456789abcdef0123456789abcdef";
const MOCK_SPAN_ID = "fedcba9876543210";
const MOCK_TRACEPARENT = `00-${MOCK_TRACE_ID}-${MOCK_SPAN_ID}-01`;

class MockTraceEventConcern implements TraceEventConcern {
  // Records the start time of the most recent traceRun callback entry.
  // Used by ordering assertions that verify traceRun fires before
  // downstream side effects (e.g. mollifier buffer writes).
  public traceRunEnteredAt: number | undefined;

  async traceRun<T>(
    request: TriggerTaskRequest,
    parentStore: string | undefined,
    callback: (span: TracedEventSpan, store: string) => Promise<T>
  ): Promise<T> {
    this.traceRunEnteredAt = Date.now();
    return await callback(
      {
        traceId: MOCK_TRACE_ID,
        spanId: MOCK_SPAN_ID,
        traceContext: { traceparent: MOCK_TRACEPARENT },
        traceparent: undefined,
        setAttribute: () => {},
        failWithError: () => {},
        stop: () => {},
      },
      "test"
    );
  }

  async traceIdempotentRun<T>(
    request: TriggerTaskRequest,
    parentStore: string | undefined,
    options: {
      existingRun: TaskRun;
      idempotencyKey: string;
      incomplete: boolean;
      isError: boolean;
    },
    callback: (span: TracedEventSpan, store: string) => Promise<T>
  ): Promise<T> {
    return await callback(
      {
        traceId: "test",
        spanId: "test",
        traceContext: {},
        traceparent: undefined,
        setAttribute: () => {},
        failWithError: () => {},
        stop: () => {},
      },
      "test"
    );
  }

  async traceDebouncedRun<T>(
    request: TriggerTaskRequest,
    parentStore: string | undefined,
    options: {
      existingRun: TaskRun;
      debounceKey: string;
      incomplete: boolean;
      isError: boolean;
    },
    callback: (span: TracedEventSpan, store: string) => Promise<T>
  ): Promise<T> {
    return await callback(
      {
        traceId: "test",
        spanId: "test",
        traceContext: {},
        traceparent: undefined,
        setAttribute: () => {},
        failWithError: () => {},
        stop: () => {},
      },
      "test"
    );
  }
}

type TriggerRacepoint = { promise: Promise<void>; resolve: (value: void) => void };

class MockTriggerRacepointSystem implements TriggerRacepointSystem {
  private racepoints: Record<string, TriggerRacepoint | undefined> = {};

  async waitForRacepoint({ id }: { racepoint: TriggerRacepoints; id: string }): Promise<void> {
    const racepoint = this.racepoints[id];

    if (racepoint) {
      return racepoint.promise;
    }

    return Promise.resolve();
  }

  registerRacepoint(racepoint: TriggerRacepoints, id: string): TriggerRacepoint {
    const { promise, resolve } = promiseWithResolvers<void>();
    this.racepoints[id] = { promise, resolve };

    return { promise, resolve };
  }
}

describe("RunEngineTriggerTaskService", () => {
  containerTest("should trigger a task with minimal options", async ({ prisma, redisOptions }) => {
    const engine = new RunEngine({
      prisma,
      worker: {
        redis: redisOptions,
        workers: 1,
        tasksPerWorker: 10,
        pollIntervalMs: 100,
      },
      queue: {
        redis: redisOptions,
      },
      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.0005,
      },
      tracer: trace.getTracer("test", "0.0.0"),
    });

    const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

    const taskIdentifier = "test-task";

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

    const queuesManager = new DefaultQueueManager(prisma, engine);

    const idempotencyKeyConcern = new IdempotencyKeyConcern(
      prisma,
      engine,
      new MockTraceEventConcern()
    );

    const triggerTaskService = new RunEngineTriggerTaskService({
      engine,
      prisma,
      payloadProcessor: new MockPayloadProcessor(),
      queueConcern: queuesManager,
      idempotencyKeyConcern,
      validator: new MockTriggerTaskValidator(),
      traceEventConcern: new MockTraceEventConcern(),
      tracer: trace.getTracer("test", "0.0.0"),
      metadataMaximumSize: 1024 * 1024 * 1, // 1MB
    });

    const result = await triggerTaskService.call({
      taskId: taskIdentifier,
      environment: authenticatedEnvironment,
      body: { payload: { test: "test" } },
    });

    expect(result).toBeDefined();
    expect(result?.run.friendlyId).toBeDefined();
    expect(result?.run.status).toBe("PENDING");
    expect(result?.isCached).toBe(false);

    const run = await prisma.taskRun.findUnique({
      where: {
        id: result?.run.id,
      },
    });

    expect(run).toBeDefined();
    expect(run?.friendlyId).toBe(result?.run.friendlyId);
    expect(run?.engine).toBe("V2");
    expect(run?.queuedAt).toBeDefined();
    expect(run?.queue).toBe(`task/${taskIdentifier}`);

    // Lets make sure the task is in the queue
    const queueLength = await engine.runQueue.lengthOfQueue(
      authenticatedEnvironment,
      `task/${taskIdentifier}`
    );
    expect(queueLength).toBe(1);

    await engine.quit();
  });

  containerTest(
    "routes scheduled-lineage runs to a separate worker queue that dequeues independently",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
          // Disable the background master-queue consumers so our manual
          // processMasterQueueForEnvironment + dequeue calls are deterministic.
          masterQueueConsumersDisabled: true,
          processWorkerQueueDebounceMs: 50,
        },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      try {
        const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

        // Turn the per-org split flag on in-memory — the resolver reads this
        // object directly (no DB round-trip on the trigger hot path).
        (authenticatedEnvironment.organization as { featureFlags?: unknown }).featureFlags = {
          workerQueueScheduledSplitEnabled: true,
        };

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

        const triggerTaskService = new RunEngineTriggerTaskService({
          engine,
          prisma,
          payloadProcessor: new MockPayloadProcessor(),
          queueConcern: new DefaultQueueManager(prisma, engine),
          idempotencyKeyConcern: new IdempotencyKeyConcern(
            prisma,
            engine,
            new MockTraceEventConcern()
          ),
          validator: new MockTriggerTaskValidator(),
          traceEventConcern: new MockTraceEventConcern(),
          tracer: trace.getTracer("test", "0.0.0"),
          metadataMaximumSize: 1024 * 1024 * 1,
        });

        // A standard run (default triggerSource) stays on the region queue.
        const standardResult = await triggerTaskService.call({
          taskId: taskIdentifier,
          environment: authenticatedEnvironment,
          body: { payload: { kind: "standard" } },
        });
        assertNonNullable(standardResult);

        // A scheduled run routes to the `<region>:scheduled` queue. Descendants
        // would too, via rootTriggerSource propagation.
        const scheduledResult = await triggerTaskService.call({
          taskId: taskIdentifier,
          environment: authenticatedEnvironment,
          body: { payload: { kind: "scheduled" } },
          options: { triggerSource: "schedule" },
        });
        assertNonNullable(scheduledResult);

        const standardRun = await prisma.taskRun.findUniqueOrThrow({
          where: { id: standardResult.run.id },
        });
        const scheduledRun = await prisma.taskRun.findUniqueOrThrow({
          where: { id: scheduledResult.run.id },
        });

        // Producer routing: the persisted worker queue carries the class.
        const baseWorkerQueue = standardRun.workerQueue;
        expect(scheduledRun.workerQueue).toBe(`${baseWorkerQueue}:scheduled`);

        // Move both runs from the env queue onto their respective worker queues.
        await engine.runQueue.processMasterQueueForEnvironment(authenticatedEnvironment.id, 10);
        await setTimeout(500);

        // Dequeue isolation: the scheduled queue yields only the scheduled run...
        const dequeuedScheduled = await engine.dequeueFromWorkerQueue({
          consumerId: "test-scheduled-consumer",
          workerQueue: `${baseWorkerQueue}:scheduled`,
        });
        expect(dequeuedScheduled.length).toBe(1);
        assertNonNullable(dequeuedScheduled[0]);
        expect(dequeuedScheduled[0].run.id).toBe(scheduledResult.run.id);

        // ...and the base queue yields only the standard run.
        const dequeuedStandard = await engine.dequeueFromWorkerQueue({
          consumerId: "test-standard-consumer",
          workerQueue: baseWorkerQueue,
        });
        expect(dequeuedStandard.length).toBe(1);
        assertNonNullable(dequeuedStandard[0]);
        expect(dequeuedStandard[0].run.id).toBe(standardResult.run.id);
      } finally {
        await engine.quit();
      }
    }
  );

  // The BatchQueue worker rebuilds body.options from Redis-stored items
  // (Record<string, unknown>), so the Phase-2 schema coercion doesn't apply
  // to in-flight items enqueued before the schema fix. The defensive
  // `typeof === "number"` coercion at the engine.trigger call site is what
  // prevents these from failing at prisma.taskRun.create with
  // "Argument concurrencyKey: Expected String or Null, provided Int".
  containerTest(
    "coerces a numeric concurrencyKey to a string at the engine.trigger boundary",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: { redis: redisOptions },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "test-task";
      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: new DefaultQueueManager(prisma, engine),
        idempotencyKeyConcern: new IdempotencyKeyConcern(
          prisma,
          engine,
          new MockTraceEventConcern()
        ),
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024 * 1,
      });

      const result = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        // Cast through `any` to simulate the in-flight Redis batch-item shape
        // (Record<string, unknown>) that bypasses the BatchItemNDJSON schema.
        body: { payload: { userId: 51262 }, options: { concurrencyKey: 51262 as any } },
      });

      expect(result).toBeDefined();
      const run = await prisma.taskRun.findUnique({ where: { id: result!.run.id } });
      expect(run?.concurrencyKey).toBe("51262");

      await engine.quit();
    }
  );

  containerTest("should handle idempotency keys correctly", async ({ prisma, redisOptions }) => {
    const engine = new RunEngine({
      prisma,
      worker: {
        redis: redisOptions,
        workers: 1,
        tasksPerWorker: 10,
        pollIntervalMs: 100,
      },
      queue: {
        redis: redisOptions,
      },
      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.0005,
      },
      tracer: trace.getTracer("test", "0.0.0"),
    });

    const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

    const taskIdentifier = "test-task";

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

    const queuesManager = new DefaultQueueManager(prisma, engine);

    const idempotencyKeyConcern = new IdempotencyKeyConcern(
      prisma,
      engine,
      new MockTraceEventConcern()
    );

    const triggerTaskService = new RunEngineTriggerTaskService({
      engine,
      prisma,
      payloadProcessor: new MockPayloadProcessor(),
      queueConcern: queuesManager,
      idempotencyKeyConcern,
      validator: new MockTriggerTaskValidator(),
      traceEventConcern: new MockTraceEventConcern(),
      tracer: trace.getTracer("test", "0.0.0"),
      metadataMaximumSize: 1024 * 1024 * 1, // 1MB
    });

    const result = await triggerTaskService.call({
      taskId: taskIdentifier,
      environment: authenticatedEnvironment,
      body: {
        payload: { test: "test" },
        options: {
          idempotencyKey: "test-idempotency-key",
        },
      },
    });

    expect(result).toBeDefined();
    expect(result?.run.friendlyId).toBeDefined();
    expect(result?.run.status).toBe("PENDING");
    expect(result?.isCached).toBe(false);

    const run = await prisma.taskRun.findUnique({
      where: {
        id: result?.run.id,
      },
    });

    expect(run).toBeDefined();
    expect(run?.friendlyId).toBe(result?.run.friendlyId);
    expect(run?.engine).toBe("V2");
    expect(run?.queuedAt).toBeDefined();
    expect(run?.queue).toBe(`task/${taskIdentifier}`);

    // Lets make sure the task is in the queue
    const queueLength = await engine.runQueue.lengthOfQueue(
      authenticatedEnvironment,
      `task/${taskIdentifier}`
    );
    expect(queueLength).toBe(1);

    // Now lets try to trigger the same task with the same idempotency key
    const cachedResult = await triggerTaskService.call({
      taskId: taskIdentifier,
      environment: authenticatedEnvironment,
      body: {
        payload: { test: "test" },
        options: {
          idempotencyKey: "test-idempotency-key",
        },
      },
    });

    expect(cachedResult).toBeDefined();
    expect(cachedResult?.run.friendlyId).toBe(result?.run.friendlyId);
    expect(cachedResult?.isCached).toBe(true);

    await engine.quit();
  });

  containerTest(
    "should handle idempotency keys when the engine throws an RunDuplicateIdempotencyKeyError",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
        },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
        logLevel: "debug",
      });

      const parentTask = "parent-task";

      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

      const taskIdentifier = "test-task";

      //create background worker
      await setupBackgroundWorker(engine, authenticatedEnvironment, [parentTask, taskIdentifier]);

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

      //dequeue parent and create the attempt
      await setTimeout(500);
      const dequeued = await engine.dequeueFromWorkerQueue({
        consumerId: "test_12345",
        workerQueue: "main",
      });
      await engine.startRunAttempt({
        runId: parentRun1.id,
        snapshotId: dequeued[0].snapshot.id,
      });

      const parentRun2 = await engine.trigger(
        {
          number: 2,
          friendlyId: "run_p2",
          environment: authenticatedEnvironment,
          taskIdentifier: parentTask,
          payload: "{}",
          payloadType: "application/json",
          context: {},
          traceContext: {},
          traceId: "t12346",
          spanId: "s12346",
          queue: `task/${parentTask}`,
          isTest: false,
          tags: [],
          workerQueue: "main",
        },
        prisma
      );

      await setTimeout(500);
      const dequeued2 = await engine.dequeueFromWorkerQueue({
        consumerId: "test_12345",
        workerQueue: "main",
      });
      await engine.startRunAttempt({
        runId: parentRun2.id,
        snapshotId: dequeued2[0].snapshot.id,
      });

      const queuesManager = new DefaultQueueManager(prisma, engine);

      const idempotencyKeyConcern = new IdempotencyKeyConcern(
        prisma,
        engine,
        new MockTraceEventConcern()
      );

      const triggerRacepointSystem = new MockTriggerRacepointSystem();

      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: queuesManager,
        idempotencyKeyConcern,
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024 * 1, // 1MB
        triggerRacepointSystem,
      });

      const idempotencyKey = "test-idempotency-key";

      const racepoint = triggerRacepointSystem.registerRacepoint("idempotencyKey", idempotencyKey);

      const childTriggerPromise1 = triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: {
          payload: { test: "test" },
          options: {
            idempotencyKey,
            parentRunId: parentRun1.friendlyId,
            resumeParentOnCompletion: true,
          },
        },
      });

      const childTriggerPromise2 = triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: {
          payload: { test: "test" },
          options: {
            idempotencyKey,
            parentRunId: parentRun2.friendlyId,
            resumeParentOnCompletion: true,
          },
        },
      });

      await setTimeout(500);

      // Now we can resolve the racepoint
      racepoint.resolve();

      const result = await childTriggerPromise1;
      const result2 = await childTriggerPromise2;

      expect(result).toBeDefined();
      expect(result?.run.friendlyId).toBeDefined();
      expect(result?.run.status).toBe("PENDING");

      const run = await prisma.taskRun.findUnique({
        where: {
          id: result?.run.id,
        },
      });

      expect(run).toBeDefined();
      expect(run?.friendlyId).toBe(result?.run.friendlyId);
      expect(run?.engine).toBe("V2");
      expect(run?.queuedAt).toBeDefined();
      expect(run?.queue).toBe(`task/${taskIdentifier}`);

      expect(result2).toBeDefined();
      expect(result2?.run.friendlyId).toBe(result?.run.friendlyId);

      const parent1ExecutionData = await engine.getRunExecutionData({ runId: parentRun1.id });
      assertNonNullable(parent1ExecutionData);
      expect(parent1ExecutionData.snapshot.executionStatus).toBe("EXECUTING_WITH_WAITPOINTS");

      const parent2ExecutionData = await engine.getRunExecutionData({ runId: parentRun2.id });
      assertNonNullable(parent2ExecutionData);
      expect(parent2ExecutionData.snapshot.executionStatus).toBe("EXECUTING_WITH_WAITPOINTS");

      const parent1RunWaitpoint = await prisma.taskRunWaitpoint.findFirst({
        where: {
          taskRunId: parentRun1.id,
        },
        include: {
          waitpoint: true,
        },
      });

      assertNonNullable(parent1RunWaitpoint);
      expect(parent1RunWaitpoint.waitpoint.type).toBe("RUN");
      expect(parent1RunWaitpoint.waitpoint.completedByTaskRunId).toBe(result?.run.id);

      const parent2RunWaitpoint = await prisma.taskRunWaitpoint.findFirst({
        where: {
          taskRunId: parentRun2.id,
        },
        include: {
          waitpoint: true,
        },
      });

      assertNonNullable(parent2RunWaitpoint);
      expect(parent2RunWaitpoint.waitpoint.type).toBe("RUN");
      expect(parent2RunWaitpoint.waitpoint.completedByTaskRunId).toBe(result2?.run.id);

      await engine.quit();
    }
  );

  containerTest(
    "should resolve queue names correctly when locked to version",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
        },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "test-task";

      // Create a background worker with a specific version
      const worker = await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier, {
        preset: "small-1x",
      });

      // Create a specific queue for this worker
      const specificQueue = await prisma.taskQueue.create({
        data: {
          name: "specific-queue",
          friendlyId: "specific-queue",
          projectId: authenticatedEnvironment.projectId,
          runtimeEnvironmentId: authenticatedEnvironment.id,
          workers: {
            connect: {
              id: worker.worker.id,
            },
          },
        },
      });

      // Associate the task with the queue
      await prisma.backgroundWorkerTask.update({
        where: {
          workerId_slug: {
            workerId: worker.worker.id,
            slug: taskIdentifier,
          },
        },
        data: {
          queueId: specificQueue.id,
        },
      });

      const queuesManager = new DefaultQueueManager(prisma, engine);
      const idempotencyKeyConcern = new IdempotencyKeyConcern(
        prisma,
        engine,
        new MockTraceEventConcern()
      );

      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: queuesManager,
        idempotencyKeyConcern,
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024 * 1, // 1MB
      });

      // Test case 1: Trigger with lockToVersion but no specific queue
      const result1 = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: {
          payload: { test: "test" },
          options: {
            lockToVersion: worker.worker.version,
          },
        },
      });

      expect(result1).toBeDefined();
      expect(result1?.run.queue).toBe("specific-queue");

      // Test case 2: Trigger with lockToVersion and specific queue
      const result2 = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: {
          payload: { test: "test" },
          options: {
            lockToVersion: worker.worker.version,
            queue: {
              name: "specific-queue",
            },
          },
        },
      });

      expect(result2).toBeDefined();
      expect(result2?.run.queue).toBe("specific-queue");
      expect(result2?.run.lockedQueueId).toBe(specificQueue.id);

      // Test case 3: Try to use non-existent queue with locked version (should throw)
      await expect(
        triggerTaskService.call({
          taskId: taskIdentifier,
          environment: authenticatedEnvironment,
          body: {
            payload: { test: "test" },
            options: {
              lockToVersion: worker.worker.version,
              queue: {
                name: "non-existent-queue",
              },
            },
          },
        })
      ).rejects.toThrow(
        `Specified queue 'non-existent-queue' not found or not associated with locked version '${worker.worker.version}'`
      );

      // Test case 4: Trigger with a non-existent queue without a locked version
      const result4 = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: {
          payload: { test: "test" },
          options: {
            queue: {
              name: "non-existent-queue",
            },
          },
        },
      });

      expect(result4).toBeDefined();
      expect(result4?.run.queue).toBe("non-existent-queue");
      expect(result4?.run.status).toBe("PENDING");

      await engine.quit();
    }
  );

  containerTest(
    "should fall back to the writer when a stale replica returns no row for a locked task",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
        },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "test-task";

      const worker = await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      // A read replica that has not yet caught up to the BackgroundWorkerTask
      // row: it is the real database for every query except the locked-task
      // lookup, which comes back empty (the TRI-10868 false-negative window).
      const staleReplica = new Proxy(prisma, {
        get(target, prop, receiver) {
          if (prop === "backgroundWorkerTask") {
            const delegate = Reflect.get(target, prop, receiver);
            return new Proxy(delegate, {
              get(taskTarget, taskProp, taskReceiver) {
                if (taskProp === "findFirst") {
                  return async () => null;
                }
                const value = Reflect.get(taskTarget, taskProp, taskReceiver);
                return typeof value === "function" ? value.bind(taskTarget) : value;
              },
            });
          }
          const value = Reflect.get(target, prop, receiver);
          return typeof value === "function" ? value.bind(target) : value;
        },
      }) as typeof prisma;

      // Noop cache so every resolve misses the cache and exercises the
      // replica -> writer fallback. The writer is the real `prisma`.
      const queuesManager = new DefaultQueueManager(
        prisma,
        engine,
        staleReplica,
        new NoopTaskMetadataCache()
      );

      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: queuesManager,
        idempotencyKeyConcern: new IdempotencyKeyConcern(
          prisma,
          engine,
          new MockTraceEventConcern()
        ),
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024 * 1,
      });

      // The task IS registered on the locked worker, but the replica returns
      // nothing. Before the fix this threw "not found on locked version"; now
      // the writer fallback resolves the registered row.
      const result = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: {
          payload: { test: "test" },
          options: {
            lockToVersion: worker.worker.version,
          },
        },
      });

      expect(result).toBeDefined();
      expect(result?.run.status).toBe("PENDING");
      expect(result?.run.queue).toBe(`task/${taskIdentifier}`);

      // A genuinely unregistered task must still throw, even with the writer
      // fallback — the writer has no row either, so the 422 is correct.
      await expect(
        triggerTaskService.call({
          taskId: "not-a-registered-task",
          environment: authenticatedEnvironment,
          body: {
            payload: { test: "test" },
            options: {
              lockToVersion: worker.worker.version,
            },
          },
        })
      ).rejects.toThrow(
        `Task 'not-a-registered-task' not found on locked version '${worker.worker.version}'`
      );

      await engine.quit();
    }
  );

  containerTest(
    "should preserve runFriendlyId across retries when RunDuplicateIdempotencyKeyError is thrown",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
        },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
        logLevel: "debug",
      });

      const parentTask = "parent-task";
      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "test-task";

      // Create background worker
      await setupBackgroundWorker(engine, authenticatedEnvironment, [parentTask, taskIdentifier]);

      // Create parent runs and start their attempts (required for resumeParentOnCompletion)
      const parentRun1 = await engine.trigger(
        {
          number: 1,
          friendlyId: "run_p1",
          environment: authenticatedEnvironment,
          taskIdentifier: parentTask,
          payload: "{}",
          payloadType: "application/json",
          context: {},
          traceContext: {},
          traceId: "t12345",
          spanId: "s12345",
          queue: `task/${parentTask}`,
          isTest: false,
          tags: [],
          workerQueue: "main",
        },
        prisma
      );

      await setTimeout(500);
      const dequeued = await engine.dequeueFromWorkerQueue({
        consumerId: "test_12345",
        workerQueue: "main",
      });
      await engine.startRunAttempt({
        runId: parentRun1.id,
        snapshotId: dequeued[0].snapshot.id,
      });

      const parentRun2 = await engine.trigger(
        {
          number: 2,
          friendlyId: "run_p2",
          environment: authenticatedEnvironment,
          taskIdentifier: parentTask,
          payload: "{}",
          payloadType: "application/json",
          context: {},
          traceContext: {},
          traceId: "t12346",
          spanId: "s12346",
          queue: `task/${parentTask}`,
          isTest: false,
          tags: [],
          workerQueue: "main",
        },
        prisma
      );

      await setTimeout(500);
      const dequeued2 = await engine.dequeueFromWorkerQueue({
        consumerId: "test_12345",
        workerQueue: "main",
      });
      await engine.startRunAttempt({
        runId: parentRun2.id,
        snapshotId: dequeued2[0].snapshot.id,
      });

      const queuesManager = new DefaultQueueManager(prisma, engine);
      const idempotencyKeyConcern = new IdempotencyKeyConcern(
        prisma,
        engine,
        new MockTraceEventConcern()
      );

      const triggerRacepointSystem = new MockTriggerRacepointSystem();

      // Track all friendlyIds passed to the payload processor
      const processedFriendlyIds: string[] = [];
      class TrackingPayloadProcessor implements PayloadProcessor {
        async process(request: TriggerTaskRequest): Promise<IOPacket> {
          processedFriendlyIds.push(request.friendlyId);
          return {
            data: JSON.stringify(request.body.payload),
            dataType: "application/json",
          };
        }
      }

      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new TrackingPayloadProcessor(),
        queueConcern: queuesManager,
        idempotencyKeyConcern,
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024 * 1, // 1MB
        triggerRacepointSystem,
      });

      const idempotencyKey = "test-preserve-friendly-id";
      const racepoint = triggerRacepointSystem.registerRacepoint("idempotencyKey", idempotencyKey);

      // Trigger two concurrent requests with same idempotency key
      // One will succeed, one will fail with RunDuplicateIdempotencyKeyError and retry
      const childTriggerPromise1 = triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: {
          payload: { test: "test1" },
          options: {
            idempotencyKey,
            parentRunId: parentRun1.friendlyId,
            resumeParentOnCompletion: true,
          },
        },
      });

      const childTriggerPromise2 = triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: {
          payload: { test: "test2" },
          options: {
            idempotencyKey,
            parentRunId: parentRun2.friendlyId,
            resumeParentOnCompletion: true,
          },
        },
      });

      await setTimeout(500);

      // Resolve the racepoint to allow both requests to proceed
      racepoint.resolve();

      const result1 = await childTriggerPromise1;
      const result2 = await childTriggerPromise2;

      // Both should return the same run (one created, one cached)
      expect(result1).toBeDefined();
      expect(result2).toBeDefined();
      expect(result1?.run.friendlyId).toBe(result2?.run.friendlyId);

      // The key assertion: When a retry happens due to RunDuplicateIdempotencyKeyError,
      // the same friendlyId should be used. We expect exactly 2 calls to payloadProcessor
      // (one for each concurrent request), not 3 (which would indicate a new friendlyId on retry)
      // Since the retry returns early from the idempotency cache, payloadProcessor is not called again.
      expect(processedFriendlyIds.length).toBe(2);

      // Verify that we have exactly 2 unique friendlyIds (one per original request)
      const uniqueFriendlyIds = new Set(processedFriendlyIds);
      expect(uniqueFriendlyIds.size).toBe(2);

      await engine.quit();
    }
  );

  containerTest(
    "should reject invalid debounce.delay when no explicit delay is provided",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
        },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "test-task";

      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      const queuesManager = new DefaultQueueManager(prisma, engine);
      const idempotencyKeyConcern = new IdempotencyKeyConcern(
        prisma,
        engine,
        new MockTraceEventConcern()
      );

      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: queuesManager,
        idempotencyKeyConcern,
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024 * 1,
      });

      // Invalid debounce.delay format (ms not supported)
      await expect(
        triggerTaskService.call({
          taskId: taskIdentifier,
          environment: authenticatedEnvironment,
          body: {
            payload: { test: "test" },
            options: {
              debounce: {
                key: "test-key",
                delay: "300ms", // Invalid - ms not supported
              },
            },
          },
        })
      ).rejects.toThrow("Debounce requires a valid delay duration");

      await engine.quit();
    }
  );

  containerTest(
    "should reject invalid debounce.delay even when explicit delay is valid",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: {
          redis: redisOptions,
          workers: 1,
          tasksPerWorker: 10,
          pollIntervalMs: 100,
        },
        queue: {
          redis: redisOptions,
        },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "test-task";

      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      const queuesManager = new DefaultQueueManager(prisma, engine);
      const idempotencyKeyConcern = new IdempotencyKeyConcern(
        prisma,
        engine,
        new MockTraceEventConcern()
      );

      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: queuesManager,
        idempotencyKeyConcern,
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024 * 1,
      });

      // Valid explicit delay but invalid debounce.delay
      // This is the bug case: the explicit delay passes validation,
      // but debounce.delay would fail later when rescheduling
      await expect(
        triggerTaskService.call({
          taskId: taskIdentifier,
          environment: authenticatedEnvironment,
          body: {
            payload: { test: "test" },
            options: {
              delay: "5m", // Valid explicit delay
              debounce: {
                key: "test-key",
                delay: "invalid-delay", // Invalid debounce delay
              },
            },
          },
        })
      ).rejects.toThrow("Invalid debounce delay");

      await engine.quit();
    }
  );

  containerTest("should accept valid debounce.delay formats", async ({ prisma, redisOptions }) => {
    const engine = new RunEngine({
      prisma,
      worker: {
        redis: redisOptions,
        workers: 1,
        tasksPerWorker: 10,
        pollIntervalMs: 100,
      },
      queue: {
        redis: redisOptions,
      },
      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.0005,
      },
      tracer: trace.getTracer("test", "0.0.0"),
    });

    const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
    const taskIdentifier = "test-task";

    await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

    const queuesManager = new DefaultQueueManager(prisma, engine);
    const idempotencyKeyConcern = new IdempotencyKeyConcern(
      prisma,
      engine,
      new MockTraceEventConcern()
    );

    const triggerTaskService = new RunEngineTriggerTaskService({
      engine,
      prisma,
      payloadProcessor: new MockPayloadProcessor(),
      queueConcern: queuesManager,
      idempotencyKeyConcern,
      validator: new MockTriggerTaskValidator(),
      traceEventConcern: new MockTraceEventConcern(),
      tracer: trace.getTracer("test", "0.0.0"),
      metadataMaximumSize: 1024 * 1024 * 1,
    });

    // Valid debounce.delay format
    const result = await triggerTaskService.call({
      taskId: taskIdentifier,
      environment: authenticatedEnvironment,
      body: {
        payload: { test: "test" },
        options: {
          debounce: {
            key: "test-key",
            delay: "5s", // Valid format
          },
        },
      },
    });

    expect(result).toBeDefined();
    expect(result?.run.friendlyId).toBeDefined();

    await engine.quit();
  });

  // ─── Mollifier integration ──────────────────────────────────────────────────
  //
  // These tests pin the call-site behaviour of the mollifier hooks inside
  // RunEngineTriggerTaskService.call. They use the optional DI ports
  // (`evaluateGate`, `getMollifierBuffer`) added on the service constructor —
  // production wiring is unchanged (defaults to the live module-level imports).
  // Each test's regression intent lives in its own setup comment.

  class CapturingMollifierBuffer {
    public accepted: Array<{ runId: string; envId: string; orgId: string; payload: string }> = [];
    async accept(input: { runId: string; envId: string; orgId: string; payload: string }) {
      this.accepted.push(input);
      return true;
    }
    async pop() {
      return null;
    }
    async ack() {}
    async requeue() {}
    async fail() {
      return false;
    }
    async getEntry() {
      return null;
    }
    async listEnvs(): Promise<string[]> {
      return [];
    }
    async getEntryTtlSeconds(): Promise<number> {
      return -1;
    }
    async evaluateTrip() {
      return { tripped: false, count: 0 };
    }
    async close() {}
  }

  containerTest(
    "mollifier · validation throws before the gate is consulted; no buffer write",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
        queue: { redis: redisOptions },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "test-task";
      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      // Validator that fails on maxAttempts. Any validation throw must abort
      // the call BEFORE the gate runs — otherwise the gate could leak a
      // buffer write for an invalid request.
      class FailingMaxAttemptsValidator extends MockTriggerTaskValidator {
        validateMaxAttempts(): ValidationResult {
          return { ok: false, error: new Error("synthetic max-attempts failure") };
        }
      }

      const buffer = new CapturingMollifierBuffer();
      const evaluateGateSpy = vi.fn(async () => ({
        action: "mollify" as const,
        decision: {
          divert: true as const,
          reason: "per_env_rate" as const,
          count: 99,
          threshold: 1,
          windowMs: 200,
          holdMs: 500,
        },
      }));

      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: new DefaultQueueManager(prisma, engine),
        idempotencyKeyConcern: new IdempotencyKeyConcern(
          prisma,
          engine,
          new MockTraceEventConcern()
        ),
        validator: new FailingMaxAttemptsValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024,
        evaluateGate: evaluateGateSpy,
        getMollifierBuffer: () => buffer as never,
        isMollifierGloballyEnabled: () => true,
      });

      await expect(
        triggerTaskService.call({
          taskId: taskIdentifier,
          environment: authenticatedEnvironment,
          body: { payload: { test: "x" } },
        })
      ).rejects.toThrow(/synthetic max-attempts failure/);

      // Critical: the gate must NEVER be consulted when validation fails.
      // If this assertion fires, validation has been re-ordered after the
      // mollifier gate — a regression that would let invalid triggers land
      // in the buffer.
      expect(evaluateGateSpy).not.toHaveBeenCalled();
      expect(buffer.accepted).toHaveLength(0);

      await engine.quit();
    }
  );

  containerTest(
    "mollifier · mollify action writes to buffer and returns synthetic result (no Postgres row)",
    async ({ prisma, redisOptions }) => {
      // When the gate decides mollify, the call site
      // invokes `mollifyTrigger` which writes the engine.trigger snapshot
      // to the buffer and returns a synthesised `MollifySyntheticResult`
      // (run.friendlyId + notice + isCached:false). `engine.trigger` is
      // NEVER invoked on this path — the run materialises in Postgres
      // later, when the drainer replays the snapshot. The replay is
      // covered by `mollifierDrainerHandler.test.ts`; this test pins the
      // call-site integration: synthetic result + buffer write + no
      // Postgres side effect.
      const engine = new RunEngine({
        prisma,
        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
        queue: { redis: redisOptions },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "test-task";
      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      // Buffer override records the time of the accept call so we can
      // assert that traceRun fired strictly before the buffer was
      // touched. If a future change re-introduces the "skip traceRun on
      // mollify" shortcut, traceConcern.traceRunEnteredAt stays
      // undefined and the ordering assertion fails.
      class TimestampedBuffer extends CapturingMollifierBuffer {
        public acceptedAt: number | undefined;
        override async accept(input: {
          runId: string;
          envId: string;
          orgId: string;
          payload: string;
        }) {
          this.acceptedAt = Date.now();
          return await super.accept(input);
        }
      }
      const buffer = new TimestampedBuffer();
      const trippedDecision = {
        divert: true as const,
        reason: "per_env_rate" as const,
        count: 150,
        threshold: 100,
        windowMs: 200,
        holdMs: 500,
      };
      const traceConcern = new MockTraceEventConcern();

      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: new DefaultQueueManager(prisma, engine),
        idempotencyKeyConcern: new IdempotencyKeyConcern(
          prisma,
          engine,
          new MockTraceEventConcern()
        ),
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: traceConcern,
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024,
        evaluateGate: async () => ({ action: "mollify", decision: trippedDecision }),
        getMollifierBuffer: () => buffer as never,
        isMollifierGloballyEnabled: () => true,
      });

      const result = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: { payload: { hello: "world" } },
      });

      // Pre-modifier span creation: traceRun must run BEFORE the buffer
      // is touched. Customer-visible effect — the run span lands in
      // ClickHouse from the moment the trigger returns, even when the
      // drainer is offline, so buffered runs are visible in the trace
      // view immediately rather than only after drain.
      expect(traceConcern.traceRunEnteredAt).toBeDefined();
      expect(buffer.acceptedAt).toBeDefined();
      expect(traceConcern.traceRunEnteredAt!).toBeLessThanOrEqual(buffer.acceptedAt!);

      // Synthetic result is returned with the `mollifier.queued` notice
      // (the call-site casts the synthetic shape to `TriggerTaskServiceResult`;
      // at runtime the `notice` and `isCached: false` fields are present
      // and read by the api.v1.tasks.$taskId.trigger.ts route handler).
      expect(result).toBeDefined();
      expect(result?.run.friendlyId).toBeDefined();
      const synthetic = result as unknown as {
        run: { friendlyId: string };
        isCached: false;
        notice: { code: string; message: string; docs: string };
      };
      expect(synthetic.isCached).toBe(false);
      expect(synthetic.notice.code).toBe("mollifier.queued");
      expect(synthetic.notice.message).toBeTypeOf("string");
      expect(synthetic.notice.docs).toBeTypeOf("string");

      // The mollify branch must flag `isMollified: true` on the result so
      // the trigger route can skip `saveRequestIdempotency`. Caching the
      // synthetic runId in the request-idempotency table would mean a
      // lost-response SDK retry (same `x-trigger-request-idempotency-key`
      // header) hits a PG miss in `handleRequestIdempotency` and falls
      // through to a fresh trigger — producing a duplicate buffer entry
      // for trigger calls without a task-level idempotency key. The
      // bounded behaviour (accept retry-as-fresh-trigger during the
      // buffer window) is the deliberate choice; a stale-cache lookup
      // returning null is not.
      expect(result?.isMollified).toBe(true);

      // buffer.accept ran — Redis has the canonical engine.trigger snapshot
      // under the synthesised friendlyId. The drainer will read this and
      // replay it through engine.trigger to materialise the run.
      expect(buffer.accepted).toHaveLength(1);
      expect(buffer.accepted[0]!.runId).toBe(result!.run.friendlyId);
      expect(buffer.accepted[0]!.envId).toBe(authenticatedEnvironment.id);
      expect(buffer.accepted[0]!.orgId).toBe(authenticatedEnvironment.organizationId);
      // Payload is a JSON-serialised MollifierSnapshot (the engine.trigger
      // input). Schema is internal to the engine, so we only assert that
      // it parses and references the friendlyId — anything more specific
      // would couple the mollifier-layer test to engine-layer fields.
      const snapshot = JSON.parse(buffer.accepted[0]!.payload) as {
        traceId?: string;
        spanId?: string;
        traceContext?: { traceparent?: string };
      };

      // Regression guard for the dashboard trace-tree bug: the mollifier
      // snapshot MUST carry a W3C `traceparent` in `traceContext`,
      // seeded from the same span traceRun opened. Without it, the
      // drainer replays through engine.trigger with empty traceContext
      // and every downstream `recordRunDebugLog`
      // (QUEUED/EXECUTING/FINISHED/run:notify…) gets a fresh traceId +
      // null parentId — the run-detail page can only show the root
      // span. Both the mollify and pass-through paths now flow through
      // `traceEventConcern.traceRun`; this assertion pins the
      // seeding-from-the-run-span contract.
      expect(snapshot.traceContext?.traceparent).toMatch(
        /^00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}$/
      );
      expect(snapshot.traceContext!.traceparent).toContain(snapshot.traceId);
      expect(snapshot.traceContext!.traceparent).toContain(snapshot.spanId);
      // The snapshot inherits the *run span's* traceId/spanId (from the
      // event handed in by traceRun), not a separately-generated OTel
      // span. This is what lets the drainer's `mollifier.drained` span
      // and downstream engine.trigger materialisation parent on the
      // same ClickHouse trace the customer sees from the moment trigger
      // returns.
      expect(snapshot.traceId).toBe(MOCK_TRACE_ID);
      expect(snapshot.spanId).toBe(MOCK_SPAN_ID);

      // Postgres has NOT been written: engine.trigger was never called on
      // the mollify path. The run materialises only when the drainer
      // replays the snapshot. Regression intent: if a future change makes
      // the mollify branch fall through to engine.trigger (re-introducing
      // phase-1 dual-write), this assertion fails loudly.
      const pgRun = await prisma.taskRun.findFirst({
        where: { friendlyId: result!.run.friendlyId },
      });
      expect(pgRun).toBeNull();

      await engine.quit();
    }
  );

  containerTest(
    "mollifier · pass_through action does NOT call buffer.accept",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
        queue: { redis: redisOptions },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "test-task";
      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      const buffer = new CapturingMollifierBuffer();
      const getBufferSpy = vi.fn(() => buffer as never);

      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: new DefaultQueueManager(prisma, engine),
        idempotencyKeyConcern: new IdempotencyKeyConcern(
          prisma,
          engine,
          new MockTraceEventConcern()
        ),
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024,
        evaluateGate: async () => ({ action: "pass_through" }),
        getMollifierBuffer: getBufferSpy,
        isMollifierGloballyEnabled: () => true,
      });

      const result = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: { payload: { test: "x" } },
      });

      expect(result).toBeDefined();
      // Postgres has the run, no buffer side-effects
      expect(buffer.accepted).toHaveLength(0);
      // getMollifierBuffer must not be called either — the call site short-circuits
      // before touching the singleton when the gate says pass_through.
      expect(getBufferSpy).not.toHaveBeenCalled();
      // Pass-through must NOT set `isMollified` — `result.run` is a real
      // PG row, and the trigger route's `saveRequestIdempotency` is
      // safe to call. Setting the flag here would silently skip the
      // request-idempotency cache for every non-mollified trigger on a
      // mollifier-enabled org, breaking lost-response retry dedup.
      expect(result?.isMollified).toBeFalsy();

      await engine.quit();
    }
  );

  containerTest(
    "mollifier · idempotency-key match short-circuits BEFORE the gate is consulted",
    async ({ prisma, redisOptions }) => {
      // SCENARIO: a trigger arrives with an idempotency key matching an
      // already-created run. `IdempotencyKeyConcern.handleTriggerRequest`
      // (line 236 of triggerTask.server.ts) detects the match BEFORE the
      // mollifier gate runs and returns `{ isCached: true, run }`. The
      // service early-returns. The gate is never consulted, buffer.accept
      // never fires, no orphan entry is created.
      //
      // Regression intent: if IdempotencyKeyConcern were re-ordered to run
      // AFTER evaluateGate, every idempotent retry on a flagged org would
      // produce an orphan buffer entry — the audit-trail invariant ("every
      // buffered runId has a matching TaskRun") would silently start failing
      // for retries. This test pins the current order.

      const engine = new RunEngine({
        prisma,
        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
        queue: { redis: redisOptions },
        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.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "test-task";
      await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);

      const idempotencyKeyConcern = new IdempotencyKeyConcern(
        prisma,
        engine,
        new MockTraceEventConcern()
      );

      // Setup: normal trigger to create the cached run (no mollifier).
      const baseline = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: new DefaultQueueManager(prisma, engine),
        idempotencyKeyConcern,
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024,
      });
      const first = await baseline.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: { payload: { test: "x" }, options: { idempotencyKey: "regression-key-5" } },
      });
      expect(first?.isCached).toBe(false);

      // Action: same idempotency key, with a mollify-stub gate that WOULD
      // create an orphan if reached. The concern must short-circuit first.
      const buffer = new CapturingMollifierBuffer();
      const evaluateGateSpy = vi.fn(async () => ({
        action: "mollify" as const,
        decision: {
          divert: true as const,
          reason: "per_env_rate" as const,
          count: 150,
          threshold: 100,
          windowMs: 200,
          holdMs: 500,
        },
      }));

      const mollifierService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: new DefaultQueueManager(prisma, engine),
        idempotencyKeyConcern,
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024,
        evaluateGate: evaluateGateSpy,
        getMollifierBuffer: () => buffer as never,
        isMollifierGloballyEnabled: () => true,
      });

      const cached = await mollifierService.call({
        taskId: taskIdentifier,
        environment: authenticatedEnvironment,
        body: { payload: { test: "x" }, options: { idempotencyKey: "regression-key-5" } },
      });

      // Customer sees the cached run, isCached=true
      expect(cached).toBeDefined();
      expect(cached?.isCached).toBe(true);
      expect(cached?.run.friendlyId).toBe(first?.run.friendlyId);

      // Critical: the gate must NEVER be consulted on a cached-idempotency replay.
      expect(evaluateGateSpy).not.toHaveBeenCalled();
      expect(buffer.accepted).toHaveLength(0);

      await engine.quit();
    }
  );
});

describe("DefaultQueueManager task metadata cache", () => {
  containerTest(
    "warm cache returns metadata without falling through to PG",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
        queue: { redis: redisOptions },
        runLock: { redis: redisOptions },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": { name: "small-1x", cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
          },
          baseCostInCents: 0.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "cached-task";
      const setup = await setupBackgroundWorker(engine, environment, taskIdentifier);

      const redis = new Redis(redisOptions);
      const cache = new RedisTaskMetadataCache({ redis });

      // Pre-populate cache with AGENT triggerSource; DB row has the default STANDARD.
      // If the read path hits the cache, the resulting TaskRun.taskKind reflects the
      // cached value. If it falls through to PG, it reflects STANDARD.
      await cache.populateByCurrentWorker(environment.id, setup.worker.id, [
        {
          slug: taskIdentifier,
          ttl: null,
          triggerSource: "AGENT",
          queueId: null,
          queueName: `task/${taskIdentifier}`,
        },
      ]);

      const queuesManager = new DefaultQueueManager(prisma, engine, undefined, cache);
      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: queuesManager,
        idempotencyKeyConcern: new IdempotencyKeyConcern(
          prisma,
          engine,
          new MockTraceEventConcern()
        ),
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024,
      });

      const result = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment,
        body: { payload: { test: "x" } },
      });

      assertNonNullable(result);
      expect(result.run.taskIdentifier).toBe(taskIdentifier);
      expect((result.run.annotations as { taskKind?: string } | null)?.taskKind).toBe("AGENT");

      await redis.quit();
      await engine.quit();
    }
  );

  containerTest(
    "cache miss falls through to PG and back-fills the cache",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
        queue: { redis: redisOptions },
        runLock: { redis: redisOptions },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": { name: "small-1x", cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
          },
          baseCostInCents: 0.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "miss-task";
      await setupBackgroundWorker(engine, environment, taskIdentifier);

      const redis = new Redis(redisOptions);
      const cache = new RedisTaskMetadataCache({ redis });

      // Cache starts empty. Sanity-check both keyspaces.
      expect(await cache.getCurrent(environment.id, taskIdentifier)).toBeNull();

      const queuesManager = new DefaultQueueManager(prisma, engine, undefined, cache);
      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: queuesManager,
        idempotencyKeyConcern: new IdempotencyKeyConcern(
          prisma,
          engine,
          new MockTraceEventConcern()
        ),
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024,
      });

      const result = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment,
        body: { payload: { test: "x" } },
      });

      assertNonNullable(result);
      expect((result.run.annotations as { taskKind?: string } | null)?.taskKind).toBe("STANDARD");

      // Back-fill is fire-and-forget; poll with a bounded timeout to avoid CI flakes.
      let backfilled = await cache.getCurrent(environment.id, taskIdentifier);
      for (let i = 0; i < 40 && !backfilled; i++) {
        await setTimeout(25);
        backfilled = await cache.getCurrent(environment.id, taskIdentifier);
      }
      expect(backfilled).not.toBeNull();
      expect(backfilled?.triggerSource).toBe("STANDARD");
      expect(backfilled?.queueName).toBe(`task/${taskIdentifier}`);

      await redis.quit();
      await engine.quit();
    }
  );

  containerTest(
    "queue-override + ttl path returns taskKind from cache without a BWT lookup",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
        queue: { redis: redisOptions },
        runLock: { redis: redisOptions },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": { name: "small-1x", cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
          },
          baseCostInCents: 0.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "override-task";
      const setup = await setupBackgroundWorker(engine, environment, taskIdentifier);

      const redis = new Redis(redisOptions);
      const cache = new RedisTaskMetadataCache({ redis });

      // Cache says AGENT; DB row says STANDARD. Caller provides both a queue
      // override and an explicit TTL — the hot path the PR regressed.
      await cache.populateByCurrentWorker(environment.id, setup.worker.id, [
        {
          slug: taskIdentifier,
          ttl: null,
          triggerSource: "AGENT",
          queueId: null,
          queueName: `task/${taskIdentifier}`,
        },
      ]);

      const queuesManager = new DefaultQueueManager(prisma, engine, undefined, cache);
      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: queuesManager,
        idempotencyKeyConcern: new IdempotencyKeyConcern(
          prisma,
          engine,
          new MockTraceEventConcern()
        ),
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024,
      });

      const result = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment,
        body: {
          payload: { test: "x" },
          options: {
            queue: { name: "caller-queue" },
            ttl: "5m",
          },
        },
      });

      assertNonNullable(result);
      expect(result.run.queue).toBe("caller-queue");
      expect((result.run.annotations as { taskKind?: string } | null)?.taskKind).toBe("AGENT");

      await redis.quit();
      await engine.quit();
    }
  );

  containerTest(
    "locked-version trigger reads from by-worker keyspace, not env keyspace",
    async ({ prisma, redisOptions }) => {
      const engine = new RunEngine({
        prisma,
        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
        queue: { redis: redisOptions },
        runLock: { redis: redisOptions },
        machines: {
          defaultMachine: "small-1x",
          machines: {
            "small-1x": { name: "small-1x", cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
          },
          baseCostInCents: 0.0005,
        },
        tracer: trace.getTracer("test", "0.0.0"),
      });

      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
      const taskIdentifier = "keyspace-task";
      const worker = await setupBackgroundWorker(engine, environment, taskIdentifier);

      const redis = new Redis(redisOptions);
      const cache = new RedisTaskMetadataCache({ redis });

      // Populate the two keyspaces with conflicting triggerSource values so we
      // can tell which keyspace the read used. The real worker's by-worker
      // hash gets AGENT; the env hash gets SCHEDULED (seeded via a throwaway
      // worker id since `populateByCurrentWorker` writes both keyspaces and
      // we want the real worker's by-worker hash untouched).
      await cache.populateByWorker(worker.worker.id, [
        {
          slug: taskIdentifier,
          ttl: null,
          triggerSource: "AGENT",
          queueId: null,
          queueName: `task/${taskIdentifier}`,
        },
      ]);
      await cache.populateByCurrentWorker(environment.id, "dummy-worker-for-env-seed", [
        {
          slug: taskIdentifier,
          ttl: null,
          triggerSource: "SCHEDULED",
          queueId: null,
          queueName: `task/${taskIdentifier}`,
        },
      ]);

      const queuesManager = new DefaultQueueManager(prisma, engine, undefined, cache);
      const triggerTaskService = new RunEngineTriggerTaskService({
        engine,
        prisma,
        payloadProcessor: new MockPayloadProcessor(),
        queueConcern: queuesManager,
        idempotencyKeyConcern: new IdempotencyKeyConcern(
          prisma,
          engine,
          new MockTraceEventConcern()
        ),
        validator: new MockTriggerTaskValidator(),
        traceEventConcern: new MockTraceEventConcern(),
        tracer: trace.getTracer("test", "0.0.0"),
        metadataMaximumSize: 1024 * 1024,
      });

      // Locked → by-worker keyspace → AGENT
      const locked = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment,
        body: {
          payload: { test: "x" },
          options: { lockToVersion: worker.worker.version },
        },
      });
      assertNonNullable(locked);
      expect((locked.run.annotations as { taskKind?: string } | null)?.taskKind).toBe("AGENT");

      // Not locked → env keyspace → SCHEDULED
      const current = await triggerTaskService.call({
        taskId: taskIdentifier,
        environment,
        body: { payload: { test: "y" } },
      });
      assertNonNullable(current);
      expect((current.run.annotations as { taskKind?: string } | null)?.taskKind).toBe("SCHEDULED");

      await redis.quit();
      await engine.quit();
    }
  );
});
