test(webapp): split triggerTask engine test into per-concern files (#4167)

The engine `triggerTask` suite was a single 2447-line file with 23
`containerTest` cases, each spinning its own Postgres + Redis. vitest
shards by whole file, so all 23 container setups landed on one shard and
dominated its wall-clock. The recorded entry in `test-timings.json`
badly under-counts the real cost (it does not capture the
per-`containerTest` container startup that dominates on CI), so the
duration-sharding sequencer treated the file as light and stacked it,
producing one ~21 minute shard.

Splitting does not reduce the number of container setups; it lets those
23 cases distribute across shards instead of stacking on one. The webapp
unit-test stage is gated by its slowest shard, so this cuts the stage's
wall-clock roughly in half.

## CI timing (before vs after)

Real CI wall-clock of the `Unit Tests: Webapp` shards (`--shard=i/10`).
"Before" is sampled from recent runs on other branches (unsplit file,
from `main`); "after" is this PR.

| Shard | Before (s) | After (s) |
|------:|-----------:|----------:|
| 1  | 250 | 359 |
| 2  | 444 | 411 |
| 3  | 497 | 659 |
| 4  | **1257** | 284 |
| 5  | 545 | 641 |
| 6  | 284 | 644 |
| 7  | 244 | 214 |
| 8  | 340 | 445 |
| 9  | 188 | 395 |
| 10 | 234 | 567 |
| **Slowest shard (gates the stage)** | **~1247s (≈21m)** | **659s
(≈11m)** |
| Sum of all shards | 4283 | 4619 |

Before: shard 4 is the long pole at 1237s / 1247s / 1257s across three
sampled runs (the `triggerTask` file plus whatever else the packer put
with it). After: the six pieces spread across shards, the slowest drops
to 659s. The small rise in summed time is the extra per-file container
startup, paid in parallel across shards, so the gating number still
falls by about 10 minutes.

## Change

Split into six per-concern files that share a `triggerTaskTestHelpers`
module (the `vi.mock` calls stay per-file, since vitest hoists them):

- `triggerTask.test.ts` (3): trigger + concurrencyKey coercion
- `triggerTask.idempotency.test.ts` (4): idempotency + queue resolution
- `triggerTask.debounce.test.ts` (4): retries + debounce validation
- `triggerTask.mollifier.test.ts` (4): mollifier call-site behaviour
- `triggerTask.metadataCache.test.ts` (4): DefaultQueueManager task
metadata cache
- `triggerTask.residency.test.ts` (4): child run residency inheritance

All 23 cases are preserved. The file's `test-timings.json` entry is
split across the new files so bin-packing stays balanced.

While rewriting these files, cleanup was moved to `onTestFinished(() =>
engine.quit())` so an `engine`/`Redis` leaked on a failing assertion no
longer persists on the worker-scoped Redis and cascades into later cases
(`hookTimeout` raised to 60s so the after-cleanup gets the full budget).
Prisma lookups switched from `findUnique` to `findFirst` to match the
repo convention.

Verified: all six files run green locally (23/23), oxlint and oxfmt
clean.

Author: d-cs

apps/webapp/test/engine/triggerTask.debounce.test.ts

@@ -0,0 +1,342 @@
+import { describe, expect, onTestFinished, vi } from "vitest";
+
+// db.server + splitMode are mocked so the idempotency dedup client resolves to
+// the container prisma passed into the concern (split stays off).
+vi.mock("~/db.server", () => ({
+  prisma: {},
+  $replica: {},
+  runOpsNewPrisma: {},
+  runOpsLegacyPrisma: {},
+}));
+
+vi.mock("~/v3/runOpsMigration/splitMode.server", () => ({ isSplitEnabled: async () => false }));
+
+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 { Redis } from "ioredis";
+import { IdempotencyKeyConcern } from "~/runEngine/concerns/idempotencyKeys.server";
+import { DefaultQueueManager } from "~/runEngine/concerns/queues.server";
+import { RedisTaskMetadataCache } from "~/services/taskMetadataCache.server";
+import { RunEngineTriggerTaskService } from "../../app/runEngine/services/triggerTask.server";
+import { setTimeout } from "node:timers/promises";
+import {
+  MockPayloadProcessor,
+  MockTraceEventConcern,
+  MockTriggerTaskValidator,
+} from "./triggerTaskTestHelpers";
+
+vi.setConfig({ testTimeout: 60_000, hookTimeout: 60_000 });
+
+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"),
+      });
+      onTestFinished(() => engine.quit());
+
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const taskIdentifier = "cached-task";
+      const setup = await setupBackgroundWorker(engine, environment, taskIdentifier);
+
+      const redis = new Redis(redisOptions);
+      onTestFinished(() => redis.quit());
+      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");
+    }
+  );
+
+  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"),
+      });
+      onTestFinished(() => engine.quit());
+
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const taskIdentifier = "miss-task";
+      await setupBackgroundWorker(engine, environment, taskIdentifier);
+
+      const redis = new Redis(redisOptions);
+      onTestFinished(() => redis.quit());
+      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}`);
+    }
+  );
+
+  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"),
+      });
+      onTestFinished(() => engine.quit());
+
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const taskIdentifier = "override-task";
+      const setup = await setupBackgroundWorker(engine, environment, taskIdentifier);
+
+      const redis = new Redis(redisOptions);
+      onTestFinished(() => redis.quit());
+      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");
+    }
+  );
+
+  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"),
+      });
+      onTestFinished(() => engine.quit());
+
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const taskIdentifier = "keyspace-task";
+      const worker = await setupBackgroundWorker(engine, environment, taskIdentifier);
+
+      const redis = new Redis(redisOptions);
+      onTestFinished(() => redis.quit());
+      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");
+    }
+  );
+});